Zarithas
a7b659c951
- Added row copy functionality (Ctrl+C) - Improved row selection visual contrast - Fixed data state issues when navigating between stages Server Log tab - Added new Server Log tab to the main application - Implemented live filtering with wildcard support - Enabled auto-refresh capability Multiagent selector - Added file loading support for device lists
262 lines
8.8 KiB
Python
262 lines
8.8 KiB
Python
# 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 <https://www.gnu.org/licenses/>.
|
|
|
|
import datetime
|
|
import logging
|
|
|
|
from bson import ObjectId
|
|
from textual.app import ComposeResult
|
|
from textual.containers import Container, Vertical
|
|
from textual.widgets import Button, DataTable, Input, Static
|
|
|
|
from services.API import AirlockAPIWrapper
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
|
|
def skipback(days):
|
|
"""
|
|
Generate a MongoDB ObjectId for a given number of days ago from today.
|
|
"""
|
|
adjusted_days = days
|
|
date_days_ago = datetime.datetime.now(datetime.UTC) - datetime.timedelta(
|
|
days=adjusted_days
|
|
)
|
|
timestamp = int(date_days_ago.timestamp())
|
|
hex_timestamp = format(timestamp, "08x")
|
|
objectid_hex = hex_timestamp + "0000000000000000"
|
|
return ObjectId(objectid_hex)
|
|
|
|
|
|
class ServerLogWidget(Vertical):
|
|
"""Widget for displaying server activity logs in a DataTable."""
|
|
|
|
DEFAULT_CSS = """
|
|
ServerLogWidget {
|
|
width: 100%;
|
|
height: 100%;
|
|
}
|
|
|
|
ServerLogWidget #status_bar {
|
|
width: 100%;
|
|
height: auto;
|
|
background: $surface;
|
|
padding: 1;
|
|
margin-bottom: 1;
|
|
}
|
|
|
|
ServerLogWidget DataTable {
|
|
height: 1fr;
|
|
border: solid $primary;
|
|
}
|
|
|
|
ServerLogWidget #button_container {
|
|
width: 100%;
|
|
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 {
|
|
margin-right: 1;
|
|
}
|
|
"""
|
|
|
|
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:
|
|
"""Initialize the DataTable and load server logs."""
|
|
self.load_logs()
|
|
|
|
def load_logs(self) -> None:
|
|
"""Load server logs from the API and populate the DataTable."""
|
|
table = self.query_one("#server_log_table", DataTable)
|
|
status = self.query_one("#status_bar", Static)
|
|
|
|
try:
|
|
status.update("⏳ Loading server logs (last 72 hours)...")
|
|
|
|
# Create a fake checkpoint for 3 days ago (72 hours)
|
|
checkpoint = str(skipback(3))
|
|
|
|
# Get server logs from API
|
|
logs = self.api.server_logs(checkpoint=checkpoint)
|
|
|
|
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]
|
|
self.columns = [col for col in first_log.keys() if col != "checkpoint"]
|
|
|
|
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 self.columns:
|
|
value = log_entry.get(col, "")
|
|
# Format datetime column to be more readable
|
|
if col == "datetime" and value:
|
|
try:
|
|
# Parse ISO format and convert to readable format
|
|
dt = datetime.datetime.fromisoformat(
|
|
str(value).replace("Z", "+00:00")
|
|
)
|
|
value = dt.strftime("%Y-%m-%d %H:%M:%S")
|
|
except Exception:
|
|
# If parsing fails, just use the original value
|
|
pass
|
|
row_data.append(str(value))
|
|
table.add_row(*row_data)
|
|
|
|
status.update(
|
|
f"✅ Loaded {len(logs)} log entries from the last 72 hours"
|
|
)
|
|
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."""
|
|
button_id = event.button.id
|
|
|
|
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()
|