"""Logging API endpoints for AniWorld. Provides endpoints for reading log configuration, listing log files, tailing/downloading individual log files, testing logging, and cleanup. """ from __future__ import annotations import logging import os from pathlib import Path from typing import Any, Dict, List, Optional from fastapi import APIRouter, Depends, HTTPException, status from fastapi.responses import FileResponse from src.server.services.config_service import get_config_service from src.server.utils.dependencies import require_auth logger = logging.getLogger(__name__) router = APIRouter(prefix="/api/logging", tags=["logging"]) _LOG_DIR = Path("logs") # --------------------------------------------------------------------------- # Helpers # --------------------------------------------------------------------------- def _log_dir() -> Path: """Return the log directory, creating it if necessary.""" _LOG_DIR.mkdir(exist_ok=True) return _LOG_DIR def _list_log_files() -> List[Dict[str, Any]]: """Return metadata for all .log files in the log directory.""" result: List[Dict[str, Any]] = [] log_dir = _log_dir() for entry in sorted(log_dir.iterdir()): if entry.is_file() and entry.suffix in {".log", ".txt"}: stat = entry.stat() result.append( { "name": entry.name, "size_mb": round(stat.st_size / (1024 * 1024), 2), "modified": stat.st_mtime, } ) return result # --------------------------------------------------------------------------- # Endpoints # --------------------------------------------------------------------------- @router.get("/config") def get_logging_config( auth: Optional[dict] = Depends(require_auth), ) -> Dict[str, Any]: """Return current logging configuration as used by the frontend. Maps the internal ``LoggingConfig`` model fields to the shape expected by ``logging-config.js``. """ try: config_service = get_config_service() app_config = config_service.load_config() lc = app_config.logging return { "success": True, "config": { # Primary fields (match the model) "log_level": lc.level, "log_file": lc.file, "max_bytes": lc.max_bytes, "backup_count": lc.backup_count, # UI-only flags – defaults; not yet persisted in the model "enable_console_logging": True, "enable_console_progress": False, "enable_fail2ban_logging": False, }, } except Exception as exc: logger.exception("Failed to read logging config") raise HTTPException( status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, detail=f"Failed to read logging config: {exc}", ) from exc @router.get("/files") def list_files( auth: Optional[dict] = Depends(require_auth), ) -> Dict[str, Any]: """List all available log files with metadata.""" try: return {"success": True, "files": _list_log_files()} except Exception as exc: logger.exception("Failed to list log files") raise HTTPException( status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, detail=f"Failed to list log files: {exc}", ) from exc @router.get("/files/{filename}/tail") def tail_file( filename: str, lines: int = 100, auth: Optional[dict] = Depends(require_auth), ) -> Dict[str, Any]: """Return the last *lines* lines of a log file. Args: filename: Name of the log file (no path traversal). lines: Number of lines to return (default 100). Returns: Dict with ``success``, ``lines``, ``showing_lines``, ``total_lines``. """ # Prevent path traversal safe_name = Path(filename).name file_path = _log_dir() / safe_name if not file_path.exists(): raise HTTPException( status_code=status.HTTP_404_NOT_FOUND, detail=f"Log file not found: {safe_name}", ) try: all_lines = file_path.read_text(encoding="utf-8", errors="replace").splitlines() tail = all_lines[-lines:] if len(all_lines) > lines else all_lines return { "success": True, "lines": tail, "showing_lines": len(tail), "total_lines": len(all_lines), } except Exception as exc: logger.exception("Failed to tail log file %s", safe_name) raise HTTPException( status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, detail=f"Failed to read log file: {exc}", ) from exc @router.get("/files/{filename}/download") def download_file( filename: str, auth: Optional[dict] = Depends(require_auth), ) -> FileResponse: """Download a log file as an attachment.""" safe_name = Path(filename).name file_path = _log_dir() / safe_name if not file_path.exists(): raise HTTPException( status_code=status.HTTP_404_NOT_FOUND, detail=f"Log file not found: {safe_name}", ) return FileResponse( path=str(file_path), filename=safe_name, media_type="text/plain", ) @router.post("/test") def test_logging( auth: dict = Depends(require_auth), ) -> Dict[str, Any]: """Write test log messages at all levels.""" logging.getLogger("aniworld.test").debug("Test DEBUG message") logging.getLogger("aniworld.test").info("Test INFO message") logging.getLogger("aniworld.test").warning("Test WARNING message") logging.getLogger("aniworld.test").error("Test ERROR message") return {"success": True, "message": "Test messages written to log"} @router.post("/cleanup") def cleanup_logs( payload: Dict[str, Any], auth: dict = Depends(require_auth), ) -> Dict[str, Any]: """Delete log files older than *days* days. Args: payload: JSON body with ``days`` (int) field. Returns: Dict with ``success`` and ``message`` describing what was deleted. """ import time days = payload.get("days", 30) try: days = int(days) if days < 1: raise ValueError("days must be >= 1") except (TypeError, ValueError) as exc: raise HTTPException( status_code=status.HTTP_400_BAD_REQUEST, detail=f"Invalid days value: {exc}", ) from exc cutoff = time.time() - days * 86400 removed: List[str] = [] errors: List[str] = [] for entry in _log_dir().iterdir(): if entry.is_file() and entry.suffix in {".log", ".txt"}: if entry.stat().st_mtime < cutoff: try: entry.unlink() removed.append(entry.name) except OSError as exc: errors.append(f"{entry.name}: {exc}") message = f"Removed {len(removed)} file(s) older than {days} days." if errors: message += f" Errors: {'; '.join(errors)}" logger.info( "Log cleanup by %s: removed=%s days=%s", auth.get("username", "unknown"), removed, days, ) return {"success": True, "message": message, "removed": removed}