68 lines
2.2 KiB
Python
68 lines
2.2 KiB
Python
from __future__ import annotations
|
|
|
|
import os
|
|
from datetime import datetime
|
|
from typing import List, Optional, Tuple
|
|
|
|
|
|
class LogsService:
|
|
def __init__(self, log_dir: str) -> None:
|
|
self.log_dir = log_dir
|
|
|
|
def _parse_time(self, line: str) -> Optional[datetime]:
|
|
try:
|
|
prefix = line.split("|", 1)[0].strip()
|
|
return datetime.strptime(prefix, "%Y-%m-%d %H:%M:%S")
|
|
except Exception:
|
|
return None
|
|
|
|
def _list_log_files(self) -> List[str]:
|
|
if not os.path.isdir(self.log_dir):
|
|
return []
|
|
files = [
|
|
os.path.join(self.log_dir, name)
|
|
for name in os.listdir(self.log_dir)
|
|
if name.startswith(("app.log", "access.log", "error.log"))
|
|
]
|
|
return sorted(files)
|
|
|
|
def query(
|
|
self,
|
|
keyword: str = "",
|
|
start: Optional[str] = None,
|
|
end: Optional[str] = None,
|
|
page: int = 1,
|
|
size: int = 50,
|
|
) -> Tuple[int, List[str]]:
|
|
start_dt = datetime.fromisoformat(start) if start else None
|
|
end_dt = datetime.fromisoformat(end) if end else None
|
|
|
|
matched: List[Tuple[datetime, str]] = []
|
|
for path in self._list_log_files():
|
|
try:
|
|
with open(path, "r", encoding="utf-8") as f:
|
|
for line in f:
|
|
line = line.rstrip("\n")
|
|
if "path=/logs/query" in line or "path=/logs/" in line:
|
|
continue
|
|
if keyword and keyword not in line:
|
|
continue
|
|
ts = self._parse_time(line)
|
|
if start_dt and ts and ts < start_dt:
|
|
continue
|
|
if end_dt and ts and ts > end_dt:
|
|
continue
|
|
matched.append((ts or datetime.min, line))
|
|
except Exception:
|
|
continue
|
|
|
|
matched.sort(key=lambda item: item[0], reverse=True)
|
|
total = len(matched)
|
|
if size <= 0:
|
|
size = 50
|
|
if page <= 0:
|
|
page = 1
|
|
start_idx = (page - 1) * size
|
|
end_idx = start_idx + size
|
|
return total, [line for _, line in matched[start_idx:end_idx]]
|