Fix for API Data Retrieval

Performance Optimization - History Logging was spawning a new tokio runtime every function call, it now uses the global run time
Documentation - Added a lot more span events for better logging

TUI/Widgets/multiagentselector.py

@@ -14,6 +14,7 @@
 #    along with this program.  If not, see <https://www.gnu.org/licenses/>.
 
 import difflib
+from pathlib import Path
 import re
 from typing import List, Optional
 
@@ -57,7 +58,7 @@ class MultiAgentSelector(Widget):
 
     def compose(self):
         yield Header(show_clock=True, icon="⚙")
-        title_text = Static("🖥️ Agent Selector", id="selector_title")
+        title_text = Static("🖥️ Agent Selector", id="selector_title")
         title_text.styles.margin = (0, 0, 0, 1)
         yield title_text
 
@@ -77,7 +78,7 @@ class MultiAgentSelector(Widget):
                 text_area.styles.overflow_y = "auto"
                 yield text_area
 
-                with Horizontal(id="switch_search_container"):
+                with Horizontal(id="switch_container"):
                     switch = Switch(value=False, id="match_switch")
                     switch.styles.width = "auto"
                     switch.styles.margin = (1, 0, 0, 0)
@@ -89,8 +90,13 @@ class MultiAgentSelector(Widget):
                     switch_label.styles.margin = (2, 1, 0, 0)
                     yield switch_label
 
+                with Horizontal(id="action_buttons_container"):
+                    load_file = Button("📂 Load File", id="load_file_button")
+                    load_file.styles.margin = (1, 1, 0, 1)
+                    yield load_file
+
                     search = Button("🔍 Search", id="search_button")
-                    search.styles.margin = (1, 0, 0, 0)
+                    search.styles.margin = (1, 0, 0, 1)
                     yield search
 
                 with Horizontal() as select_buttons:
@@ -152,6 +158,9 @@ class MultiAgentSelector(Widget):
             ]
             self.post_message(self.AgentsSelected(selected_agents))
             event.stop()
+        elif btn_id == "load_file_button":
+            self._load_from_file()
+            event.stop()
         elif btn_id == "search_button":
             self.update_matches()
             event.stop()
@@ -166,7 +175,7 @@ class MultiAgentSelector(Widget):
             match_list.add_option((name, name))
         unmatched_label = self.query_one("#unmatched_label", Static)
         if unmatched:
-            unmatched_label.update(f"⚠️ No matches for: {', '.join(unmatched)}")
+            unmatched_label.update(f"⚠️ No matches for: {', '.join(unmatched)}")
         else:
             unmatched_label.update("")
 
@@ -214,3 +223,118 @@ class MultiAgentSelector(Widget):
                 else:
                     unmatched.append(name)
         return sorted(matched), unmatched
