"""Safe, copyable diagnostics for durable document-rewrite failures.""" from __future__ import annotations import re import traceback from datetime import UTC, datetime from pathlib import Path from typing import Any _MAX_MESSAGE_LENGTH = 2_000 _MAX_TRACE_FRAMES = 10 _SECRET_PATTERNS = ( re.compile(r"sk-[A-Za-z0-9_-]{3,}"), re.compile(r"(?:Bearer\s+)[A-Za-z0-9\-._~+/]+=*", re.IGNORECASE), re.compile( r"((?:api[_-]?key|authorization|access[_-]?token|refresh[_-]?token|token|secret|password)\s*[:=]\s*)[^\s,;]+", re.IGNORECASE, ), re.compile(r"(https?://[^\s:/]+:)[^\s@/]+@", re.IGNORECASE), ) def _redact(value: Any, *, limit: int = _MAX_MESSAGE_LENGTH) -> str: text = str(value or "") for pattern in _SECRET_PATTERNS: replacement = r"\1[REDACTED]" if pattern.groups else "[REDACTED]" text = pattern.sub(replacement, text) return text[:limit] def _display_trace_path(filename: str) -> str: """Keep application-relative paths useful across Windows and Linux hosts.""" normalized = filename.replace("\\", "/") for marker in ("/app/backend/", "/backend/", "/site-packages/"): if marker in normalized: return normalized.split(marker, 1)[1] if marker != "/site-packages/" else f"site-packages/{normalized.split(marker, 1)[1]}" try: return Path(normalized).name or normalized except Exception: # pragma: no cover - defensive for exotic traceback paths return normalized def _root_exception(exc: BaseException) -> BaseException: current = exc seen: set[int] = set() while id(current) not in seen: seen.add(id(current)) next_error = current.__cause__ or current.__context__ if next_error is None: break current = next_error return current def build_document_rewrite_diagnostic( *, job_id: str | None, stage: str, error_code: str, message: str, exc: BaseException | None = None, ) -> dict[str, Any]: """Build a bounded diagnostic suitable for an authenticated API response. Trace frames intentionally omit source lines and locals. This preserves the file/line/function evidence needed for Docker/Linux compatibility work without returning prompts, document contents, headers, or credentials. """ safe_message = _redact(message) or error_code or "全文改写失败" exception_type = type(exc).__name__ if exc is not None else None trace_tail: list[str] = [] root_type: str | None = None root_message: str | None = None if exc is not None: frames = traceback.extract_tb(exc.__traceback__)[-_MAX_TRACE_FRAMES:] trace_tail = [ f"{_display_trace_path(frame.filename)}:{frame.lineno} in {frame.name}" for frame in frames ] root = _root_exception(exc) if root is not exc: root_type = type(root).__name__ root_message = _redact(root) diagnostic: dict[str, Any] = { "diagnosticId": f"drw-{job_id}" if job_id else f"drw-{datetime.now(UTC).strftime('%Y%m%d%H%M%S%f')}", "jobId": job_id, "stage": stage or "generation", "errorCode": error_code or exception_type or "DOCUMENT_REWRITE_FAILED", "exceptionType": exception_type, "message": safe_message, "traceTail": trace_tail, "occurredAt": datetime.now(UTC).isoformat(), } if root_type: diagnostic["rootExceptionType"] = root_type if root_message: diagnostic["rootMessage"] = root_message return diagnostic __all__ = ["build_document_rewrite_diagnostic"]