pipeline-llm/pipeline_llm/trace.py

"""pipeline_llm.trace — 模型调用上行/下行原文追踪（2026-09-10 用户定夺）。

存储结构（唯一收敛点：所有路径的上游 HTTP 往返都经 inference._post_upstream /
_http_request，在这一层各记一次，全链唯一）：
  文件：filesroot/llm_trace/<YYYYMMDD>/<call_id>_<seq>.req.json / .resp.json
  库：  llm_call_trace 一行一次上游往返（重试/异步轮询的中间轮次都落，1:N）

call_id 关联链：chat_inference 入口生成 → ctx['call_id'] → 每次往返落 trace 行
→ govern_settle 写 llm_usage.call_id（同值）。按 usage 流水即可反查全部原文。

铁律：
  1. 脱敏后落盘——Authorization/api-key 等敏感 header 值替换为 [REDACTED]
     （真 key 不出进程；文件不得成为密钥泄露面）。URL 去 query（可能带签名）。
  2. 先写文件、后写库行（行含文件路径）；任何失败只记日志绝不阻断调用——
     审计尽力而为，禁止反噬计费/调用链。record_roundtrip 永不抛异常。
  3. 保留期：TRACE_RETENTION_DAYS 天，cleanup_expired 删过期日期目录+库行
     （记账 worker 循环每日调一次）。文件须纳入部署备份（dbackup 只备库，
     filesroot/llm_trace 需另行同步备份，否则"唯一存储"不成立）。
  4. 下载走专用端点 wwwroot/api/llm_trace_file.dspy（登录 + 机构隔离 +
     路径必须落在 llm_trace 目录内）。禁止复用 /idfile——那是登录即可的
     全平台通用口，会把跨机构会话原文泄露给任意登录用户。
"""

import json
import logging
import os
import shutil
from datetime import datetime, timedelta

logger = logging.getLogger("pipeline_llm.trace")

_TRACE_DIR = 'llm_trace'
TRACE_RETENTION_DAYS = 90

# 敏感 header 关键词（子串匹配，全小写比较）：命中即值替换 [REDACTED]
_SENSITIVE_HEADER_KEYS = (
    'authorization', 'api-key', 'apikey', 'x-api-key', 'proxy-authorization',
    'cookie', 'x-goog-api-key', 'x-auth-token', 'x-dashscope-apikey',
    'x-aliyun-api-key', 'token',
)


def _filesroot():
    from appPublic.jsonConfig import getConfig
    root = None
    try:
        root = getConfig().filesroot
    except Exception:
        root = None
    return os.path.abspath(root or __import__('tempfile').gettempdir())


def trace_root():
    """llm_trace 目录绝对路径（下载端点做路径穿越校验用）。"""
    return os.path.join(_filesroot(), _TRACE_DIR)


def _redact_headers(headers):
    out = {}
    for k, v in (headers or {}).items():
        lk = str(k).lower()
        if any(s in lk for s in _SENSITIVE_HEADER_KEYS):
            out[k] = '[REDACTED]'
        else:
            out[k] = v
    return out


def _dump(obj):
    try:
        return json.dumps(obj, ensure_ascii=False, default=str, indent=None)
    except Exception:
        return str(obj)[:65535]


async def _write_file(rel, content):
    import aiofiles
    fp = os.path.join(_filesroot(), rel)
    os.makedirs(os.path.dirname(fp), exist_ok=True)
    async with aiofiles.open(fp, 'w', encoding='utf-8') as f:
        await f.write(content)


