# -*- coding: utf-8 -*- """ ActivityTracker — Zentrales Activity-Tracking als IService. Protokolliert Verbindungen, Intents, Fehler, Trainings und Conversations. Daten werden in JSON-Dateien unter data/activity/ persistiert mit konfigurierbarer Retention. """ from __future__ import annotations import asyncio import json import time from dataclasses import asdict from datetime import datetime, timedelta, timezone from pathlib import Path from typing import TYPE_CHECKING, Any from trixy_core.service.iservice import IService from trixy_core.service.enums import ServicePriority, ServiceGroup from trixy_core.events.decorators import TrixyEvent from trixy_core.activity.models import ( SatelliteConnectionEntry, ConfigConnectionEntry, TrainingRunEntry, IntentLogEntry, ErrorLogEntry, ConversationEntry, ) from trixy_core.utils.debug import pinfo, pdebug, perror if TYPE_CHECKING: from trixy_core.application import IApplication # Maximale Eintraege pro Kategorie MAX_ENTRIES = 10000 # Flush-Intervall in Sekunden FLUSH_INTERVAL = 60 # Cleanup-Intervall in Sekunden (1 Stunde) CLEANUP_INTERVAL = 3600 def _now_iso() -> str: """Gibt den aktuellen Zeitstempel als ISO-8601 String zurueck.""" return datetime.now(timezone.utc).isoformat() def _parse_iso(iso_str: str) -> datetime | None: """Parst einen ISO-8601 String, gibt None bei Fehler zurueck.""" if not iso_str: return None try: return datetime.fromisoformat(iso_str) except (ValueError, TypeError): return None class ActivityTracker(IService): """ Zentraler Activity-Tracker fuer Server-Aktivitaeten. Lauscht auf Events und bietet Logging-Methoden fuer externe Aufrufer. Persistiert Daten in JSON-Dateien und bereinigt alte Eintraege basierend auf retention_days. """ NAME = "ActivityTracker" PRIORITY = ServicePriority.CORE GROUP = ServiceGroup.UTILITY DEPENDENCIES: list[str] = [] # Dateinamen fuer JSON-Persistenz _FILES = { "satellite_connections": "satellite_connections.json", "config_connections": "config_connections.json", "training_runs": "training_runs.json", "intent_log": "intent_log.json", "error_log": "error_log.json", "conversations": "conversations.json", } def __init__( self, application: "IApplication", retention_days: int = 7, data_directory: str = "data/activity", ) -> None: super().__init__(application) self._retention_days = retention_days self._data_dir = Path(data_directory) # In-Memory-Listen self._satellite_connections: list[dict[str, Any]] = [] self._config_connections: list[dict[str, Any]] = [] self._training_runs: list[dict[str, Any]] = [] self._intent_log: list[dict[str, Any]] = [] self._error_log: list[dict[str, Any]] = [] self._conversations: list[dict[str, Any]] = [] # Hintergrund-Tasks self._flush_task: asyncio.Task | None = None self._cleanup_task: asyncio.Task | None = None self._dirty = False async def start(self) -> None: """Startet den Tracker: Daten laden, Events registrieren, Hintergrund-Tasks starten.""" self._data_dir.mkdir(parents=True, exist_ok=True) self._load_all() self._cleanup_old_entries() # Event-Handler registrieren self._application.events.register("satellite_connected", self._on_satellite_connected) self._application.events.register("satellite_disconnected", self._on_satellite_disconnected) self._application.events.register("intent_handled", self._on_intent_handled) self._application.events.register("conversation_started", self._on_conversation_started) self._application.events.register("conversation_ended", self._on_conversation_ended) # Periodische Tasks starten self._flush_task = asyncio.create_task(self._periodic_flush()) self._cleanup_task = asyncio.create_task(self._periodic_cleanup()) pinfo(f"ActivityTracker gestartet (Retention: {self._retention_days} Tage, " f"Pfad: {self._data_dir})") async def stop(self) -> None: """Stoppt den Tracker und flusht ausstehende Daten.""" if self._flush_task: self._flush_task.cancel() self._flush_task = None if self._cleanup_task: self._cleanup_task.cancel() self._cleanup_task = None self._flush_all() pinfo("ActivityTracker gestoppt") # ========================================================================= # Event-Handler # ========================================================================= async def _on_satellite_connected(self, event_name: str, event_data: Any) -> None: """Neuen Satellite-Verbindungseintrag erstellen.""" entry = SatelliteConnectionEntry( satellite_id=getattr(event_data, "satellite_id", ""), alias=getattr(event_data, "alias", ""), room=getattr(event_data, "room", ""), connected_at=_now_iso(), ip_address=getattr(event_data, "ip_address", ""), ) self._satellite_connections.append(asdict(entry)) self._trim_list(self._satellite_connections) self._dirty = True pdebug(f"[ActivityTracker] Satellite verbunden: {entry.alias}") async def _on_satellite_disconnected(self, event_name: str, event_data: Any) -> None: """Disconnected-Zeitpunkt im letzten passenden Eintrag setzen.""" sid = getattr(event_data, "satellite_id", "") now = _now_iso() # Letzten offenen Eintrag fuer diesen Satellite finden for entry in reversed(self._satellite_connections): if entry.get("satellite_id") == sid and not entry.get("disconnected_at"): entry["disconnected_at"] = now self._dirty = True pdebug(f"[ActivityTracker] Satellite getrennt: {sid}") return async def _on_intent_handled(self, event_name: str, event_data: Any) -> None: """Neuen Intent-Log-Eintrag erstellen.""" # Plugin-Name aus IntentRegistry ermitteln plugin_name = "" try: from trixy_core.nlp.intent_registry import IntentRegistry registry = IntentRegistry.get_instance() intent_name = getattr(event_data, "intent", "") definition = registry.get(intent_name) if definition: plugin_name = definition.plugin_name except Exception: pass entry = IntentLogEntry( timestamp=_now_iso(), intent=getattr(event_data, "intent", ""), plugin=plugin_name, satellite_id=getattr(event_data, "satellite_id", ""), room_id=getattr(event_data, "room_id", ""), original_text=getattr(event_data, "original_text", ""), success=getattr(event_data, "success", True), ) self._intent_log.append(asdict(entry)) self._trim_list(self._intent_log) self._dirty = True async def _on_conversation_started(self, event_name: str, event_data: Any) -> None: """Neuen Conversation-Eintrag erstellen.""" entry = ConversationEntry( session_id=getattr(event_data, "conversation_id", ""), satellite_id=getattr(event_data, "satellite_id", ""), started_at=_now_iso(), ) self._conversations.append(asdict(entry)) self._trim_list(self._conversations) self._dirty = True async def _on_conversation_ended(self, event_name: str, event_data: Any) -> None: """Completed-Zeitpunkt im passenden Conversation-Eintrag setzen.""" session_id = getattr(event_data, "conversation_id", "") now = _now_iso() for entry in reversed(self._conversations): if entry.get("session_id") == session_id and not entry.get("completed_at"): entry["completed_at"] = now self._dirty = True return # ========================================================================= # Oeffentliche Logging-Methoden (fuer externe Aufrufer) # ========================================================================= def log_config_connection(self, peer: str) -> None: """Protokolliert eine neue Config-Tool-Verbindung.""" entry = ConfigConnectionEntry( peer_address=peer, connected_at=_now_iso(), ) self._config_connections.append(asdict(entry)) self._trim_list(self._config_connections) self._dirty = True def log_config_disconnection(self, peer: str) -> None: """Setzt den Disconnected-Zeitpunkt fuer eine Config-Tool-Verbindung.""" now = _now_iso() for entry in reversed(self._config_connections): if entry.get("peer_address") == peer and not entry.get("disconnected_at"): entry["disconnected_at"] = now self._dirty = True return def log_training_start(self, trainer_id: str, model_name: str) -> None: """Protokolliert den Start eines Trainings.""" entry = TrainingRunEntry( trainer_id=trainer_id, model_name=model_name, started_at=_now_iso(), ) self._training_runs.append(asdict(entry)) self._trim_list(self._training_runs) self._dirty = True def log_training_complete(self, trainer_id: str, success: bool, error: str = "") -> None: """Protokolliert den Abschluss eines Trainings.""" now = _now_iso() for entry in reversed(self._training_runs): if entry.get("trainer_id") == trainer_id and not entry.get("completed_at"): entry["completed_at"] = now entry["success"] = success entry["error"] = error # Dauer berechnen started = _parse_iso(entry.get("started_at", "")) if started: completed = _parse_iso(now) if completed: entry["duration_seconds"] = (completed - started).total_seconds() self._dirty = True return def log_error(self, source: str, message: str, satellite_id: str = "") -> None: """Protokolliert einen Fehler.""" entry = ErrorLogEntry( timestamp=_now_iso(), source=source, message=message, satellite_id=satellite_id, ) self._error_log.append(asdict(entry)) self._trim_list(self._error_log) self._dirty = True # ========================================================================= # Query-Methoden (fuer AdminCommands Plugin) # ========================================================================= def get_last_intents(self, count: int = 5) -> list[dict[str, Any]]: """Liefert die letzten N Intents.""" return self._intent_log[-count:] def get_last_intents_by_room(self, room: str, count: int = 5) -> list[dict[str, Any]]: """Liefert die letzten N Intents fuer einen bestimmten Raum.""" filtered = [e for e in self._intent_log if e.get("room_id") == room] return filtered[-count:] def get_last_conversations(self, count: int = 5) -> list[dict[str, Any]]: """Liefert die letzten N Conversations.""" return self._conversations[-count:] def get_last_errors(self, count: int = 5) -> list[dict[str, Any]]: """Liefert die letzten N Fehler.""" return self._error_log[-count:] def get_error_count_since(self, hours: int = 24) -> int: """Zaehlt Fehler der letzten N Stunden.""" cutoff = datetime.now(timezone.utc) - timedelta(hours=hours) count = 0 for entry in self._error_log: ts = _parse_iso(entry.get("timestamp", "")) if ts and ts >= cutoff: count += 1 return count def get_last_training(self, trainer_id: str = "") -> dict[str, Any] | None: """Liefert den letzten Trainings-Lauf (optional gefiltert nach trainer_id).""" for entry in reversed(self._training_runs): if not trainer_id or entry.get("trainer_id") == trainer_id: return entry return None def get_last_config_connection(self) -> dict[str, Any] | None: """Liefert die letzte Config-Tool-Verbindung.""" if self._config_connections: return self._config_connections[-1] return None def get_satellite_connection_history(self, alias: str) -> list[dict[str, Any]]: """Liefert die Verbindungshistorie fuer einen Satellite (nach Alias).""" return [ e for e in self._satellite_connections if e.get("alias", "").lower() == alias.lower() ] def get_intent_stats(self, days: int = 7) -> dict[str, int]: """Liefert Intent-Statistiken der letzten N Tage.""" cutoff = datetime.now(timezone.utc) - timedelta(days=days) stats: dict[str, int] = {} for entry in self._intent_log: ts = _parse_iso(entry.get("timestamp", "")) if ts and ts >= cutoff: intent_name = entry.get("intent", "unknown") stats[intent_name] = stats.get(intent_name, 0) + 1 return stats def clear_all(self) -> None: """Loescht alle Activity-Daten.""" self._satellite_connections.clear() self._config_connections.clear() self._training_runs.clear() self._intent_log.clear() self._error_log.clear() self._conversations.clear() self._dirty = True self._flush_all() pinfo("[ActivityTracker] Alle Daten geloescht") # ========================================================================= # Persistenz # ========================================================================= def _load_all(self) -> None: """Laedt alle JSON-Dateien in den Speicher.""" mapping = { "satellite_connections": "_satellite_connections", "config_connections": "_config_connections", "training_runs": "_training_runs", "intent_log": "_intent_log", "error_log": "_error_log", "conversations": "_conversations", } for key, attr in mapping.items(): file_path = self._data_dir / self._FILES[key] data = self._load_json(file_path) setattr(self, attr, data) total = sum( len(getattr(self, attr)) for attr in mapping.values() ) pdebug(f"[ActivityTracker] {total} Eintraege geladen") def _flush_all(self) -> None: """Schreibt alle In-Memory-Daten in JSON-Dateien.""" if not self._dirty: return mapping = { "satellite_connections": self._satellite_connections, "config_connections": self._config_connections, "training_runs": self._training_runs, "intent_log": self._intent_log, "error_log": self._error_log, "conversations": self._conversations, } for key, data in mapping.items(): file_path = self._data_dir / self._FILES[key] self._save_json(file_path, data) self._dirty = False def _load_json(self, path: Path) -> list[dict[str, Any]]: """Laedt eine JSON-Datei, gibt leere Liste bei Fehler zurueck.""" if not path.exists(): return [] try: with open(path, "r", encoding="utf-8") as f: data = json.load(f) if isinstance(data, list): return data except (json.JSONDecodeError, OSError) as e: perror(f"[ActivityTracker] Fehler beim Laden von {path}: {e}") return [] def _save_json(self, path: Path, data: list[dict[str, Any]]) -> None: """Speichert Daten als JSON-Datei.""" try: with open(path, "w", encoding="utf-8") as f: json.dump(data, f, ensure_ascii=False, indent=2) except OSError as e: perror(f"[ActivityTracker] Fehler beim Speichern von {path}: {e}") def _trim_list(self, lst: list) -> None: """Begrenzt eine Liste auf MAX_ENTRIES Eintraege.""" if len(lst) > MAX_ENTRIES: del lst[: len(lst) - MAX_ENTRIES] # ========================================================================= # Retention / Cleanup # ========================================================================= def _cleanup_old_entries(self) -> None: """Entfernt Eintraege die aelter als retention_days sind.""" cutoff = datetime.now(timezone.utc) - timedelta(days=self._retention_days) before = sum(len(lst) for lst in [ self._satellite_connections, self._config_connections, self._training_runs, self._intent_log, self._error_log, self._conversations, ]) # Timestamp-Feld pro Kategorie ts_fields = { "_satellite_connections": "connected_at", "_config_connections": "connected_at", "_training_runs": "started_at", "_intent_log": "timestamp", "_error_log": "timestamp", "_conversations": "started_at", } for attr, ts_field in ts_fields.items(): lst: list[dict] = getattr(self, attr) filtered = [] for entry in lst: ts = _parse_iso(entry.get(ts_field, "")) if ts is None or ts >= cutoff: filtered.append(entry) setattr(self, attr, filtered) after = sum(len(lst) for lst in [ self._satellite_connections, self._config_connections, self._training_runs, self._intent_log, self._error_log, self._conversations, ]) removed = before - after if removed > 0: self._dirty = True pdebug(f"[ActivityTracker] {removed} alte Eintraege bereinigt") # ========================================================================= # Hintergrund-Tasks # ========================================================================= async def _periodic_flush(self) -> None: """Periodischer Flush der In-Memory-Daten.""" while True: try: await asyncio.sleep(FLUSH_INTERVAL) self._flush_all() except asyncio.CancelledError: break except Exception as e: perror(f"[ActivityTracker] Flush-Fehler: {e}") async def _periodic_cleanup(self) -> None: """Periodische Bereinigung alter Eintraege.""" while True: try: await asyncio.sleep(CLEANUP_INTERVAL) self._cleanup_old_entries() self._flush_all() except asyncio.CancelledError: break except Exception as e: perror(f"[ActivityTracker] Cleanup-Fehler: {e}")