138 lines
4.7 KiB
Python
Executable File
138 lines
4.7 KiB
Python
Executable File
#!/usr/bin/env python3
|
|
"""Reprocessa eventos Chatwoot inbound que ficaram pendentes em raw_events.
|
|
|
|
Uso seguro:
|
|
PYTHONPATH=. python scripts/reprocess_pending_chatwoot_raw_events.py --dry-run
|
|
PYTHONPATH=. python scripts/reprocess_pending_chatwoot_raw_events.py --limit 50
|
|
PYTHONPATH=. python scripts/reprocess_pending_chatwoot_raw_events.py --include-errors --limit 20
|
|
|
|
Critério principal v4928.1.5.25:
|
|
source_system='chatwoot' + event_type='message_created' + payload.message_type='incoming'
|
|
|
|
Não usa sender.type=contact porque em emails Chatwoot esse campo pode vir vazio.
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import argparse
|
|
import asyncio
|
|
import json
|
|
import os
|
|
import sys
|
|
from pathlib import Path
|
|
from typing import Any
|
|
|
|
PROJECT_ROOT = Path(__file__).resolve().parents[1]
|
|
sys.path.insert(0, str(PROJECT_ROOT))
|
|
os.chdir(PROJECT_ROOT)
|
|
|
|
from sqlalchemy import text
|
|
|
|
from app.db import engine
|
|
from app.webhooks_chatwoot import process_saved_chatwoot_raw_event
|
|
|
|
|
|
def _as_dict(value: Any) -> dict[str, Any]:
|
|
if isinstance(value, dict):
|
|
return value
|
|
if isinstance(value, str):
|
|
try:
|
|
parsed = json.loads(value)
|
|
return parsed if isinstance(parsed, dict) else {}
|
|
except Exception:
|
|
return {}
|
|
return {}
|
|
|
|
|
|
def _preview(value: Any, limit: int = 180) -> str:
|
|
text_value = " ".join(str(value or "").replace("\r", " ").replace("\n", " ").split())
|
|
return text_value if len(text_value) <= limit else text_value[: limit - 1].rstrip() + "…"
|
|
|
|
|
|
def load_events(*, limit: int, include_errors: bool, source_event_id: str | None) -> list[dict[str, Any]]:
|
|
where = [
|
|
"source_system = 'chatwoot'",
|
|
"event_type = 'message_created'",
|
|
"COALESCE(payload #>> '{message_type}', '') = 'incoming'",
|
|
]
|
|
params: dict[str, Any] = {"limit": int(limit)}
|
|
|
|
if source_event_id:
|
|
where.append("source_event_id = :source_event_id")
|
|
params["source_event_id"] = str(source_event_id)
|
|
elif include_errors:
|
|
where.append("processed = FALSE")
|
|
where.append("ignored = FALSE")
|
|
where.append("processing_error IS NOT NULL")
|
|
else:
|
|
where.append("processed = FALSE")
|
|
where.append("ignored = FALSE")
|
|
where.append("processing_error IS NULL")
|
|
|
|
sql = text(f"""
|
|
SELECT
|
|
id::text,
|
|
source_event_id,
|
|
created_at,
|
|
conversation_id,
|
|
contact_id,
|
|
processing_error,
|
|
payload
|
|
FROM raw_events
|
|
WHERE {' AND '.join(where)}
|
|
ORDER BY created_at ASC
|
|
LIMIT :limit
|
|
""")
|
|
|
|
with engine.begin() as conn:
|
|
return [dict(row) for row in conn.execute(sql, params).mappings().all()]
|
|
|
|
|
|
async def main_async() -> int:
|
|
parser = argparse.ArgumentParser()
|
|
parser.add_argument("--limit", type=int, default=50, help="Máximo de eventos a reprocessar")
|
|
parser.add_argument("--dry-run", action="store_true", help="Lista eventos sem processar")
|
|
parser.add_argument("--include-errors", action="store_true", help="Reprocessa eventos com processing_error")
|
|
parser.add_argument("--source-event-id", help="Reprocessa só uma mensagem Chatwoot específica")
|
|
args = parser.parse_args()
|
|
|
|
rows = load_events(limit=args.limit, include_errors=args.include_errors, source_event_id=args.source_event_id)
|
|
print(f"Eventos Chatwoot incoming selecionados: {len(rows)}")
|
|
|
|
for row in rows[:80]:
|
|
payload = _as_dict(row.get("payload"))
|
|
print(
|
|
f"- {row['created_at']} raw={row['id']} msg={row.get('source_event_id')} "
|
|
f"conv={payload.get('conversation', {}).get('id') or row.get('conversation_id')} "
|
|
f"email={payload.get('sender', {}).get('email') or '-'} "
|
|
f"err={row.get('processing_error') or '-'} "
|
|
f"content={_preview(payload.get('content'))}"
|
|
)
|
|
|
|
if args.dry_run or not rows:
|
|
print("Dry-run: nenhuma alteração aplicada." if args.dry_run else "Nada para processar.")
|
|
return 0
|
|
|
|
summary: dict[str, int] = {}
|
|
for row in rows:
|
|
payload = _as_dict(row.get("payload"))
|
|
try:
|
|
result = await process_saved_chatwoot_raw_event(row["id"], payload)
|
|
status = str(result.get("status") or "processed")
|
|
except Exception as exc:
|
|
status = f"exception:{type(exc).__name__}"
|
|
print(f"ERRO raw={row['id']} msg={row.get('source_event_id')}: {exc}")
|
|
summary[status] = summary.get(status, 0) + 1
|
|
|
|
print("\nResumo:")
|
|
for status, total in sorted(summary.items(), key=lambda item: (-item[1], item[0])):
|
|
print(f"{status}: {total}")
|
|
return 0
|
|
|
|
|
|
def main() -> int:
|
|
return asyncio.run(main_async())
|
|
|
|
|
|
if __name__ == "__main__":
|
|
raise SystemExit(main())
|