143 lines
4.5 KiB
Python
143 lines
4.5 KiB
Python
# Copyright (c) 2025 Patrick Motsch
|
|
# All rights reserved.
|
|
"""
|
|
Admin log viewer routes for the backend API.
|
|
Sysadmin-only endpoint for viewing gateway application logs.
|
|
Reads from the daily rotating log files (log_app_YYYYMMDD.log).
|
|
"""
|
|
|
|
import os
|
|
import glob
|
|
import logging
|
|
from datetime import datetime
|
|
from fastapi import APIRouter, HTTPException, Depends, Request, Query
|
|
from fastapi.responses import PlainTextResponse
|
|
from modules.auth import limiter, requireSysAdmin
|
|
from modules.shared.configuration import APP_CONFIG
|
|
from modules.datamodels.datamodelUam import User
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
router = APIRouter(
|
|
prefix="/api/admin/logs",
|
|
tags=["Admin Logs"],
|
|
responses={
|
|
401: {"description": "Unauthorized"},
|
|
403: {"description": "Forbidden - Sysadmin only"},
|
|
500: {"description": "Internal server error"},
|
|
},
|
|
)
|
|
|
|
|
|
def _getLogDir() -> str:
|
|
"""Resolve the configured log directory (same logic as app.py initLogging)."""
|
|
logDir = APP_CONFIG.get("APP_LOGGING_LOG_DIR", "./")
|
|
if not os.path.isabs(logDir):
|
|
gatewayDir = os.path.dirname(os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
|
|
logDir = os.path.join(gatewayDir, logDir)
|
|
return logDir
|
|
|
|
|
|
def _getLogFiles() -> list[str]:
|
|
"""Return all log_app_*.log files sorted by name descending (newest first)."""
|
|
logDir = _getLogDir()
|
|
pattern = os.path.join(logDir, "log_app_*.log")
|
|
files = glob.glob(pattern)
|
|
files.sort(reverse=True)
|
|
return files
|
|
|
|
|
|
def _readLastNLines(filePath: str, n: int) -> list[str]:
|
|
"""Read last n lines from a file efficiently."""
|
|
lines = []
|
|
try:
|
|
with open(filePath, "r", encoding="utf-8", errors="replace") as f:
|
|
lines = f.readlines()
|
|
except Exception as e:
|
|
logger.warning(f"Failed to read log file {filePath}: {e}")
|
|
return lines[-n:] if len(lines) > n else lines
|
|
|
|
|
|
@router.get("")
|
|
@limiter.limit("30/minute")
|
|
def getLogEntries(
|
|
request: Request,
|
|
count: int = Query(default=200, ge=1, le=50000, description="Number of log entries to return"),
|
|
currentUser: User = Depends(requireSysAdmin),
|
|
) -> dict:
|
|
"""
|
|
Get the last N log entries from the gateway log files.
|
|
Merges across multiple daily log files if needed.
|
|
"""
|
|
try:
|
|
logFiles = _getLogFiles()
|
|
|
|
if not logFiles:
|
|
return {"lines": [], "totalCollected": 0, "logDir": _getLogDir()}
|
|
|
|
collectedLines: list[str] = []
|
|
remaining = count
|
|
|
|
for filePath in logFiles:
|
|
if remaining <= 0:
|
|
break
|
|
|
|
lines = _readLastNLines(filePath, remaining)
|
|
collectedLines = lines + collectedLines
|
|
remaining = count - len(collectedLines)
|
|
|
|
finalLines = collectedLines[-count:] if len(collectedLines) > count else collectedLines
|
|
|
|
return {
|
|
"lines": [line.rstrip("\n").rstrip("\r") for line in finalLines],
|
|
"totalCollected": len(finalLines),
|
|
"logDir": _getLogDir(),
|
|
}
|
|
|
|
except Exception as e:
|
|
logger.error(f"Failed to read log files: {e}")
|
|
raise HTTPException(status_code=500, detail=f"Failed to read log files: {str(e)}")
|
|
|
|
|
|
@router.get("/download")
|
|
@limiter.limit("10/minute")
|
|
def downloadLog(
|
|
request: Request,
|
|
count: int = Query(default=1000, ge=1, le=100000, description="Number of log entries to download"),
|
|
currentUser: User = Depends(requireSysAdmin),
|
|
) -> PlainTextResponse:
|
|
"""
|
|
Download the last N log entries as a plain text file.
|
|
"""
|
|
try:
|
|
logFiles = _getLogFiles()
|
|
|
|
if not logFiles:
|
|
return PlainTextResponse("No log files found.", media_type="text/plain")
|
|
|
|
collectedLines: list[str] = []
|
|
remaining = count
|
|
|
|
for filePath in logFiles:
|
|
if remaining <= 0:
|
|
break
|
|
|
|
lines = _readLastNLines(filePath, remaining)
|
|
collectedLines = lines + collectedLines
|
|
remaining = count - len(collectedLines)
|
|
|
|
finalLines = collectedLines[-count:] if len(collectedLines) > count else collectedLines
|
|
content = "".join(finalLines)
|
|
timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
|
|
|
|
return PlainTextResponse(
|
|
content=content,
|
|
media_type="text/plain",
|
|
headers={
|
|
"Content-Disposition": f'attachment; filename="gateway_log_{timestamp}.log"',
|
|
},
|
|
)
|
|
|
|
except Exception as e:
|
|
logger.error(f"Failed to download log files: {e}")
|
|
raise HTTPException(status_code=500, detail=f"Failed to download log files: {str(e)}")
|