Bot de Telegram para monitoreo de notificaciones judiciales del PJ Paraguay. Multi-tenant con onboarding conversacional, audit log, y polling automático. Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
58 lines
1.7 KiB
Python
58 lines
1.7 KiB
Python
"""
|
|
audit.py — Audit log de toda la actividad del bot.
|
|
|
|
Tabla audit_log en la misma DB SQLite que tenants.
|
|
La función log() es fire-and-tolerant: nunca levanta excepciones
|
|
para no romper el flujo principal de la aplicación.
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
from datetime import datetime, timezone
|
|
from zoneinfo import ZoneInfo
|
|
|
|
import structlog
|
|
|
|
logger = structlog.get_logger(__name__)
|
|
|
|
|
|
async def log(
|
|
evento: str,
|
|
chat_id: int | None = None,
|
|
detalle: dict | None = None,
|
|
resultado: str = "ok",
|
|
duracion_ms: int | None = None,
|
|
tenant: dict | None = None,
|
|
) -> None:
|
|
"""
|
|
Registra un evento en audit_log.
|
|
Si falla (DB no disponible, error de escritura), solo loggea a structlog.
|
|
Nunca levanta excepciones.
|
|
"""
|
|
from database import get_conn # import tardío para evitar circular imports
|
|
|
|
now_utc = datetime.now(timezone.utc)
|
|
timestamp = now_utc.isoformat()
|
|
|
|
timestamp_local: str | None = None
|
|
if tenant:
|
|
tz_str = tenant.get("timezone") or "America/Asuncion"
|
|
try:
|
|
timestamp_local = now_utc.astimezone(ZoneInfo(tz_str)).isoformat()
|
|
except Exception:
|
|
timestamp_local = None
|
|
|
|
detalle_json = json.dumps(detalle, ensure_ascii=False) if detalle else None
|
|
|
|
try:
|
|
conn = get_conn()
|
|
await conn.execute(
|
|
"""INSERT INTO audit_log
|
|
(timestamp, timestamp_local, chat_id, evento, detalle, resultado, duracion_ms)
|
|
VALUES (?, ?, ?, ?, ?, ?, ?)""",
|
|
(timestamp, timestamp_local, chat_id, evento, detalle_json, resultado, duracion_ms),
|
|
)
|
|
await conn.commit()
|
|
except Exception as exc:
|
|
logger.error("audit_log_write_error", evento=evento, chat_id=chat_id, error=str(exc))
|