Sanitized mirror from private repository - 2026-04-18 11:13:17 UTC

dashboard/api/routers/logs.py

@@ -0,0 +1,59 @@
+"""Unified log viewer routes."""
+
+from fastapi import APIRouter, Query
+from pathlib import Path
+
+router = APIRouter(tags=["logs"])
+
+LOG_DIR = Path("/app/logs") if Path("/app/logs").exists() else Path("/tmp")
+
+LOG_FILES = {
+    "stack-restart": "stack-restart.log",
+    "backup": "backup-validator.log",
+    "gmail-lz": "gmail-organizer.log",
+    "gmail-dvish": "gmail-organizer-dvish.log",
+    "proton": "proton-organizer.log",
+    "receipt": "receipt-tracker.log",
+    "drift": "config-drift.log",
+    "digest": "email-digest.log",
+    "disk": "disk-predictor.log",
+    "changelog": "changelog-generator.log",
+    "subscription": "subscription-auditor.log",
+    "pr-review": "pr-reviewer.log",
+}
+
+
+@router.get("/logs")
+def list_logs():
+    """List available log files with sizes."""
+    result = []
+    for name, filename in LOG_FILES.items():
+        path = LOG_DIR / filename
+        if path.exists():
+            stat = path.stat()
+            result.append({
+                "name": name,
+                "filename": filename,
+                "size_bytes": stat.st_size,
+                "modified": stat.st_mtime,
+            })
+    return result
+
+
+@router.get("/logs/{log_name}")
+def get_log(log_name: str, tail: int = Query(200, le=2000), search: str = Query(None)):
+    """Get log file contents."""
+    if log_name not in LOG_FILES:
+        return {"error": f"Unknown log: {log_name}", "lines": []}
+    path = LOG_DIR / LOG_FILES[log_name]
+    if not path.exists():
+        return {"lines": [], "total": 0}
+
+    with open(path) as f:
+        all_lines = f.readlines()
+
+    if search:
+        all_lines = [l for l in all_lines if search.lower() in l.lower()]
+
+    lines = all_lines[-tail:]
+    return {"lines": [l.rstrip() for l in lines], "total": len(all_lines)}