dekun
90 lines
2.6 KiB
Python
90 lines
2.6 KiB
Python
"""中控:读取 PM2 进程 stdout/stderr 日志(系统日志页)."""
|
|
from __future__ import annotations
|
|
|
|
import os
|
|
import time
|
|
from pathlib import Path
|
|
from typing import Any
|
|
|
|
LOG_TARGETS: dict[str, dict[str, str]] = {
|
|
"binance": {"label": "币安", "pm2_name": "crypto_binance"},
|
|
"gate": {"label": "Gate", "pm2_name": "crypto_gate"},
|
|
"okx": {"label": "OKX", "pm2_name": "crypto_okx"},
|
|
"hub": {"label": "中控", "pm2_name": "manual-trading-hub"},
|
|
}
|
|
|
|
DEFAULT_LINES = 200
|
|
MAX_LINES = 500
|
|
DEFAULT_TAIL_BYTES = 400_000
|
|
|
|
|
|
def pm2_logs_dir() -> Path:
|
|
raw = (os.getenv("PM2_HOME") or "").strip()
|
|
base = Path(raw) if raw else Path.home() / ".pm2"
|
|
return base / "logs"
|
|
|
|
|
|
def log_file_paths(pm2_name: str) -> tuple[Path, Path]:
|
|
logs_dir = pm2_logs_dir()
|
|
return logs_dir / f"{pm2_name}-out.log", logs_dir / f"{pm2_name}-error.log"
|
|
|
|
|
|
def tail_lines(
|
|
path: Path,
|
|
lines: int = DEFAULT_LINES,
|
|
*,
|
|
max_bytes: int = DEFAULT_TAIL_BYTES,
|
|
) -> str:
|
|
if not path.is_file():
|
|
return ""
|
|
try:
|
|
size = path.stat().st_size
|
|
with path.open("rb") as handle:
|
|
if size <= max_bytes:
|
|
data = handle.read()
|
|
else:
|
|
handle.seek(max(0, size - max_bytes))
|
|
data = handle.read()
|
|
text = data.decode("utf-8", errors="replace")
|
|
parts = text.splitlines()
|
|
if len(parts) > lines:
|
|
parts = parts[-lines:]
|
|
return "\n".join(parts)
|
|
except OSError:
|
|
return ""
|
|
|
|
|
|
def system_logs_meta() -> dict[str, Any]:
|
|
return {
|
|
"ok": True,
|
|
"targets": [
|
|
{"key": key, "label": cfg["label"], "pm2_name": cfg["pm2_name"]}
|
|
for key, cfg in LOG_TARGETS.items()
|
|
],
|
|
"default_lines": DEFAULT_LINES,
|
|
"max_lines": MAX_LINES,
|
|
}
|
|
|
|
|
|
def load_system_logs(target: str, lines: int = DEFAULT_LINES) -> dict[str, Any]:
|
|
key = (target or "").strip().lower()
|
|
if key not in LOG_TARGETS:
|
|
raise KeyError(key)
|
|
cfg = LOG_TARGETS[key]
|
|
line_count = max(20, min(MAX_LINES, int(lines or DEFAULT_LINES)))
|
|
out_path, err_path = log_file_paths(cfg["pm2_name"])
|
|
return {
|
|
"ok": True,
|
|
"key": key,
|
|
"label": cfg["label"],
|
|
"pm2_name": cfg["pm2_name"],
|
|
"lines": line_count,
|
|
"out": tail_lines(out_path, line_count),
|
|
"err": tail_lines(err_path, line_count),
|
|
"out_exists": out_path.is_file(),
|
|
"err_exists": err_path.is_file(),
|
|
"out_path": str(out_path),
|
|
"err_path": str(err_path),
|
|
"updated_at": int(time.time()),
|
|
}
|