Files
biliup-next/src/biliup_next/infra/log_reader.py
2026-04-01 00:44:58 +08:00

43 lines
1.6 KiB
Python

from __future__ import annotations
from pathlib import Path
ALLOWED_LOG_FILES = {
"monitor.log": Path("/home/theshy/biliup/logs/system/monitor.log"),
"monitorSrt.log": Path("/home/theshy/biliup/logs/system/monitorSrt.log"),
"monitorSongs.log": Path("/home/theshy/biliup/logs/system/monitorSongs.log"),
"upload.log": Path("/home/theshy/biliup/logs/system/upload.log"),
"session_top_comment.py.log": Path("/home/theshy/biliup/logs/system/session_top_comment.py.log"),
"add_to_collection.py.log": Path("/home/theshy/biliup/logs/system/add_to_collection.py.log"),
}
class LogReader:
def list_logs(self) -> dict[str, object]:
return {
"items": [
{
"name": name,
"path": str(path),
"exists": path.exists(),
}
for name, path in sorted(ALLOWED_LOG_FILES.items())
]
}
def tail(self, name: str, lines: int = 200, contains: str | None = None) -> dict[str, object]:
if name not in ALLOWED_LOG_FILES:
raise ValueError(f"unsupported log: {name}")
path = ALLOWED_LOG_FILES[name]
if not path.exists():
return {"name": name, "path": str(path), "exists": False, "content": ""}
content = path.read_text(encoding="utf-8", errors="replace").splitlines()
if contains:
content = [line for line in content if contains in line]
return {
"name": name,
"path": str(path),
"exists": True,
"content": "\n".join(content[-max(1, min(lines, 1000)):]),
}