feat: add HTTP log query API endpoints

Co-Authored-By: Claude Opus 4.7 <noreply@anthropic.com>

web/backend/routers/logs.py

@@ -0,0 +1,103 @@
+"""HTTP log query endpoints."""
+
+import logging
+import sqlite3
+from pathlib import Path
+from typing import Optional
+
+from fastapi import APIRouter, Depends, Query
+
+from ..auth.dependencies import get_current_user
+from ..services.db_schema import query_http_logs, query_http_log_stats
+
+logger = logging.getLogger(__name__)
+
+router = APIRouter(prefix="/api/logs", tags=["logs"])
+
+_db_path = Path(__file__).resolve().parent.parent.parent.parent / "data" / "web_data.db"
+
+
+def _count_http_logs(
+    method: str = None,
+    path: str = None,
+    status_code: int = None,
+    start_time: str = None,
+    end_time: str = None,
+) -> int:
+    """Count total matching HTTP log rows for pagination."""
+    conn = sqlite3.connect(_db_path)
+    try:
+        clauses = []
+        params = []
+        if method:
+            clauses.append("method = ?")
+            params.append(method)
+        if path:
+            clauses.append("path LIKE ?")
+            params.append(f"%{path}%")
+        if status_code is not None:
+            clauses.append("status_code = ?")
+            params.append(status_code)
+        if start_time:
+            clauses.append("timestamp >= ?")
+            params.append(start_time)
+        if end_time:
+            clauses.append("timestamp <= ?")
+            params.append(end_time)
+
+        where = (" WHERE " + " AND ".join(clauses)) if clauses else ""
+        row = conn.execute(
+            f"SELECT COUNT(*) as cnt FROM http_logs{where}", params
+        ).fetchone()
+        return row["cnt"] if row else 0
+    finally:
+        conn.close()
+
+
+@router.get("")
+async def list_logs(
+    page: int = Query(1, ge=1),
+    page_size: int = Query(50, ge=1, le=200),
+    method: Optional[str] = None,
+    status_code: Optional[int] = None,
+    path: Optional[str] = None,
+    start_date: Optional[str] = None,
+    end_date: Optional[str] = None,
+    current_user: dict = Depends(get_current_user),
+):
+    """List HTTP logs with filters and pagination."""
+    offset = (page - 1) * page_size
+    items = query_http_logs(
+        method=method,
+        path=path,
+        status_code=status_code,
+        start_time=start_date,
+        end_time=end_date,
+        limit=page_size,
+        offset=offset,
+    )
+    total = _count_http_logs(
+        method=method,
+        path=path,
+        status_code=status_code,
+        start_time=start_date,
+        end_time=end_date,
+    )
+    return {"items": items, "total": total}
+
+
+@router.get("/stats")
+async def log_stats(
+    current_user: dict = Depends(get_current_user),
+):
+    """Get today's HTTP log statistics."""
+    raw = query_http_log_stats()
+    total = raw.get("total", 0)
+    errors = raw.get("errors", 0)
+    avg_duration = raw.get("avg_duration")
+    return {
+        "today_count": total,
+        "error_count": errors,
+        "avg_duration_ms": round(avg_duration, 2) if avg_duration else 0.0,
+        "error_rate": round(errors / total, 4) if total > 0 else 0.0,
+    }