53 lines
1.9 KiB
Python
53 lines
1.9 KiB
Python
from __future__ import annotations
|
|
|
|
from pathlib import Path
|
|
|
|
|
|
class LogReader:
|
|
def __init__(self, root_dir: Path | None = None):
|
|
self.root_dir = (root_dir or Path(__file__).resolve().parents[3]).resolve()
|
|
self.log_dirs = [
|
|
self.root_dir / "logs",
|
|
self.root_dir / "runtime" / "logs",
|
|
self.root_dir / "data" / "workspace" / "logs",
|
|
]
|
|
|
|
def _allowed_log_files(self) -> dict[str, Path]:
|
|
items: dict[str, Path] = {}
|
|
for log_dir in self.log_dirs:
|
|
if not log_dir.exists():
|
|
continue
|
|
for path in sorted(p for p in log_dir.rglob("*.log") if p.is_file()):
|
|
items.setdefault(path.name, path.resolve())
|
|
return items
|
|
|
|
def list_logs(self) -> dict[str, object]:
|
|
allowed_log_files = self._allowed_log_files()
|
|
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]:
|
|
allowed_log_files = self._allowed_log_files()
|
|
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)):]),
|
|
}
|