crypto_monitor_user/lib/hub/hub_system_logs_lib.py

"""中控:读取 PM2 进程 stdout/stderr 日志(系统日志页)."""
from __future__ import annotations

import json
import os
import subprocess
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
_PATH_CACHE_TTL_SEC = 30

_path_cache: dict[str, tuple[Path, Path]] = {}
_path_cache_at = 0.0


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 _pm2_jlist() -> list[dict[str, Any]]:
    try:
        proc = subprocess.run(
            ["pm2", "jlist"],
            capture_output=True,
            text=True,
            timeout=8,
            check=False,
        )
        if proc.returncode != 0:
            return []
        data = json.loads(proc.stdout or "[]")
        return data if isinstance(data, list) else []
    except (OSError, subprocess.SubprocessError, json.JSONDecodeError, ValueError):
        return []


def _newest_match(pattern: str, logs_dir: Path) -> Path | None:
    matches = [p for p in logs_dir.glob(pattern) if p.is_file()]
    if not matches:
        return None
    return max(matches, key=lambda p: p.stat().st_mtime)


def _glob_log_paths(pm2_name: str) -> tuple[Path, Path]:
    logs_dir = pm2_logs_dir()
    out_path = Path()
    err_path = Path()
    if not logs_dir.is_dir():
        return out_path, err_path
    slugs = {pm2_name, pm2_name.replace("_", "-")}
    for slug in slugs:
        if not out_path.is_file():
            hit = _newest_match(f"{slug}-out*.log", logs_dir)
            if hit is not None:
                out_path = hit
        if not err_path.is_file():
            hit = _newest_match(f"{slug}-error*.log", logs_dir)
            if hit is not None:
                err_path = hit
    return out_path, err_path


def resolve_log_paths(pm2_name: str) -> tuple[Path, Path]:
    global _path_cache_at
    now = time.time()
    if now - _path_cache_at > _PATH_CACHE_TTL_SEC:
        _path_cache.clear()
        _path_cache_at = now
    cached = _path_cache.get(pm2_name)
    if cached is not None:
        return cached

    out_path = Path()
    err_path = Path()
    for proc in _pm2_jlist():
        if proc.get("name") != pm2_name:
            continue
        env = proc.get("pm2_env") or {}
        raw_out = (env.get("pm_out_log_path") or "").strip()
        raw_err = (env.get("pm_err_log_path") or "").strip()
        if raw_out:
            out_path = Path(raw_out)
        if raw_err:
            err_path = Path(raw_err)
        break

    glob_out, glob_err = _glob_log_paths(pm2_name)
    if not out_path.is_file() and glob_out.is_file():
        out_path = glob_out
    if not err_path.is_file() and glob_err.is_file():
        err_path = glob_err

    result = (out_path, err_path)
    _path_cache[pm2_name] = result
    return result


def log_file_paths(pm2_name: str) -> tuple[Path, Path]:
    return resolve_log_paths(pm2_name)


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 = resolve_log_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) if str(out_path) else "",
        "err_path": str(err_path) if str(err_path) else "",
        "updated_at": int(time.time()),
    }