attachment_classifier/order_logger.py

# -*- coding: utf-8 -*-
"""对话日志模块。

设计目标：每个总排号一个日志文件，纯文本格式，完整记录"这次调用到底发生了
什么"——system prompt、few-shot、实际发给模型的用户消息、模型每一次原始
回复（含被格式校验拒绝、触发重试的那些），以及最终的解析结果。成功失败都写，
不因为调用失败就少记东西；恰恰是失败的调用最需要完整日志去排查。

文件命名: {log_dir}/{总排号}_{时间戳}.log
同一个总排号如果被反复处理（比如手动重跑），不会覆盖旧日志，每次生成新文件，
方便对比"改了提示词前后，同一条记录的判断有没有变化"。
"""
from __future__ import annotations

import re
from datetime import datetime
from pathlib import Path

from llm_client import LLMCallResult
from parser import ParsedResult

# 总排号可能含有对文件名不友好的字符（如斜杠），做一次保守清理
_UNSAFE_FILENAME_CHARS = re.compile(r'[\\/:*?"<>|]')


def _safe_filename_part(zong_pai_hao: str) -> str:
    return _UNSAFE_FILENAME_CHARS.sub("_", zong_pai_hao)


class OrderLogger:
    """负责单个总排号在一次分类过程中的完整日志记录。

    典型用法（在 classifier.py 里）：
        logger = OrderLogger(log_dir, zong_pai_hao, mode)
        logger.log_param_text(param_text)
        # ... 每次调用LLM后:
        logger.log_llm_attempt(attempt_no, call_result, parsed_or_none, format_error_or_none)
        # ... 最终:
        logger.log_final_result(status, parsed_result_or_none)
        logger.close()
    """

    def __init__(self, log_dir: str | Path, zong_pai_hao: str, mode: str):
        self._log_dir = Path(log_dir)
        self._log_dir.mkdir(parents=True, exist_ok=True)
        timestamp = datetime.now().strftime("%Y%m%d_%H%M%S_%f")
        filename = f"{_safe_filename_part(zong_pai_hao)}_{timestamp}.log"
        self._path = self._log_dir / filename
        self._lines: list[str] = []

        self._write_header(zong_pai_hao, mode)

    @property
    def path(self) -> Path:
        return self._path

    def _write_header(self, zong_pai_hao: str, mode: str) -> None:
        self._lines.append("=" * 70)
        self._lines.append(f"总排号: {zong_pai_hao}")
        self._lines.append(f"分类模式: {mode}")
        self._lines.append(f"开始时间: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}")
        self._lines.append("=" * 70)
        self._flush()

    def log_param_text(self, param_text: str | None) -> None:
        self._lines.append("")
        self._lines.append("--- 新参数原文 ---")
        self._lines.append(param_text if param_text else "(空)")
        self._flush()

    def log_llm_attempt(
        self,
        attempt_no: int,
        call_result: LLMCallResult,
        parsed: ParsedResult | None,
        format_error: str | None,
    ) -> None:
        """记录一次 LLM 调用的完整往返，无论成功还是失败都调用本方法。

        call_result.error 非 None 表示网络/接口层面调用失败（对话都没发起成功
        或没拿到回复）；call_result.error 为 None 但 format_error 非 None 表示
        调用成功但模型输出格式不合规；两者都为 None 表示本次调用完全成功。
        """
        self._lines.append("")
        self._lines.append(f"--- 第 {attempt_no} 次尝试 ---")

        self._lines.append("")
        self._lines.append("[发送给模型的完整对话]")
        for msg in call_result.messages:
            role = msg["role"]
            content = msg["content"]
            self._lines.append(f"  [{role}]")
            for line in content.splitlines():
                self._lines.append(f"    {line}")

        self._lines.append("")
        if call_result.error is not None:
            self._lines.append("[调用结果] 网络/接口调用失败")
            self._lines.append(f"  错误信息: {call_result.error}")
        else:
            self._lines.append("[调用结果] 已收到模型回复")
            self._lines.append("[模型原始回复]")
            for line in (call_result.raw_response or "").splitlines():
                self._lines.append(f"  {line}")

            # 调用统计：模型名、单次调用耗时、token 用量与缓存命中情况。
            # 失败调用（未拿到 usage）时相应字段显示"无"，保证排查信息完整。
            self._lines.append("")
            self._lines.append("[调用统计]")
            self._lines.append(f"  模型: {call_result.model}")
            if call_result.elapsed_ms is not None:
                self._lines.append(f"  单次调用耗时: {call_result.elapsed_ms:.1f} ms")
            else:
                self._lines.append("  单次调用耗时: 无（调用失败）")
            u = call_result.usage
            if isinstance(u, dict):
                self._lines.append(
                    f"  tokens: prompt={u.get('prompt_tokens')} "
                    f"completion={u.get('completion_tokens')} "
                    f"total={u.get('total_tokens')}"
                )
                prompt = u.get("prompt_tokens") or 0
                hit = u.get("prompt_cache_hit_tokens") or 0
                rate = round(hit / prompt, 4) if prompt else 0.0
                self._lines.append(
                    f"  缓存: hit={u.get('prompt_cache_hit_tokens')} "
                    f"miss={u.get('prompt_cache_miss_tokens')} "
                    f"命中率={rate}"
                )
            else:
                self._lines.append("  用量统计: 无（调用失败或未返回 usage）")

            self._lines.append("")
            if format_error is not None:
                self._lines.append("[格式校验] 失败")
                self._lines.append(f"  原因: {format_error}")
            else:
                self._lines.append("[格式校验] 通过")
                if parsed is not None:
                    self._lines.append(f"  has_attachment = {parsed.has_attachment}")
                    self._lines.append(f"  types = {parsed.types}")
        self._flush()

    def log_final_result(self, status: str, parsed: ParsedResult | None) -> None:
        self._lines.append("")
        self._lines.append("=" * 70)
        self._lines.append(f"最终状态: {status}")
        if parsed is not None:
            self._lines.append(f"最终结果: has_attachment={parsed.has_attachment}, types={parsed.types}")
        self._lines.append(f"结束时间: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}")
        self._lines.append("=" * 70)
        self._flush()

    def _flush(self) -> None:
        """每次追加内容后立即落盘，避免进程中途异常退出导致日志丢失。"""
        with self._path.open("a", encoding="utf-8") as f:
            f.write("\n".join(self._lines) + "\n")
        self._lines = []

    def close(self) -> None:
        """预留的显式收尾方法；当前实现下 _flush 已即时写盘，close 是空操作，
        但保留该接口方便未来切换为缓冲写入而不改动调用方代码。"""
        pass