Add log tab in admin panel

src/routers/logs.py

@@ -0,0 +1,81 @@
+import os
+from typing import Optional
+
+from fastapi import APIRouter, Depends, HTTPException, Query, status
+from pydantic import BaseModel
+
+from core.config import LOG_FILE
+from core.dependencies import require_admin
+
+router = APIRouter(prefix="/api/admin/logs", tags=["admin-logs"])
+
+_LEVEL_ORDER = {"DEBUG": 0, "INFO": 1, "WARNING": 2, "ERROR": 3, "CRITICAL": 4}
+
+
+class LogsResponse(BaseModel):
+    lines: list[str]
+    log_file: Optional[str]
+    total_returned: int
+
+
+def _read_tail(path: str, n: int) -> list[str]:
+    """Return the last *n* lines from *path* using a memory-efficient seek."""
+    size = os.path.getsize(path)
+    if size == 0:
+        return []
+    # Read in chunks from the end
+    chunk = 1 << 14  # 16 KB
+    lines: list[bytes] = []
+    with open(path, "rb") as fh:
+        pos = size
+        while len(lines) <= n and pos > 0:
+            pos = max(pos - chunk, 0)
+            fh.seek(pos)
+            data = fh.read(min(chunk, size - pos))
+            lines = data.split(b"\n") + lines
+            # If we have more lines than needed, trim from the front
+            if len(lines) > n + 1:
+                lines = lines[-(n + 1):]
+    # Drop empty last element caused by trailing newline
+    result = [ln.decode("utf-8", errors="replace") for ln in lines if ln]
+    return result[-n:] if len(result) > n else result
+
+
+@router.get("", response_model=LogsResponse)
+async def get_logs(
+    _: dict = Depends(require_admin),
+    lines: int = Query(200, ge=1, le=2000, description="Number of lines to return (tail)"),
+    level: Optional[str] = Query(None, description="Minimum log level: DEBUG, INFO, WARNING, ERROR, CRITICAL"),
+):
+    if not LOG_FILE:
+        raise HTTPException(
+            status_code=status.HTTP_404_NOT_FOUND,
+            detail="Log file not configured (LOG_FILE env variable is not set).",
+        )
+    if not os.path.isfile(LOG_FILE):
+        return LogsResponse(lines=[], log_file=LOG_FILE, total_returned=0)
+
+    raw = _read_tail(LOG_FILE, lines)
+
+    if level:
+        level_upper = level.upper()
+        if level_upper not in _LEVEL_ORDER:
+            raise HTTPException(
+                status_code=status.HTTP_422_UNPROCESSABLE_ENTITY,
+                detail=f"Invalid level '{level}'. Must be one of: {', '.join(_LEVEL_ORDER)}.",
+            )
+        min_order = _LEVEL_ORDER[level_upper]
+        filtered: list[str] = []
+        for line in raw:
+            for lvl, order in _LEVEL_ORDER.items():
+                if lvl in line and order >= min_order:
+                    filtered.append(line)
+                    break
+            else:
+                # Lines that don't match any known level keyword are included
+                # only when filtering at DEBUG (include everything)
+                if min_order == 0:
+                    filtered.append(line)
+        raw = filtered
+
+    return LogsResponse(lines=raw, log_file=LOG_FILE, total_returned=len(raw))