async def record_roundtrip(ctx, seq, kind, url, method, headers, req_body,
                           status_code, resp_data, elapsed_ms, err=''):
    """一次上游 HTTP 往返 → 请求/响应两个文件 + 一行 llm_call_trace。

    ctx       govern_resolve 返回的上下文（须含 call_id；缺 call_id 直接跳过——
              没有关联键的孤儿 trace 无意义）
    seq       往返序号（同步重试 0..2；异步提交 0，轮询 1..N）
    kind      chat / gen（同步生成） / submit / query
    resp_data 响应体（dict/str 均可；网络异常时 None）
    err       错误摘要（网络异常/非200 等）

    永不抛出（铁律 2）。
    """
    try:
        ctx = ctx or {}
        call_id = str(ctx.get('call_id') or '').strip()
        if not call_id:
            return
        day = datetime.now().strftime('%Y%m%d')
        base = '%s/%s/%s_%03d' % (_TRACE_DIR, day, call_id, int(seq or 0))
        # URL 去 query（可能带临时签名参数，属敏感信息）
        safe_url = str(url or '').split('?')[0][:500]
        req_path = ''
        resp_path = ''
        try:
            req_rel = base + '.req.json'
            await _write_file(req_rel, _dump({
                'url': safe_url, 'method': method, 'kind': kind,
                'headers': _redact_headers(headers),
                'body': req_body,
            }))
            req_path = req_rel
        except Exception as e:
            # 请求文件失败不阻断响应文件与库行（部分留痕优于无痕）
            print("[trace] 请求文件写入失败(call=%s seq=%s): %s" % (call_id, seq, e))
        try:
            resp_rel = base + '.resp.json'
            await _write_file(resp_rel, _dump({
                'status_code': int(status_code or 0),
                'elapsed_ms': int(elapsed_ms or 0),
                'error': err or '',
                'body': resp_data,
            }))
            resp_path = resp_rel
        except Exception as e:
            print("[trace] 响应文件写入失败(call=%s seq=%s): %s" % (call_id, seq, e))
        # 库行（先文件后行：行存在 ⇒ 文件已尝试写入；路径为空=该侧写失败）
        from appPublic.uniqueID import getID
        from .gateway import _get_db
        db, dbname = _get_db()
        async with db.sqlorContext(dbname) as sor:
            await sor.C('llm_call_trace', {
                'id': getID(),
                'call_id': call_id,
                'seq': int(seq or 0),
                'kind': str(kind or '')[:20],
                'url': safe_url,
                'method': str(method or '')[:10],
                'status_code': int(status_code or 0),
                'elapsed_ms': int(elapsed_ms or 0),
                'req_path': req_path,
                'resp_path': resp_path,
                'org_id': ctx.get('org_id') or '',
                'note': str(err or '')[:200],
            })
            await sor.sqlExe("COMMIT", {})
    except Exception as e:
        logger.warning("pipeline_llm.trace: trace 记录失败（不阻断调用）: %s", e)


async def cleanup_expired(days=None):
    """清理过期 trace：按日期目录整删文件 + 删对应库行。返回 (文件目录数, 库行数)。

    目录名即日期（YYYYMMDD），过期判定以目录名为准（不逐文件 stat）；
    库行按 created_at 删。目录删失败（权限/占用）不阻断库行清理。
    """
    days = int(days or TRACE_RETENTION_DAYS)
    cutoff_day = (datetime.now() - timedelta(days=days)).strftime('%Y%m%d')
    cutoff_ts = (datetime.now() - timedelta(days=days)).strftime('%Y-%m-%d %H:%M:%S')
    removed_dirs = 0
    root = trace_root()
    try:
        for name in sorted(os.listdir(root)):
            if not (name.isdigit() and len(name) == 8):
                continue
            if name < cutoff_day:
                try:
                    shutil.rmtree(os.path.join(root, name))
                    removed_dirs += 1
                except Exception as e:
                    logger.warning("pipeline_llm.trace: 清理目录 %s 失败: %s", name, e)
    except FileNotFoundError:
        pass
    removed_rows = 0
    try:
        from .gateway import _get_db
        db, dbname = _get_db()
        async with db.sqlorContext(dbname) as sor:
            await sor.sqlExe(
                "DELETE FROM llm_call_trace WHERE created_at<${c}$", {"c": cutoff_ts})
            await sor.sqlExe("COMMIT", {})
            recs = await sor.sqlExe("SELECT ROW_COUNT() AS n", {})
            removed_rows = int(getattr(recs[0], 'n', 0)) if recs else 0
    except Exception as e:
        logger.warning("pipeline_llm.trace: 清理库行失败: %s", e)
    if removed_dirs or removed_rows:
        logger.info("pipeline_llm.trace: 清理过期 trace 目录 %d 个、库行 %d 条（保留 %d 天）",
                    removed_dirs, removed_rows, days)
    return removed_dirs, removed_rows