# Copyright (C) 2025 James Brotosky, Brandon Wickline # # This program is free software: you can redistribute it and/or modify # it under the terms of the GNU Affero General Public License as published # by the Free Software Foundation, either version 3 of the License, or # (at your option) any later version. # # This program is distributed in the hope that it will be useful, # but WITHOUT ANY WARRANTY; without even the implied warranty of # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the # GNU Affero General Public License for more details. # # You should have received a copy of the GNU Affero General Public License # along with this program. If not, see . from __future__ import annotations from datetime import datetime import logging import os import pandas as pd from textual.app import ComposeResult from textual.binding import Binding from textual.containers import Horizontal, Vertical from textual.screen import Screen from textual.widgets import Button, DataTable, Footer, Header, Static from TUI.Screens.allowlistselectionscreen import AllowlistSelectionScreen from utils.configmanager import load_env logger = logging.getLogger(__name__) def _load_working_dir() -> str: """ Load the working directory from environment variables or use the current working directory. """ wd = os.environ.get("WORKING_DIR") if wd: return wd return os.getcwd() class OTPActivitiesWidget(Static): """ Reusable widget that contains the sessions table (left) and an Activity Preview (right). The right side shows an Activity Preview that takes ~75% vertical space, and a lower area with Continue button. """ DEFAULT_CSS = """ OTPActivitiesWidget { height: 1fr; } #main_row { width: 100%; height: 100%; layout: horizontal; } #left_panel { width: 60%; min-width: 60; border: none; } #right_panel { width: 40%; min-width: 40; border: none; layout: vertical; } #activity_preview_container { height: 1fr; border: none; padding: 1 1; } #activity_buttons { height: auto; min-height: 3; padding: 1 1; content-align: center middle; } """ def compose(self) -> ComposeResult: # Layout: horizontal main row with left & right panels with Horizontal(id="main_row"): # Left: sessions area with Vertical(id="left_panel"): yield Static("OTP Sessions", classes="panel-title") with Vertical(id="sessions_table_container"): self.sessions_table = DataTable(id="sessions_table") self.sessions_table.styles.width = "100%" yield self.sessions_table # Right: Activity Preview (top 3/4) + buttons (bottom 1/4) with Vertical(id="right_panel"): # Activity preview area (takes ~75% of right panel) yield Static("Activity Preview", classes="panel-title") with Vertical(id="activity_preview_container"): self.activities_table = DataTable(id="activity_preview_table") yield self.activities_table # Button area at the bottom (Continue) with Horizontal(id="activity_buttons"): self.continue_btn = Button("Continue", id="activity_continue_btn") self.continue_btn.styles.width = "100%" yield self.continue_btn async def on_mount(self) -> None: # Configure sessions table and activities preview self.sessions_table.clear() self.sessions_table.add_columns( "otpid", "hostname", "status", "purpose", "granted" ) self.activities_table.clear() # activities_table columns are dynamically added when activities are loaded. # Selection behavior self.sessions_table.cursor_type = "row" try: self.sessions_table.zebra_stripes = True except Exception: pass self.activities_table.cursor_type = "row" try: self.activities_table.zebra_stripes = True except Exception: pass # Store state self._sessions_df: pd.DataFrame | None = None self._activities_df: pd.DataFrame | None = None self._selected_session_otpid: str | int | None = None async def on_button_pressed(self, event) -> None: # type: ignore[override] """ Handle Continue button for the Activity Preview area. """ # Try to resolve the button object from the event btn = ( getattr(event, "button", None) or getattr(event, "sender", None) or getattr(event, "control", None) or getattr(event, "widget", None) ) btn_id = ( getattr(btn, "id", None) or getattr(event, "button_id", None) or getattr(event, "id", None) ) # ---- Continue ---- if btn is self.continue_btn or btn_id == getattr(self.continue_btn, "id", None): if self._activities_df is None or self._activities_df.empty: logger.info("Continue pressed but no activities loaded.") self.app.notify( "No activities loaded to continue with.", severity="warning" ) return # Copy activities DataFrame to pass to new screen activities_copy = self._activities_df.copy() otpid = self._selected_session_otpid # Optionally include hostname if available hostname = None try: if self._sessions_df is not None: df = self._sessions_df.reset_index(drop=True) match = df[df["otpid"] == otpid] if not match.empty: hostname = match.iloc[0].get("hostname") except Exception: hostname = None # Create and push ActivityDetailScreen, handing the data try: detail_screen = ActivityDetailScreen( activities_copy, otpid=otpid, hostname=hostname ) await self.app.push_screen(detail_screen) except Exception as exc: logger.exception("Failed to push ActivityDetailScreen: %s", exc) return # Unknown button on widget logger.debug( "Unhandled OTPActivitiesWidget button pressed (resolved btn=%r, id=%r)", btn, btn_id, ) async def on_data_table_row_selected(self, event) -> None: # type: ignore[override] """ Robust handler for DataTable row-selection across Textual micro-versions. Tries many attribute names and shapes: - numeric index (row_key, row_index, index) - coordinate object or tuple (coordinate.row or (row, col)) - direct row values (row, values, cells) -> we try to map those back to the sessions DF - table.cursor_row fallback """ # 1) Determine the sending table (best-effort) sender = None for attr in ("sender", "table", "data_table", "control"): sender = getattr(event, attr, None) if sender is not None: break if sender is None: sender = self.sessions_table # Assume sessions_table if unknown # Only respond to selections in the sessions table if sender is not self.sessions_table: return # Helper to log and return def _bad(msg: str, *args): logger.warning(msg, *args) return None # 2) Try to extract a numeric index row_key = None for attr in ("row_key", "row", "row_index", "index"): row_key = getattr(event, attr, None) if row_key is not None: break # If coordinate: try to extract .row or tuple[0] if row_key is None: coord = getattr(event, "coordinate", None) or getattr( event, "cursor_coordinate", None ) if coord is not None: if hasattr(coord, "row"): row_key = coord.row elif isinstance(coord, (tuple, list)) and len(coord) >= 1: row_key = coord[0] # If still nothing, maybe the event provides the row's cell values directly row_values = None for attr in ("values", "cells", "row", "row_values", "selected_row_values"): val = getattr(event, attr, None) if val: # Prefer actual sequence of cell values row_values = val break # If we have row_values, try to map them back to the sessions DataFrame if row_values is not None: # Normalize into list of strings for comparison try: vals = [ "" if pd.isna(v) else str(v) for v in ( list(row_values) if not isinstance(row_values, str) else [row_values] ) ] except Exception: vals = [str(row_values)] # Try to match against the expected columns order we render if self._sessions_df is None or self._sessions_df.empty: logger.warning( "Sessions DataFrame is empty; cannot map selected row values." ) return df_ordered = self._sessions_df.reset_index(drop=True) expected_cols = ["otpid", "hostname", "status", "purpose", "granted"] # Build stringified candidates for each row in df using the same columns we show def _row_to_vals(sr): out = [] for c in expected_cols: if c in sr: v = sr[c] out.append("" if pd.isna(v) else str(v)) else: out.append("") return out match_idx = None for i, sr in df_ordered.iterrows(): cand = _row_to_vals(sr) # Compare prefix: row values might be a subset (e.g. only first 3 cols), so compare prefix only if len(vals) <= len(cand) and all( vals[j] == cand[j] for j in range(len(vals)) ): match_idx = i break if match_idx is None: # Try looser match: compare first cell only (otpid) first = vals[0] if vals else None if first is not None: for i, sr in df_ordered.iterrows(): cand0 = "" if pd.isna(sr.get("otpid")) else str(sr.get("otpid")) if cand0 == first: match_idx = i break if match_idx is None: logger.warning( "Unable to locate DataFrame row matching selected row values: %r", vals, ) return idx = int(match_idx) else: # 3) If we have a row_key, try to normalize to an int index if row_key is not None: try: idx = int(row_key) except Exception: # Try converting via string try: idx = int(str(row_key)) except Exception: idx = None if idx is None: # Final numeric fallback: use sessions_table.cursor_row if present try: idx = getattr(self.sessions_table, "cursor_row") except Exception: idx = None if idx is None: _bad("Failed to normalize row/key from event: %r", row_key) return else: # 4) Try table cursor_row as last resort try: idx = getattr(self.sessions_table, "cursor_row") except Exception: logger.warning( "Could not determine selected row from event: %r", event ) # Helpful debug hint for you to paste back if still failing: logger.debug("Event repr for debugging: %r", event) return # At this point we should have an integer idx try: idx = int(idx) except Exception: logger.exception( "Final normalization of selected row index failed: %r", idx ) return # Validate sessions df if self._sessions_df is None or self._sessions_df.empty: logger.warning("Sessions DataFrame empty; nothing to select.") return df_ordered = self._sessions_df.reset_index(drop=True) if idx < 0 or idx >= len(df_ordered): logger.warning( "Selected row index %s out of range (0..%d)", idx, len(df_ordered) - 1 ) return row_series = df_ordered.iloc[idx] otpid = row_series.get("otpid") hostname = row_series.get("hostname") # Store selected session and fetch activities self._selected_session_otpid = otpid # Obtain api from app (try multiple places) api = ( getattr(self.app, "api", None) or getattr(self, "api", None) or getattr(self.app, "airlock_api", None) ) if api is None: logger.error("No API available on self.app.api - cannot fetch activities") return logger.info( "Fetching activities for otpid=%s host=%s (selected row=%s)", otpid, hostname, idx, ) await self._fetch_activities_for_otpid(api, otpid, hostname=hostname) async def load_sessions_from_api(self, api) -> None: """ Pulls OTP session lists, adds status column, concatenates and populates the sessions table. """ try: active = api.otp_find_active() awaiting = api.otp_find_awaiting() enforced = api.otp_find_enforced() revoked = api.otp_find_revoked() except Exception as exc: logger.exception("Failed to fetch OTP session lists: %s", exc) # Present empty active = awaiting = enforced = revoked = pd.DataFrame() # Ensure DataFrame objects def _ensure_df(df): return df if isinstance(df, pd.DataFrame) else pd.DataFrame(df) active = _ensure_df(active) awaiting = _ensure_df(awaiting) enforced = _ensure_df(enforced) revoked = _ensure_df(revoked) for df, status in [ (active, "active"), (awaiting, "awaiting"), (enforced, "enforced"), (revoked, "revoked"), ]: if "status" not in df.columns: df["status"] = status combined = pd.concat([active, awaiting, enforced, revoked], ignore_index=True) if "otpid" in combined.columns: combined = combined.sort_values(by="otpid", ascending=False) self._sessions_df = combined # Populate DataTable self.sessions_table.clear() # Ensure columns exist in DF and when missing add empty column expected_cols = ["otpid", "hostname", "status", "purpose", "granted"] for col in expected_cols: if col not in combined.columns: combined[col] = "" self.sessions_table.add_columns(*expected_cols) # Add rows for _, row in combined[expected_cols].iterrows(): # Convert values to str for safe insertion vals = ["" if pd.isna(v) else v for v in row.to_list()] self.sessions_table.add_row(*[str(v) for v in vals]) logger.info("Loaded %d OTP sessions.", len(combined)) async def _fetch_activities_for_otpid(self, api, otpid, hostname=None) -> None: """ Fetch activities DataFrame for a given otpid and populate activities_table. """ try: result = api.otp_get_activities(otpid) result_df = ( result if isinstance(result, pd.DataFrame) else pd.DataFrame(result) ) except Exception as exc: logger.exception("Failed to fetch activities for otpid %s: %s", otpid, exc) result_df = pd.DataFrame() # Attach hostname if provided if hostname is not None: result_df["hostname"] = hostname if result_df.empty: logger.info("No activities found for otpid %s (host: %s)", otpid, hostname) self._activities_df = pd.DataFrame() self.activities_table.clear() return # Store and render self._activities_df = result_df.copy() # Rebuild activities_table columns from result_df self.activities_table.clear() # Ensure stable column order for col in result_df.columns: self.activities_table.add_column(col) # Add rows for _, arow in result_df.iterrows(): values = ["" if pd.isna(v) else v for v in arow.to_list()] self.activities_table.add_row(*[str(v) for v in values]) logger.info( "Loaded %d activity rows for otpid %s (host: %s)", len(result_df), otpid, hostname, ) async def export_activities(self) -> None: """ Export currently-loaded activities DataFrame to CSV. Can be called directly (programmatically) or from the button handler. """ if self._activities_df is None or self._activities_df.empty: logger.info("No activities loaded to export.") # On-screen short message await self.post_message(Static("No activities to export.")) return working_dir = _load_working_dir() timestamp = datetime.now().strftime("%Y-%m-%d_%H-%M-%S") filename = f"otp_activities_{self._selected_session_otpid}_{timestamp}.csv" file_path = os.path.join(working_dir, filename) try: self._activities_df.to_csv(file_path, index=False) logger.info("Exported activities to %s", file_path) await self.post_message(Static(f"Exported activities to: {file_path}")) except Exception as exc: logger.exception("Failed to export activities to %s: %s", file_path, exc) await self.post_message(Static("Failed to export activities; check logs.")) class ActivityDetailWidget(Static): """ Interactive widget for Activity Detail screen. Shows the provided DataFrame in a DataTable and offers Export button. Now includes Select All/None and Add to Allowlist functionality. """ DEFAULT_CSS = """ ActivityDetailWidget { height: 1fr; layout: vertical; } #detail_table_container { height: 1fr; padding: 1 1; } #selection_buttons { height: auto; min-height: 3; padding: 1 1; content-align: center middle; } #detail_buttons { height: auto; min-height: 3; padding: 1 1; content-align: center middle; } """ def __init__(self, activities_df: pd.DataFrame, otpid=None, hostname=None) -> None: super().__init__() self.activities_df = ( activities_df.copy() if isinstance(activities_df, pd.DataFrame) else pd.DataFrame(activities_df) ) # Add a unique identifier column if not present if "_row_id" not in self.activities_df.columns: self.activities_df["_row_id"] = range(len(self.activities_df)) self.otpid = otpid self.hostname = hostname self.selected_row_ids = set() # Track selected rows by unique ID self.row_key_to_id = {} # Map DataTable row keys to unique row IDs self.table_row_to_id = {} # Map table row indices to unique row IDs self._last_sort = None # Track last sort column and order def compose(self) -> ComposeResult: yield Static( f"Activity Detail (otpid={self.otpid} host={self.hostname})", classes="panel-title", ) # Table container with Vertical(id="detail_table_container"): self.detail_table = DataTable(id="detail_table") yield self.detail_table # Original buttons at bottom with Horizontal(id="detail_buttons"): self.add_allowlist_btn = Button( "Add Selected to Allowlist", id="add_allowlist_btn" ) yield self.add_allowlist_btn async def on_mount(self) -> None: await self._build_table(rebuild=True) self._update_button_states() def _update_button_states(self) -> None: """Update button states based on selection.""" has_selection = len(self.selected_row_ids) > 0 self.add_allowlist_btn.disabled = not has_selection # Update button labels with count count = len(self.selected_row_ids) len(self.activities_df) if has_selection: self.add_allowlist_btn.label = f"Add {count} Selected to Allowlist" else: self.add_allowlist_btn.label = "Add Selected to Allowlist" async def _build_table(self, rebuild: bool = True) -> None: """Rebuild the DataTable. If rebuild=False, only refresh rows.""" if rebuild: # Full rebuild: clear columns and rows self.detail_table.clear() self.detail_table.columns.clear() self.row_key_to_id.clear() self.table_row_to_id.clear() if self.activities_df is None or self.activities_df.empty: logger.info("ActivityDetailWidget mounted with empty dataframe.") return # Add columns (checkbox + data columns, excluding internal _row_id) self.detail_table.add_column("Select", key="select") for col in self.activities_df.columns: if col != "_row_id": # Don't display the internal ID column self.detail_table.add_column(col) else: # Partial rebuild: clear rows only self.detail_table.clear() self.row_key_to_id.clear() self.table_row_to_id.clear() # Add rows for table_idx, (df_idx, row) in enumerate(self.activities_df.iterrows()): # Get the unique row ID row_id = row["_row_id"] # Build values list (excluding _row_id column) vals = [] for col in self.activities_df.columns: if col != "_row_id": v = row[col] vals.append("" if pd.isna(v) else str(v)) # Check if this row is selected checkbox = "☑️" if row_id in self.selected_row_ids else "☐" # Add row to table row_key = self.detail_table.add_row(checkbox, *vals) # Map the row key and table index to the unique row ID self.row_key_to_id[row_key] = row_id self.table_row_to_id[table_idx] = row_id async def on_data_table_cell_selected(self, event: DataTable.CellSelected) -> None: # Toggle selection when the "Select" column is clicked if event.cell_key.column_key.value == "select": table_row_index = event.coordinate.row # Get the unique row ID for this table row row_id = self.table_row_to_id.get(table_row_index) if row_id is not None: # Get the row key for updating the cell row_key = event.cell_key.row_key if row_id in self.selected_row_ids: self.selected_row_ids.remove(row_id) self.detail_table.update_cell(row_key, "select", "☐") # Unchecked else: self.selected_row_ids.add(row_id) self.detail_table.update_cell(row_key, "select", "☑️") # Checked self._update_button_states() async def on_data_table_header_selected( self, event: DataTable.HeaderSelected ) -> None: column_key = event.column_key.value if event.column_key else None if not column_key: col_index = event.column_index if col_index == 0: # First column is "Select" return # Adjust for hidden _row_id column visible_cols = [ col for col in self.activities_df.columns if col != "_row_id" ] if col_index - 1 < len(visible_cols): column_key = visible_cols[col_index - 1] else: return if column_key == "select" or column_key == "_row_id": return ascending = True if self._last_sort == (column_key, True): ascending = False self._last_sort = (column_key, ascending) try: self.activities_df.sort_values( by=column_key, ascending=ascending, inplace=True ) except Exception as exc: logger.exception("Failed to sort by column %s: %s", column_key, exc) return # Only refresh rows, not columns await self._build_table(rebuild=False) async def on_button_pressed(self, event) -> None: btn = getattr(event, "button", None) or getattr(event, "sender", None) btn_id = getattr(btn, "id", None) or getattr(event, "button_id", None) if btn is self.add_allowlist_btn or btn_id == "add_allowlist_btn": await self._open_allowlist_screen() return async def _select_all(self) -> None: """Select all rows in the table.""" # Add all row IDs to selected set self.selected_row_ids = set(self.activities_df["_row_id"].tolist()) # Update all checkboxes in the table for row_key, row_id in self.row_key_to_id.items(): self.detail_table.update_cell(row_key, "select", "☑️") self._update_button_states() logger.info(f"Selected all {len(self.selected_row_ids)} rows") async def _select_none(self) -> None: """Deselect all rows in the table.""" # Clear selected set self.selected_row_ids.clear() # Update all checkboxes in the table for row_key, row_id in self.row_key_to_id.items(): self.detail_table.update_cell(row_key, "select", "☑️") self._update_button_states() logger.info("Cleared all selections") async def _open_allowlist_screen(self) -> None: """Open the allowlist selection screen with selected activities.""" if not self.selected_row_ids: self.app.notify("No rows selected", severity="warning") return # Get selected data selected_df = self.get_selected_data() # Get API from app api = getattr(self.app, "api", None) if api is None: logger.error("No API available on self.app.api") self.app.notify("API not available", severity="error") return # Create and push AllowlistSelectionScreen try: allowlist_screen = AllowlistSelectionScreen( selected_df, api=api, hostname=self.hostname, otpid=self.otpid ) await self.app.push_screen(allowlist_screen) logger.info( f"Opened allowlist screen with {len(selected_df)} selected activities" ) except ImportError as e: logger.error(f"Failed to import AllowlistSelectionScreen: {e}") self.app.notify("Allowlist screen module not found", severity="error") except Exception as e: logger.exception(f"Failed to open allowlist screen: {e}") self.app.notify( f"Error opening allowlist screen: {str(e)}", severity="error" ) async def _export_detail_activities(self) -> None: if self.activities_df is None or self.activities_df.empty: logger.info("No activities to export.") await self.mount(Static("No activities to export.", classes="notification")) return if not self.selected_row_ids: logger.info("No rows selected for export.") await self.mount( Static("No rows selected for export.", classes="notification") ) return try: working_dir = load_env("WORKING_DIR") or os.getcwd() timestamp = datetime.now().strftime("%Y-%m-%d_%H-%M-%S") filename = f"otp_activities_detail_{timestamp}.csv" file_path = os.path.join(working_dir, filename) selected_df = self.get_selected_data() selected_df.to_csv(file_path, index=False) logger.info("Exported selected activities to %s", file_path) await self.mount( Static( f"Exported selected activities to: {filename}", classes="notification", ) ) except Exception as exc: logger.exception("Failed to export detail activities: %s", exc) await self.mount( Static( "¢ Failed to export activities; check logs.", classes="notification", ) ) # Helper methods def get_selected_data(self) -> pd.DataFrame: """Return a DataFrame of the selected rows.""" if not self.selected_row_ids: return pd.DataFrame() # Filter by selected row IDs and drop the internal _row_id column selected_df = self.activities_df[ self.activities_df["_row_id"].isin(self.selected_row_ids) ].copy() if "_row_id" in selected_df.columns: selected_df = selected_df.drop(columns=["_row_id"]) return selected_df def get_selected_records(self) -> list[dict]: """Return selected rows as a list of dicts.""" if not self.selected_row_ids: return [] # Filter by selected row IDs and drop the internal _row_id column selected_df = self.activities_df[ self.activities_df["_row_id"].isin(self.selected_row_ids) ].copy() if "_row_id" in selected_df.columns: selected_df = selected_df.drop(columns=["_row_id"]) return selected_df.to_dict(orient="records") class ActivityDetailScreen(Screen): """ Screen that wraps ActivityDetailWidget. Expects a DataFrame passed on init. """ BINDINGS = [ Binding("escape", "go_back", "Back"), Binding("q", "main_menu", "Main Menu"), Binding("e", "export", "Export"), Binding("a", "select_all", "Select All"), Binding("n", "select_none", "Select None"), ] def __init__(self, activities_df: pd.DataFrame, otpid=None, hostname=None) -> None: super().__init__() self._activities_df = ( activities_df.copy() if isinstance(activities_df, pd.DataFrame) else pd.DataFrame(activities_df) ) self._otpid = otpid self._hostname = hostname def compose(self) -> ComposeResult: self.widget = ActivityDetailWidget( self._activities_df, otpid=self._otpid, hostname=self._hostname ) yield Header(show_clock=True) yield self.widget yield Footer() async def action_go_back(self) -> None: try: await self.app.pop_screen() except Exception: logger.debug("ActivityDetailScreen.action_go_back pop_screen failed.") async def action_main_menu(self) -> None: """Go back to main menu.""" while len(self.app.screen_stack) > 2: await self.app.pop_screen() async def action_export(self) -> None: # Delegate to widget export helper if hasattr(self, "widget") and self.widget is not None: await self.widget._export_detail_activities() async def action_select_all(self) -> None: """Handle 'a' key for select all.""" if hasattr(self, "widget") and self.widget is not None: await self.widget._select_all() async def action_select_none(self) -> None: """Handle 'n' key for select none.""" if hasattr(self, "widget") and self.widget is not None: await self.widget._select_none() class OTPActivitiesScreen(Screen): """ A Screen intended to be pushed into an existing Textual App. Usage: app.push_screen(OTPActivitiesScreen()) or create this screen and call `await screen.load()` inside your app lifecycle. The screen expects `self.app.api` to exist and be an AirlockAPIWrapper instance. """ BINDINGS = [ Binding("escape", "go_back", "Back"), Binding("q", "main_menu", "Main Menu"), Binding("r", "refresh_sessions", "Refresh Sessions"), Binding("e", "export_activities", "Export activities"), ] def compose(self) -> ComposeResult: yield Header() self.widget = OTPActivitiesWidget() yield self.widget yield Footer() async def on_show(self) -> None: """Restore focus to the left sessions table when the screen becomes visible.""" if hasattr(self, "widget") and hasattr(self.widget, "sessions_table"): self.widget.sessions_table.focus() async def on_mount(self) -> None: # Try to load sessions immediately api = getattr(self.app, "api", None) if api is None: logger.warning("OTPActivitiesScreen mounted but no self.app.api found.") else: await self.widget.load_sessions_from_api(api) async def action_go_back(self) -> None: """Go back one screen.""" await self.app.pop_screen() async def action_main_menu(self) -> None: """Go back to main menu.""" while len(self.app.screen_stack) > 2: await self.app.pop_screen() # Simple actions bound to keys async def action_refresh_sessions(self) -> None: api = getattr(self.app, "api", None) if api is None: logger.error("No API on app; cannot refresh sessions.") return logger.info("Refreshing OTP sessions via API.") await self.widget.load_sessions_from_api(api) # If you want an explicit method to fetch activities for a particular otpid from outside: async def fetch_activities_for_otpid(self, otpid, hostname=None) -> None: api = getattr(self.app, "api", None) if api is None: logger.error("No API on app; cannot fetch activities.") return await self.widget._fetch_activities_for_otpid(api, otpid, hostname=hostname)