project/tbsg/commons/err_log.py

# -*- coding: utf-8 -*-
"""
异常日志 — 按设备分文件记录采集过程异常，JSONL 格式（每行一个JSON对象）。

文件位置: logs/errors/{设备号}/{设备号}_YYYYMMDD.jsonl
字段: ts=本地时间, device=设备号, source=发生点标识, type=异常类名或event,
      message=错误消息, traceback=堆栈(有异常对象时), extra=附加上下文dict
用法:
    from commons import err_log
    err_log.log_error(ex.device_id, "task_exception", exc=e, extra={"task_id": tid})
设计: 记录失败静默吞掉，绝不影响采集主流程；进程内加锁，多进程各写各的设备文件互不冲突。
"""
import json
import threading
import traceback
from datetime import datetime
from pathlib import Path

_BASE = Path(__file__).resolve().parent.parent / "logs" / "errors"
_LOCK = threading.Lock()


def log_error(device_id: str, source: str, exc=None, message: str = "", extra: dict = None):
    """记录一条异常/错误事件。传 exc=异常对象(自动带类型+堆栈)，或仅 message 描述事件。"""
    try:
        rec = {
            "ts": datetime.now().strftime("%Y-%m-%d %H:%M:%S"),
            "device": str(device_id or "unknown"),
            "source": source,
            "type": type(exc).__name__ if exc else "event",
            "message": str(exc) if exc else str(message),
        }
        if exc is not None:
            tb = traceback.format_exc(limit=8)
            if tb and "NoneType: None" not in tb:
                rec["traceback"] = tb
        if extra:
            rec["extra"] = extra
        d = _BASE / rec["device"]
        d.mkdir(parents=True, exist_ok=True)
        path = d / f"{rec['device']}_{datetime.now().strftime('%Y%m%d')}.jsonl"
        with _LOCK:
            with open(path, "a", encoding="utf-8") as f:
                f.write(json.dumps(rec, ensure_ascii=False) + "\n")
    except Exception:
        pass  # 日志失败不影响主流程


def log_pressure(device_id: str, data: dict):
    """账号压力日志: 每任务结束追加一条累计统计(crawled/tasks/captcha/runtime)，
    与 account_kicked 事件对照 → 分析风控封号的压力阈值。
    文件: logs/pressure/{设备}_{日期}.jsonl"""
    try:
        rec = {"ts": datetime.now().strftime("%Y-%m-%d %H:%M:%S"),
               "device": str(device_id or "unknown")}
        rec.update(data or {})
        d = _BASE.parent / "pressure" / rec["device"]
        d.mkdir(parents=True, exist_ok=True)
        path = d / f"{rec['device']}_{datetime.now().strftime('%Y%m%d')}.jsonl"
        with _LOCK:
            with open(path, "a", encoding="utf-8") as f:
                f.write(json.dumps(rec, ensure_ascii=False) + "\n")
    except Exception:
        pass