+
+    def _load_from_file(self):
+        """Safely load device names from a text file."""
+        try:
+            # Import here to avoid issues if tkinter isn't available
+            import tkinter as tk
+            from tkinter import filedialog
+
+            # Create file dialog
+            root = tk.Tk()
+            root.withdraw()
+
+            file_path = filedialog.askopenfilename(
+                title="Select device list file",
+                filetypes=[
+                    ("Text files", "*.txt"),
+                    ("CSV files", "*.csv"),
+                    ("All files", "*.*"),
+                ],
+            )
+
+            if not file_path:
+                # User cancelled
+                return
+
+            # Validate file path
+            path_obj = Path(file_path)
+            if not path_obj.exists():
+                self.app.notify("File does not exist", severity="error", timeout=3)
+                return
+
+            if not path_obj.is_file():
+                self.app.notify(
+                    "Selected path is not a file", severity="error", timeout=3
+                )
+                return
+
+            # Check file size (limit to 1 MB for safety)
+            file_size = path_obj.stat().st_size
+            if file_size > 1_000_000:  # 1 MB
+                self.app.notify(
+                    f"File too large ({file_size:,} bytes). Maximum 1 MB.",
+                    severity="error",
+                    timeout=5,
+                )
+                return
+
+            # Read file with proper encoding to preserve emojis
+            try:
+                with open(file_path, "r", encoding="utf-8") as f:
+                    content = f.read()
+            except UnicodeDecodeError:
+                # Try with different encoding if UTF-8 fails
+                try:
+                    with open(file_path, "r", encoding="latin-1") as f:
+                        content = f.read()
+                    self.app.notify(
+                        "File loaded with Latin-1 encoding (UTF-8 failed)",
+                        severity="warning",
+                        timeout=3,
+                    )
+                except Exception as e:
+                    self.app.notify(
+                        f"Error reading file: {str(e)}", severity="error", timeout=5
+                    )
+                    return
+
+            # Validate and sanitize content
+            lines = content.split("\n")
+            valid_lines = []
+            invalid_count = 0
+
+            # Pattern for valid hostnames/device names
+            # Allows: letters, numbers, hyphens, underscores, periods, and Unicode chars
+            hostname_pattern = re.compile(r"^[\w\-\.\u0080-\uFFFF]+$", re.UNICODE)
+
+            for line in lines:
+                line = line.strip()
+                if not line:
+                    continue  # Skip empty lines
+
+                # Check if line looks like a valid hostname/device name
+                if hostname_pattern.match(line):
+                    valid_lines.append(line)
+                else:
+                    invalid_count += 1
+                    # Log but don't add invalid entries
+
+            if not valid_lines:
+                self.app.notify(
+                    "No valid device names found in file", severity="warning", timeout=3
+                )
+                return
+
+            # Update text area with validated content
+            text_area = self.query_one("#device_input", TextArea)
+            text_area.text = "\n".join(valid_lines)
+
+            # Show notification
+            msg = f"✅ Loaded {len(valid_lines)} devices from file"
+            if invalid_count > 0:
+                msg += f" ({invalid_count} invalid entries skipped)"
+
+            self.app.notify(msg, severity="information", timeout=5)
+
+        except ImportError:
+            self.app.notify(
+                "tkinter not available - cannot open file dialog",
+                severity="error",
+                timeout=3,
+            )
+        except Exception as e:
+            self.app.notify(
+                f"Error loading file: {str(e)}", severity="error", timeout=5
+            )

TUI/Widgets/serverlogwidget.py

@@ -19,7 +19,7 @@ import logging
 from bson import ObjectId
 from textual.app import ComposeResult
 from textual.containers import Container, Vertical
-from textual.widgets import Button, DataTable, Static
+from textual.widgets import Button, DataTable, Input, Static
 
 from services.API import AirlockAPIWrapper
 
@@ -67,6 +67,19 @@ class ServerLogWidget(Vertical):
         height: auto;
         layout: horizontal;
         padding: 1;
