""" 错误快照模块 保存请求错误的完整上下文，便于调试和问题排查。 特性: - 分层目录结构 - 完整的请求/响应链路记录 - 自动清理旧快照 - 与主日志系统集成 """ import json import shutil from datetime import datetime, timedelta from typing import Any, cast from pathlib import Path class ErrorSnapshotManager: """错误快照管理器""" def __init__( self, base_dir: str = "errors", max_snapshots: int = 100, max_age_days: int = 7, compress_old: bool = True ): """ 初始化错误快照管理器 Args: base_dir: 快照存储基础目录 max_snapshots: 最大保留快照数量 max_age_days: 快照最大保留天数 compress_old: 是否压缩旧快照 """ self.base_dir = Path(base_dir) self.max_snapshots = max_snapshots self.max_age_days = max_age_days self.compress_old = compress_old # 确保目录存在 self.base_dir.mkdir(parents=True, exist_ok=True) def save_snapshot( self, downstream_payload: dict[str, Any], upstream_payload: dict[str, Any], upstream_response: str, error_type: str = "request_error", metadata: dict[str, Any] | None = None ) -> str | None: """ 保存错误快照 Args: downstream_payload: 接收到的原始请求 upstream_payload: 发送给上游的请求 upstream_response: 上游返回的响应 error_type: 错误类型 metadata: 额外元数据 Returns: 快照目录路径，失败返回 None """ try: # 创建时间戳目录 timestamp = datetime.now() date_dir = timestamp.strftime("%Y-%m-%d") time_str = timestamp.strftime("%H%M%S_%f") # 目录结构: errors/2024-01-21/http_400_143052_123456/ snapshot_name = f"{error_type}_{time_str}" snapshot_dir = self.base_dir / date_dir / snapshot_name snapshot_dir.mkdir(parents=True, exist_ok=True) # 1. 保存下游请求 self._save_json( snapshot_dir / "1_downstream_request.json", downstream_payload ) # 2. 保存上游请求 self._save_json( snapshot_dir / "2_upstream_request.json", upstream_payload ) # 3. 保存上游响应 resp_ext = self._save_response( snapshot_dir / "3_upstream_response", upstream_response ) # 4. 保存摘要 summary: dict[str, Any] = { "timestamp": timestamp.isoformat(), "error_type": error_type, "snapshot_id": snapshot_name, "files": [ "1_downstream_request.json", "2_upstream_request.json", f"3_upstream_response{resp_ext}" ], "metadata": metadata or {} } self._save_json(snapshot_dir / "summary.json", summary) # 清理旧快照 self._cleanup_old_snapshots() return str(snapshot_dir) except Exception: return None def _save_json(self, path: Path, data: Any): """保存 JSON 文件""" with open(path, 'w', encoding='utf-8') as f: if isinstance(data, (dict, list)): json.dump(data, f, ensure_ascii=False, indent=2) else: f.write(str(data)) def _save_response(self, path_base: Path, response: str) -> str: """保存响应内容，自动检测格式，返回文件后缀""" if self._is_json(response): try: data = json.loads(response) self._save_json(path_base.with_suffix('.json'), data) return ".json" except json.JSONDecodeError: pass # 保存为文本 with open(path_base.with_suffix('.txt'), 'w', encoding='utf-8') as f: f.write(str(response)) return ".txt" def _is_json(self, text: str) -> bool: """检查是否为 JSON 格式""" # Pylance: text 已经是 str，不需要 isinstance 检查 # 但为了运行时健壮性，如果 text 可能为 None 或其他类型，可以保留， # 但既然类型提示是 str，我们假设它就是 str。 # 如果调用者可能传 None，类型提示应该是 Optional[str] # 兼容性处理：如果 text 为空 if not text: return False # 安全转换，防止传入了非字符串 text_str = str(text).strip() return text_str.startswith('{') or text_str.startswith('[') def _cleanup_old_snapshots(self): """清理旧快照""" try: # 获取所有日期目录 date_dirs = sorted( [d for d in self.base_dir.iterdir() if d.is_dir()], key=lambda x: x.name, reverse=True ) # 按日期清理 cutoff_date = datetime.now() - timedelta(days=self.max_age_days) cutoff_str = cutoff_date.strftime("%Y-%m-%d") for date_dir in date_dirs: if date_dir.name < cutoff_str: # 删除过期目录 shutil.rmtree(date_dir) # 按数量清理 all_snapshots: list[dict[str, Any]] = [] for date_dir in self.base_dir.iterdir(): if date_dir.is_dir(): for snapshot_dir in date_dir.iterdir(): if snapshot_dir.is_dir(): summary_file = snapshot_dir / "summary.json" if summary_file.exists(): try: with open(summary_file, 'r') as f: summary = json.load(f) all_snapshots.append({ 'path': snapshot_dir, 'timestamp': str(summary.get('timestamp', '')) }) except Exception: pass # 按时间排序，删除超出数量限制的 # 显式指定 key 函数的返回类型 def sort_key(x: dict[str, Any]) -> str: return str(x['timestamp']) all_snapshots.sort(key=sort_key, reverse=True) for snapshot in all_snapshots[self.max_snapshots:]: snapshot_path = cast(Path, snapshot['path']) shutil.rmtree(snapshot_path) # 清理空的日期目录 for date_dir in self.base_dir.iterdir(): if date_dir.is_dir() and not any(date_dir.iterdir()): date_dir.rmdir() except Exception: pass def list_snapshots( self, error_type: str | None = None, limit: int = 20 ) -> list[dict[str, Any]]: """ 列出快照 Args: error_type: 过滤错误类型 limit: 返回数量限制 Returns: 快照摘要列表 """ snapshots: list[dict[str, Any]] = [] try: for date_dir in sorted(self.base_dir.iterdir(), reverse=True): if not date_dir.is_dir(): continue for snapshot_dir in sorted(date_dir.iterdir(), reverse=True): if not snapshot_dir.is_dir(): continue summary_file = snapshot_dir / "summary.json" if not summary_file.exists(): continue try: with open(summary_file, 'r') as f: summary = json.load(f) if error_type and summary.get('error_type') != error_type: continue summary['path'] = str(snapshot_dir) snapshots.append(summary) if len(snapshots) >= limit: return snapshots except Exception: continue except Exception: pass return snapshots def get_snapshot(self, snapshot_path: str) -> dict[str, Any] | None: """ 获取完整快照内容 Args: snapshot_path: 快照目录路径 Returns: 快照完整内容 """ try: snapshot_dir = Path(snapshot_path) if not snapshot_dir.exists(): return None result: dict[str, Any] = {} for file in snapshot_dir.iterdir(): if file.suffix == '.json': with open(file, 'r', encoding='utf-8') as f: result[file.stem] = json.load(f) elif file.suffix == '.txt': with open(file, 'r', encoding='utf-8') as f: result[file.stem] = f.read() return result except Exception: return None # ==================== 全局实例 ==================== _snapshot_manager: ErrorSnapshotManager | None = None def _get_manager() -> ErrorSnapshotManager: """获取或创建快照管理器""" global _snapshot_manager if _snapshot_manager is None: try: from src.core.config import load_config config = load_config() base_dir = config.get("error_dir", "errors") except Exception: base_dir = "errors" _snapshot_manager = ErrorSnapshotManager(base_dir=base_dir) return _snapshot_manager # ==================== 便捷函数 ==================== def save_error_snapshot( downstream_payload: dict[str, Any], upstream_payload: dict[str, Any], upstream_response: str, error_type: str = "request_error" ) -> str | None: """ 保存请求错误快照 这是主要的对外接口，保持与原有 API 兼容。 Args: downstream_payload: 接收到的原始请求 (Gemini 格式) upstream_payload: 最终发送给 Google 的 Payload upstream_response: 上游返回的原始错误信息 error_type: 错误类别 (如 http_400, transform_error 等) Returns: 快照目录路径，失败返回 None """ manager = _get_manager() return manager.save_snapshot( downstream_payload=downstream_payload, upstream_payload=upstream_payload, upstream_response=upstream_response, error_type=error_type ) def list_error_snapshots( error_type: str | None = None, limit: int = 20 ) -> list[dict[str, Any]]: """ 列出错误快照 Args: error_type: 过滤错误类型 limit: 返回数量限制 Returns: 快照摘要列表 """ manager = _get_manager() return manager.list_snapshots(error_type=error_type, limit=limit) def get_error_snapshot(snapshot_path: str) -> dict[str, Any] | None: """ 获取完整错误快照 Args: snapshot_path: 快照目录路径 Returns: 快照完整内容 """ manager = _get_manager() return manager.get_snapshot(snapshot_path)