104 lines
3.6 KiB
Python
104 lines
3.6 KiB
Python
"""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"]
|