+        align: left middle;
+    }
+
+    ServerLogWidget .filter_label {
+        width: auto;
+        height: 3;
+        content-align: left middle;
+        padding-right: 1;
+    }
+
+    ServerLogWidget #filter_input {
+        width: 40;
+        margin-right: 1;
     }
 
     ServerLogWidget Button {
@@ -77,11 +90,15 @@ class ServerLogWidget(Vertical):
     def __init__(self, api: AirlockAPIWrapper):
         super().__init__()
         self.api = api
+        self.all_logs = []  # Store all logs for filtering
+        self.columns = []  # Store column names
 
     def compose(self) -> ComposeResult:
         yield Static("Loading server logs (last 72 hours)...", id="status_bar")
         yield DataTable(id="server_log_table")
         with Container(id="button_container"):
+            yield Static("Filter:", classes="filter_label")
+            yield Input(placeholder="Filter (use * and ? wildcards)", id="filter_input")
             yield Button("Refresh", id="refresh_button", variant="primary")
 
     def on_mount(self) -> None:
@@ -105,23 +122,28 @@ class ServerLogWidget(Vertical):
             if not logs:
                 status.update("ℹï¸ No server logs found in the last 72 hours.")
                 table.clear(columns=True)
+                self.all_logs = []
+                self.columns = []
                 return
 
+            # Store all logs for filtering
+            self.all_logs = logs
+
             # Clear existing data
             table.clear(columns=True)
 
             # Add columns based on the first log entry
             if logs:
                 first_log = logs[0]
-                columns = [col for col in first_log.keys() if col != "checkpoint"]
+                self.columns = [col for col in first_log.keys() if col != "checkpoint"]
 
-                for col in columns:
+                for col in self.columns:
                     table.add_column(col, key=col)
 
                 # Add rows in reverse order so newest entries are at the top
                 for log_entry in reversed(logs):
                     row_data = []
-                    for col in columns:
+                    for col in self.columns:
                         value = log_entry.get(col, "")
                         # Format datetime column to be more readable
                         if col == "datetime" and value:
@@ -143,12 +165,86 @@ class ServerLogWidget(Vertical):
                 logger.info(f"Loaded {len(logs)} server log entries")
             else:
                 status.update("ℹï¸ No log entries found.")
+                self.all_logs = []
+                self.columns = []
 
         except Exception as exc:
             error_msg = f"❌ Error loading server logs: {exc}"
             status.update(error_msg)
             logger.error(f"Failed to load server logs: {exc}", exc_info=True)
             table.clear(columns=True)
+            self.all_logs = []
+            self.columns = []
+
+    def filter_logs(self, filter_text: str) -> None:
+        """Filter the logs based on the filter text with wildcard support."""
+        import fnmatch
+
+        table = self.query_one("#server_log_table", DataTable)
+        status = self.query_one("#status_bar", Static)
+
+        if not self.all_logs:
+            return
+
+        # Clear existing data
+        table.clear(columns=True)
+
+        # Re-add columns
+        for col in self.columns:
+            table.add_column(col, key=col)
+
+        # Filter logs
+        filtered_logs = []
+        if filter_text.strip():
+            filter_pattern = filter_text.strip().lower()
+            for log_entry in self.all_logs:
+                # Check if any field matches the filter pattern
+                match = False
+                for col in self.columns:
+                    value = str(log_entry.get(col, "")).lower()
+                    if fnmatch.fnmatch(value, filter_pattern):
+                        match = True
+                        break
+                if match:
+                    filtered_logs.append(log_entry)
+        else:
+            # No filter, show all logs
+            filtered_logs = self.all_logs
+
+        # Add filtered rows in reverse order
+        for log_entry in reversed(filtered_logs):
+            row_data = []
+            for col in self.columns:
+                value = log_entry.get(col, "")
+                # Format datetime column to be more readable
+                if col == "datetime" and value:
+                    try:
+                        dt = datetime.datetime.fromisoformat(
+                            str(value).replace("Z", "+00:00")
+                        )
+                        value = dt.strftime("%Y-%m-%d %H:%M:%S")
+                    except Exception:
+                        pass
+                row_data.append(str(value))
+            table.add_row(*row_data)
+
+        if filter_text.strip():
+            status.update(
+                f"✅ Showing {len(filtered_logs)} of {len(self.all_logs)} log entries (filtered)"
+            )
+        else:
+            status.update(
+                f"✅ Loaded {len(self.all_logs)} log entries from the last 72 hours"
+            )
+
+        logger.info(
+            f"Filtered to {len(filtered_logs)} entries with pattern: {filter_text}"
+        )
+
+    def on_input_changed(self, event: Input.Changed) -> None:
+        """Handle filter input changes."""
+        if event.input.id == "filter_input":
+            self.filter_logs(event.value)
 
     def on_button_pressed(self, event: Button.Pressed) -> None:
         """Handle button presses."""
@@ -156,4 +252,10 @@ class ServerLogWidget(Vertical):
 
         if button_id == "refresh_button":
             self.load_logs()
+            # Clear the filter input when refreshing
+            try:
+                filter_input = self.query_one("#filter_input", Input)
+                filter_input.value = ""
+            except Exception:
+                pass
             event.stop()

airlock_libs/Cargo.lock

@@ -26,7 +26,7 @@ dependencies = [
 
 [[package]]
 name = "airlock_libs"
-version = "6.1.1"
+version = "7.3.0"
 dependencies = [
  "chrono",
  "crossbeam",

airlock_libs/Cargo.toml

@@ -1,6 +1,6 @@
 [package]
 name = "airlock_libs"
-version = "6.1.1"
+version = "7.3.0"
 edition = "2024"
 
 [dependencies]

airlock_libs/pyproject.toml

@@ -4,7 +4,7 @@ build-backend = "maturin"
 
 [project]
 name = "airlock_libs"
-version = "6.1.1"
+version = "7.3.0"
 description = "Airlock Digital API Wrapper"
 readme = "README.md"
 license = { text = "AGPL-3.0-only" }

airlock_libs/src/services.rs

@@ -5,7 +5,7 @@ use crate::prelude::*;
 pub fn pull_policy_exec_histories(
     py: Python<'_>,
     py_self: Py<PyAny>,
-    policy_names: String,
+    policy_names: Option<String>,
     exec_types: String,
     days: i64,
 ) -> Py<PyString> {
@@ -73,8 +73,8 @@ pub fn pull_policy_exec_histories(
             .set_draw_target(ProgressDrawTarget::stderr());
         progress_bar.lock().unwrap().set_style(
             ProgressStyle::default_bar()
-                .template("Total Completion: {spinner:.green} [{elapsed_precise}] [{bar:40.green/blue}] {pos}/{len} {message}")
-                .unwrap(),
+                .template("Total - Policy Name: {msg}: {spinner:.green} [{elapsed_precise}] [{bar:40.green/blue}] {pos}/{len}")
+                .unwrap().progress_chars("⣿⣦⣀")
         );
         let client: Client = tracer.in_span("Building HTTP Client", |cx| {
             let client_result: Result<Client, reqwest::Error> = build_client(headers);
@@ -106,7 +106,7 @@ pub fn pull_policy_exec_histories(
         });
         let cutoff: chrono::NaiveDateTime =
             Local::now().naive_local() - chrono::Duration::days(days);
-        let (tx, rx) = unbounded::<Vec<Group>>();
+        let (tx, rx) = unbounded::<(Context, Vec<Group>)>();
         let pb_clone = progress_bar.clone();
         thread::spawn(move || {
             let mut seen: HashMap<(String, String, String), Group> = if writeable_filepath.exists()
@@ -137,7 +137,14 @@ pub fn pull_policy_exec_histories(
             } else {
                 HashMap::new()
             };
-            while let Ok(parsed_responses) = rx.recv() {
+            while let Ok((cx, parsed_responses)) = rx.recv() {
+                cx.span().add_event(
+                    "Received Data from Producer",
+                    vec![KeyValue::new(
+                        "Items to Process",
+                        parsed_responses.len().to_string(),
+                    )],
+                );
                 for executions in parsed_responses {
                     if executions.checkpoint.is_empty() || executions.datetime.is_empty() {
                         continue;
@@ -166,11 +173,28 @@ pub fn pull_policy_exec_histories(
                 };
                 let data_write: String = serde_json::to_string_pretty(&final_response).unwrap();
                 match fs::write(&writeable_filepath, data_write) {
-                    Ok(_) => {}
+                    Ok(_) => {
+                        cx.span().add_event(
+                            "Writing Data to File",
+                            vec![KeyValue::new("Success", "Ok".to_string())],
+                        );
+                    }
                     Err(e) => {
-                        println!("Failed to write to: {:?}: {}", &writeable_filepath, e);
+                        cx.span().add_event(
+                            "Writing Data to File",
+                            vec![KeyValue::new("Failed", e.to_string())],
+                        );
+                        cx.span()
+                            .set_status(Status::error("Failed to Write to File"));
                     }
                 }
+                cx.span().add_event(
+                    "Finished Deduplicating Data",
+                    vec![KeyValue::new(
+                        "Items Successfully Processed",
+                        seen.len().to_string(),
+                    )],
+                );
             }
         });
         let mut first_date: Option<NaiveDate> = None;
@@ -179,18 +203,42 @@ pub fn pull_policy_exec_histories(
                 .lock()
                 .unwrap()
                 .enable_steady_tick(std::time::Duration::from_millis(100));
+            pb_clone
+                .lock()
+                .unwrap()
+                .set_message(policy_names.clone().unwrap_or("Statistics".to_string()));
             let span: opentelemetry::trace::SpanRef<'_> = cx.span();
             span.set_attribute(KeyValue::new("Days", days.to_string()));
-            span.set_attribute(KeyValue::new("Policy Name", policy_names.clone()));
+            span.set_attribute(KeyValue::new(
+                "Policy Name",
+                policy_names
+                    .clone()
+                    .unwrap_or("Statistics Monitoring".to_string()),
+            ));
             loop {
                 let execution_histories = tracer.in_span(checkpoint_number.to_string(), |cx| {
-                    let results: ApiResponse = history_logging(
+                    cx.span().add_event(
+                        "Retrieving Responses from API",
+                        vec![KeyValue::new(
+                            "Checkpoint Number",
+                            checkpoint_number.to_string(),
+                        )],
+                    );
+                    let results: ApiResponse = rt.block_on(history_logging(
                         &base_url,
                         &exec_types,
                         &checkpoint_number,
                         &policy_names,
                         &client,
+                    ));
+                    cx.span().add_event(
+                        "Got Responses from API",
+                        vec![KeyValue::new(
+                            "Items in Response",
+                            results.response.exechistories.len().to_string(),
+                        )],
                     );
+                    cx.span().set_status(Status::Ok);
                     cx.span().set_attribute(KeyValue::new(
                         "items_in_response",
                         results.response.exechistories.len().to_string(),
@@ -201,7 +249,16 @@ pub fn pull_policy_exec_histories(
                 if parsed_responses.is_empty() {
                     break;
                 }
-                tx.send(parsed_responses.clone()).unwrap();
+                match tx.send((cx.clone(), parsed_responses.clone())) {
+                    Ok(_) => {}
+                    Err(e) => {
+                        cx.span().add_event(
+                            "Failed to Send Items to Processor",
+                            vec![KeyValue::new("Response from Processor", e.to_string())],
+                        );
+                        cx.span().set_status(Status::error("Processor Failed"))
+                    }
+                }
                 checkpoint_number = parsed_responses.last().unwrap().checkpoint.clone();
                 if let Some(last_item) = parsed_responses.last()
                     && let Ok(last_date) = NaiveDate::parse_from_str(
@@ -254,21 +311,25 @@ fn build_client(headers: HeaderMap) -> Result<reqwest::Client, reqwest::Error> {
         .build()
 }
 
-#[tokio::main]
+#[tracing::instrument(name = "history_logging")]
 async fn history_logging(
     base_url: &String,
     exec_types: &String,
     checkpoint_number: &String,
-    policy_names: &String,
+    policy_names: &Option<String>,
     client: &Client,
 ) -> ApiResponse {
+    let policy_json = match policy_names {
+        Some(name) => format!(r#"[ "{}" ]"#, name), // JSON array with one element
+        None => "[]".to_string(),                   // Empty JSON array
+    };
     let payload = format!(
         r#"{{
         "type": {},
         "checkpoint": "{}",
-        "policy": ["{}"]
+        "policy": {}
     }}"#,
-        exec_types, checkpoint_number, policy_names
+        exec_types, checkpoint_number, policy_json
     );
     let res: Result<reqwest::Response, reqwest::Error> = client
         .post(format!("{}/v1/logging/exechistories", base_url))
@@ -279,7 +340,7 @@ async fn history_logging(
         Ok(res) => {
             let first_response: ApiResponse = serde_json::from_str(&res.text().await.unwrap())
                 .expect("Failed to retrieve response from API");
-            return first_response;
+            first_response
         }
         Err(_res) => {
             let failed_response: ApiResponse = ApiResponse {
@@ -288,7 +349,7 @@ async fn history_logging(
                     exechistories: vec![],
                 },
             };
-            return failed_response;
+            failed_response
         }
     }
 }

requirements.txt

@@ -23,4 +23,4 @@ pyperclip==1.11.0
 
 # Custom/Private packages
 --extra-index-url https://git.racooncity.org/api/packages/brotoskyj/pypi/simple/
-airlock_libs==6.1.1
+airlock_libs==7.3.0