live-hub-py/utils/http_logger.py

"""HTTP 请求/响应详情日志记录器

将关键 HTTP 请求的完整详情（method、url、headers、body、status、response、耗时）
以 JSON Lines 格式写入日志文件，便于在 Web 界面查看和排查问题。
"""

import json
import threading
import time
from datetime import datetime
from pathlib import Path
from typing import Any, Optional

from loguru import logger

# 日志文件路径
_LOG_DIR = Path("logs")
_LOG_FILE = _LOG_DIR / "http_detail.jsonl"
_MAX_BODY_LEN = 2000  # 单个 body 最大记录长度，避免过大

# 线程安全写锁
_write_lock = threading.Lock()

# 确保日志目录存在
_LOG_DIR.mkdir(parents=True, exist_ok=True)


def _truncate(text: Any, max_len: int = _MAX_BODY_LEN) -> str:
    """截断过长的文本"""
    if text is None:
        return ""
    s = text if isinstance(text, str) else str(text)
    if len(s) > max_len:
        return s[:max_len] + f"...[truncated {len(s) - max_len} chars]"
    return s


def _safe_headers(headers: Any) -> dict:
    """清理 headers 中的敏感信息"""
    if not headers:
        return {}
    if hasattr(headers, 'items'):
        headers = dict(headers)
    safe = {}
    sensitive = {'authorization', 'cookie', 'set-cookie', 'password'}
    for k, v in headers.items():
        if k.lower() in sensitive:
            safe[k] = '***'
        else:
            safe[k] = v
    return safe


def log_http(
    category: str,
    method: str,
    url: str,
    *,
    request_headers: Any = None,
    request_body: Any = None,
    status_code: Optional[int] = None,
    response_headers: Any = None,
    response_body: Any = None,
    duration: Optional[float] = None,
    error: Optional[str] = None,
    proxy: Optional[str] = None,
    tag: str = "",
) -> None:
    """
    记录一条 HTTP 请求/响应详情日志。

    Args:
        category: 分类（如 'douyu_login', 'geetest', 'proxy_verify', 'whitelist'）
        method: HTTP 方法
        url: 请求 URL（会被脱敏，移除敏感查询参数）
        request_headers: 请求头
        request_body: 请求体
        status_code: 响应状态码
        response_headers: 响应头
        response_body: 响应体
        duration: 耗时（秒）
        error: 错误信息
        proxy: 使用的代理
        tag: 额外标签（如账号名）
    """
    entry = {
        "timestamp": datetime.now().isoformat(timespec="milliseconds"),
        "ts": time.time(),
        "category": category,
        "tag": tag,
        "method": method.upper(),
        "url": _truncate(url, 500),
        "proxy": proxy,
        "request": {
            "headers": _safe_headers(request_headers),
            "body": _truncate(request_body),
        },
        "response": {
            "status_code": status_code,
            "headers": _safe_headers(response_headers),
            "body": _truncate(response_body),
        },
        "duration_ms": round(duration * 1000, 1) if duration is not None else None,
        "error": _truncate(error, 500) if error else None,
    }

    # 判断级别
    if error or (status_code and status_code >= 400):
        entry["level"] = "error"
    elif status_code and status_code >= 300:
        entry["level"] = "warning"
    else:
        entry["level"] = "info"

    try:
        line = json.dumps(entry, ensure_ascii=False)
        with _write_lock:
            with open(_LOG_FILE, "a", encoding="utf-8") as f:
                f.write(line + "\n")
    except Exception as e:
        logger.debug(f"写入HTTP详情日志失败: {e}")


def read_http_logs(
    *,
    limit: int = 100,
    offset: int = 0,
    category: Optional[str] = None,
    level: Optional[str] = None,
    keyword: Optional[str] = None,
) -> tuple[list[dict], int]:
    """
    读取 HTTP 详情日志，支持筛选和分页。

    Args:
        limit: 返回条数上限
        offset: 偏移量（从最新往前数）
        category: 按分类筛选
        level: 按级别筛选（info/warning/error）
        keyword: 关键词搜索（url、tag、error）

    Returns:
        (日志条目列表, 总匹配条数)，列表按时间倒序（最新在前）
    """
    if not _LOG_FILE.exists():
        return [], 0

    entries: list[dict] = []
    try:
        with open(_LOG_FILE, "r", encoding="utf-8") as f:
            for line in f:
                line = line.strip()
                if not line:
                    continue
                try:
                    entry = json.loads(line)
                except json.JSONDecodeError:
                    continue

                # 筛选
                if category and entry.get("category") != category:
                    continue
                if level and entry.get("level") != level:
                    continue
                if keyword:
                    kw = keyword.lower()
                    searchable = " ".join([
                        str(entry.get("url", "")),
                        str(entry.get("tag", "")),
                        str(entry.get("error", "")),
                        str(entry.get("method", "")),
                    ]).lower()
                    if kw not in searchable:
                        continue

                entries.append(entry)
    except Exception as e:
        logger.error(f"读取HTTP详情日志失败: {e}")
        return [], 0

    # 按时间倒序
    entries.sort(key=lambda x: x.get("ts", 0), reverse=True)
    total = len(entries)
    # 分页（offset 从最新开始算）
    page = entries[offset:offset + limit]
    return page, total


def clear_http_logs() -> int:
    """清空 HTTP 详情日志，返回清空的条数"""
    count = 0
    with _write_lock:
        if _LOG_FILE.exists():
            try:
                with open(_LOG_FILE, "r", encoding="utf-8") as f:
                    count = sum(1 for line in f if line.strip())
            except Exception:
                pass
            _LOG_FILE.write_text("", encoding="utf-8")
    return count