128 lines
3.3 KiB
Python
128 lines
3.3 KiB
Python
import json
|
|
import logging
|
|
import os
|
|
import re
|
|
import traceback
|
|
import uuid
|
|
from copy import deepcopy
|
|
from datetime import datetime
|
|
from logging.handlers import RotatingFileHandler
|
|
|
|
|
|
from paths import BASE_DIR, LOGS_DIR as LOG_DIR, ERROR_LOG_PATH
|
|
|
|
MAX_STRING_LENGTH = 4000
|
|
MAX_COLLECTION_ITEMS = 50
|
|
|
|
SENSITIVE_KEYS = {
|
|
"authorization",
|
|
"api_token",
|
|
"token",
|
|
"access_token",
|
|
"refresh_token",
|
|
"cookie",
|
|
"set-cookie",
|
|
"password",
|
|
"secret",
|
|
}
|
|
|
|
BEARER_RE = re.compile(r"Bearer\s+[A-Za-z0-9._\-]+", re.IGNORECASE)
|
|
|
|
|
|
def _ensure_logger():
|
|
os.makedirs(LOG_DIR, exist_ok=True)
|
|
logger = logging.getLogger("mp_manager.errors")
|
|
if logger.handlers:
|
|
return logger
|
|
|
|
logger.setLevel(logging.ERROR)
|
|
logger.propagate = False
|
|
handler = RotatingFileHandler(
|
|
ERROR_LOG_PATH,
|
|
maxBytes=5 * 1024 * 1024,
|
|
backupCount=5,
|
|
encoding="utf-8",
|
|
)
|
|
handler.setFormatter(logging.Formatter("%(message)s"))
|
|
logger.addHandler(handler)
|
|
return logger
|
|
|
|
|
|
def new_error_id():
|
|
return str(uuid.uuid4())
|
|
|
|
|
|
def sanitize(value, depth=0):
|
|
if depth > 6:
|
|
return "<max-depth>"
|
|
|
|
if isinstance(value, dict):
|
|
clean = {}
|
|
for key, item in list(value.items())[:MAX_COLLECTION_ITEMS]:
|
|
key_text = str(key)
|
|
if key_text.lower() in SENSITIVE_KEYS or "token" in key_text.lower():
|
|
clean[key_text] = "<redacted>"
|
|
else:
|
|
clean[key_text] = sanitize(item, depth + 1)
|
|
return clean
|
|
|
|
if isinstance(value, (list, tuple, set)):
|
|
return [sanitize(item, depth + 1) for item in list(value)[:MAX_COLLECTION_ITEMS]]
|
|
|
|
if isinstance(value, bytes):
|
|
value = value.decode("utf-8", errors="replace")
|
|
|
|
if isinstance(value, str):
|
|
text = BEARER_RE.sub("Bearer <redacted>", value)
|
|
if len(text) > MAX_STRING_LENGTH:
|
|
return text[:MAX_STRING_LENGTH] + "...<truncated>"
|
|
return text
|
|
|
|
return value
|
|
|
|
|
|
def format_exception(exc):
|
|
if exc is None:
|
|
return None
|
|
return {
|
|
"type": type(exc).__name__,
|
|
"message": sanitize(str(exc)),
|
|
"traceback": sanitize("".join(traceback.format_exception(type(exc), exc, exc.__traceback__))),
|
|
}
|
|
|
|
|
|
def log_error(event, exc=None, context=None, *, error_id=None):
|
|
"""
|
|
Registra errores técnicos en JSONL y, si la DB está disponible, también en SQLite.
|
|
Devuelve error_id para poder correlacionarlo con la terminal o respuesta HTTP.
|
|
"""
|
|
error_id = error_id or new_error_id()
|
|
context = sanitize(deepcopy(context or {}))
|
|
exception_data = format_exception(exc)
|
|
record = {
|
|
"error_id": error_id,
|
|
"timestamp": datetime.now().strftime("%Y-%m-%d %H:%M:%S"),
|
|
"event": event,
|
|
"context": context,
|
|
"exception": exception_data,
|
|
}
|
|
|
|
logger = _ensure_logger()
|
|
logger.error(json.dumps(record, ensure_ascii=False, default=str))
|
|
|
|
try:
|
|
import db
|
|
|
|
db.insert_error_log(
|
|
error_id=error_id,
|
|
event=event,
|
|
exception_type=exception_data.get("type") if exception_data else None,
|
|
exception_message=exception_data.get("message") if exception_data else None,
|
|
context=record,
|
|
)
|
|
except Exception:
|
|
# El archivo JSONL es la fuente primaria si SQLite no está disponible.
|
|
pass
|
|
|
|
return error_id
|