MoFin/xmpp_logger.py

#!/usr/bin/env python3
"""xmpp_logger.py — XMPP 消息日志采集器

记录所有 XMPP 通信事件到 JSONL 日志文件。通过 hook 方式接入：
只需在消息发送点加一行 log_xmpp() 调用，不动业务逻辑。

日志文件: gateway/logs/xmpp_messages.jsonl
自动轮转: 保留最近 7 天
"""
import json
import time as _time
from pathlib import Path
from datetime import datetime, timedelta

LOG_DIR = Path(__file__).resolve().parent / "gateway" / "logs"
LOG_FILE = LOG_DIR / "xmpp_messages.jsonl"
MAX_AGE_DAYS = 7


def log_xmpp(direction, from_jid, to_jid, body, status="ok", error=None, latency_ms=0):
    """记录一条 XMPP 消息事件。

    Args:
        direction: "out"(发送) 或 "in"(接收)
        from_jid: 发送者 JID
        to_jid: 接收者 JID
        body: 消息体（自动截取前 200 字预览）
        status: "ok" / "error" / "timeout"
        error: 错误信息（仅 status!="ok" 时）
        latency_ms: 延迟（毫秒）
    """
    LOG_DIR.mkdir(parents=True, exist_ok=True)

    entry = {
        "timestamp": datetime.now().strftime("%Y-%m-%d %H:%M:%S"),
        "direction": direction,
        "from": from_jid,
        "to": to_jid,
        "body_preview": (body or "")[:200].replace("\n", " "),
        "status": status,
        "error": str(error)[:200] if error else None,
        "latency_ms": latency_ms,
        "epoch": _time.time(),
    }

    with open(LOG_FILE, "a", encoding="utf-8") as f:
        f.write(json.dumps(entry, ensure_ascii=False) + "\n")

    # 每次写入后检查是否需要轮转（采样：每 20 条轮转一次）
    if LOG_FILE.stat().st_size > 500 * 1024:  # >500KB
        _rotate()


def _rotate():
    """保留最近 MAX_AGE_DAYS 天，丢弃旧日志"""
    if not LOG_FILE.exists():
        return
    cutoff = datetime.now() - timedelta(days=MAX_AGE_DAYS)
    kept = []
    try:
        with open(LOG_FILE, "r", encoding="utf-8") as f:
            for line in f:
                try:
                    e = json.loads(line)
                    if e["timestamp"][:10] >= cutoff.strftime("%Y-%m-%d"):
                        kept.append(line)
                except Exception:
                    continue
        with open(LOG_FILE, "w", encoding="utf-8") as f:
            f.writelines(kept)
    except Exception:
        pass


def query(since=None, agent=None, status=None, limit=50):
    """查询消息日志

    Args:
        since: ISO datetime string, 只返回此时间之后的消息
        agent: JID 片段，筛选发送或接收方包含此字符串的消息
        status: 筛选状态 "ok"/"error"/"timeout"
        limit: 最大返回条数（默认 50）
    """
    if not LOG_FILE.exists():
        return []
    results = []
    try:
        with open(LOG_FILE, "r", encoding="utf-8") as f:
            for line in f:
                try:
                    e = json.loads(line)
                    if since and e["timestamp"] < since:
                        continue
                    if agent and agent not in e["from"] and agent not in e["to"]:
                        continue
                    if status and e["status"] != status:
                        continue
                    results.append(e)
                except Exception:
                    continue
    except Exception:
        pass
    results.sort(key=lambda x: x["timestamp"], reverse=True)
    return results[:limit]


def stats():
    """获取消息统计：今日 + 本周"""
    if not LOG_FILE.exists():
        return {"today": {"sent": 0, "failed": 0, "latency_avg": 0}, "week": {"sent": 0, "failed": 0}}

    now = datetime.now()
    today = now.strftime("%Y-%m-%d")
    week_ago = (now - timedelta(days=7)).strftime("%Y-%m-%d")

    latencies = []
    s = {"today": {"sent": 0, "failed": 0}, "week": {"sent": 0, "failed": 0}}

    try:
        with open(LOG_FILE, "r", encoding="utf-8") as f:
            for line in f:
                try:
                    e = json.loads(line)
                    d = e["timestamp"][:10]
                    if d >= week_ago:
                        s["week"]["sent"] += 1
                        if e["status"] != "ok":
                            s["week"]["failed"] += 1
                    if d == today:
                        s["today"]["sent"] += 1
                        if e["status"] != "ok":
                            s["today"]["failed"] += 1
                        if e.get("latency_ms"):
                            latencies.append(e["latency_ms"])
                except Exception:
                    continue
    except Exception:
        pass

    s["today"]["latency_avg"] = round(sum(latencies) / len(latencies)) if latencies else 0
    return s


def health():
    """快速健康检查：返回最后消息年龄 + 最近1h错误率"""
    if not LOG_FILE.exists():
        return {"status": "no_data", "last_message_age_sec": -1, "error_rate_1h": 0}

    now_epoch = _time.time()
    last_epoch = 0
    recent_total = 0
    recent_errors = 0
    one_hour_ago = now_epoch - 3600

    try:
        with open(LOG_FILE, "r", encoding="utf-8") as f:
            for line in f:
                try:
                    e = json.loads(line)
                    ep = e.get("epoch", 0)
                    if ep > last_epoch:
                        last_epoch = ep
                    if ep > one_hour_ago:
                        recent_total += 1
                        if e["status"] != "ok":
                            recent_errors += 1
                except Exception:
                    continue
    except Exception:
        pass

    last_age = int(now_epoch - last_epoch) if last_epoch else -1
    err_rate = round(recent_errors / recent_total * 100, 1) if recent_total else 0

    # 状态判断
    if last_age < 0:
        st = "no_data"
    elif last_age > 600:  # >10min no message
        st = "critical"
    elif err_rate > 50:
        st = "degraded"
    else:
        st = "ok"

    return {"status": st, "last_message_age_sec": last_age, "error_rate_1h": err_rate}