import json import sqlite3 import threading from datetime import datetime from pathlib import Path from typing import Any _ALLOWED_LEVELS = {"DEBUG", "INFO", "WARNING", "ERROR", "CRITICAL"} class LogStore: def __init__(self, db_path: Path, max_rows: int = 50000) -> None: self.db_path = db_path self.max_rows = max(1000, int(max_rows)) self._lock = threading.Lock() self._writes_since_prune = 0 def _connect(self) -> sqlite3.Connection: self.db_path.parent.mkdir(parents=True, exist_ok=True) conn = sqlite3.connect(self.db_path) conn.row_factory = sqlite3.Row return conn def ensure_schema(self) -> None: with self._connect() as conn: conn.execute( """ CREATE TABLE IF NOT EXISTS logs ( id INTEGER PRIMARY KEY AUTOINCREMENT, created_at TEXT NOT NULL DEFAULT (STRFTIME('%Y-%m-%d %H:%M:%f', 'now', 'localtime')), level TEXT NOT NULL, category TEXT NOT NULL, event TEXT NOT NULL, message TEXT NOT NULL, username TEXT, backup_id INTEGER, request_path TEXT, request_method TEXT, status_code INTEGER, details_json TEXT ) """ ) conn.execute("CREATE INDEX IF NOT EXISTS idx_logs_created_at ON logs(created_at DESC)") conn.execute("CREATE INDEX IF NOT EXISTS idx_logs_level ON logs(level)") conn.execute("CREATE INDEX IF NOT EXISTS idx_logs_category ON logs(category)") conn.commit() def write( self, *, level: str, category: str, event: str, message: str, username: str | None = None, backup_id: int | None = None, request_path: str | None = None, request_method: str | None = None, status_code: int | None = None, details: dict[str, Any] | None = None, ) -> None: lvl = (level or "INFO").upper() if lvl not in _ALLOWED_LEVELS: lvl = "INFO" safe_category = (category or "app").strip()[:50] or "app" safe_event = (event or "event").strip()[:80] or "event" safe_message = (message or "").strip()[:3000] or "-" safe_username = (username or "").strip()[:120] or None safe_path = (request_path or "").strip()[:500] or None safe_method = (request_method or "").strip().upper()[:16] or None details_json = json.dumps(details or {}, ensure_ascii=False) with self._lock: with self._connect() as conn: conn.execute( """ INSERT INTO logs ( level, category, event, message, username, backup_id, request_path, request_method, status_code, details_json ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?) """, ( lvl, safe_category, safe_event, safe_message, safe_username, backup_id, safe_path, safe_method, status_code, details_json, ), ) conn.commit() self._writes_since_prune += 1 if self._writes_since_prune >= 100: self._prune_if_needed() self._writes_since_prune = 0 def _prune_if_needed(self) -> None: with self._connect() as conn: row = conn.execute("SELECT COUNT(*) AS cnt FROM logs").fetchone() count = int(row["cnt"] if row is not None else 0) if count <= self.max_rows: return to_delete = count - self.max_rows conn.execute( """ DELETE FROM logs WHERE id IN ( SELECT id FROM logs ORDER BY created_at ASC, id ASC LIMIT ? ) """, (to_delete,), ) conn.commit() def query_logs( self, *, level: str | None = None, category: str | None = None, search: str | None = None, limit: int = 200, ) -> list[dict[str, Any]]: safe_limit = min(max(int(limit), 1), 1000) query = ( "SELECT id, created_at, level, category, event, message, username, " "backup_id, request_path, request_method, status_code, details_json " "FROM logs WHERE 1=1" ) params: list[Any] = [] if level: query += " AND level = ?" params.append(level.upper()) if category: query += " AND category = ?" params.append(category) if search: query += " AND (message LIKE ? OR event LIKE ? OR username LIKE ? OR details_json LIKE ?)" like = f"%{search.strip()}%" params.extend([like, like, like, like]) query += " ORDER BY created_at DESC, id DESC LIMIT ?" params.append(safe_limit) with self._connect() as conn: rows = conn.execute(query, params).fetchall() result: list[dict[str, Any]] = [] for row in rows: details_raw = row["details_json"] or "{}" try: details = json.loads(details_raw) except json.JSONDecodeError: details = {"raw": details_raw} result.append( { "id": row["id"], "created_at": row["created_at"], "level": row["level"], "category": row["category"], "event": row["event"], "message": row["message"], "username": row["username"], "backup_id": row["backup_id"], "request_path": row["request_path"], "request_method": row["request_method"], "status_code": row["status_code"], "details": details, } ) return result def list_categories(self) -> list[str]: with self._connect() as conn: rows = conn.execute("SELECT DISTINCT category FROM logs ORDER BY category ASC").fetchall() return [str(row["category"]) for row in rows if row["category"]] def count_by_level_last_hours(self, hours: int = 24) -> dict[str, int]: safe_hours = min(max(int(hours), 1), 24 * 30) with self._connect() as conn: rows = conn.execute( """ SELECT level, COUNT(*) AS cnt FROM logs WHERE created_at >= DATETIME('now', 'localtime', ?) GROUP BY level """, (f"-{safe_hours} hours",), ).fetchall() result = {"DEBUG": 0, "INFO": 0, "WARNING": 0, "ERROR": 0, "CRITICAL": 0} for row in rows: level = str(row["level"]) if level in result: result[level] = int(row["cnt"]) return result def clear(self) -> int: with self._connect() as conn: row = conn.execute("SELECT COUNT(*) AS cnt FROM logs").fetchone() count = int(row["cnt"] if row is not None else 0) conn.execute("DELETE FROM logs") conn.commit() return count @staticmethod def compact_details(details: dict[str, Any], max_len: int = 800) -> str: if not details: return "" raw = json.dumps(details, ensure_ascii=False, separators=(",", ":")) if len(raw) <= max_len: return raw return f"{raw[:max_len]}..." @staticmethod def now_iso() -> str: return datetime.now().strftime("%Y-%m-%d %H:%M:%S")