#!/usr/bin/env python3 """Clean noisy pending Operations items created from mailbox/system messages. Dry-run by default. With --apply, marks obvious bounce/NDR/system tasks as ``skipped`` and stores a cleanup marker in task metadata. This does not delete messages, raw events, or Chatwoot conversations. """ from __future__ import annotations import argparse import json from sqlalchemy import text from app.db import engine SQL_TERMS = [ "postmaster", "mailer-daemon", "mail delivery subsystem", "mail delivery system", "microsoft exchange", "office 365", "undeliverable", "returned mail", "delivery status notification", "non-delivery report", "non delivery report", "your message couldn't be delivered", "your message couldnt be delivered", "recipient wasn't found", "recipient was not found", "unknown to address", "delivery has failed", "mail delivery failed", "remote server returned", "550 5.1.1", "5.1.10", "wasn't found at", ] NOISE_ACTION_CODES = [ "IGNORE_BOUNCE", "IGNORE_SPAM", "NO_ACTION", ] def _json(value: object) -> str: return json.dumps(value or {}, ensure_ascii=False, default=str) def find_candidates(limit: int = 200) -> list[dict]: term_params = {f"term_{idx}": f"%{term}%" for idx, term in enumerate(SQL_TERMS)} term_sql = " OR ".join([f"noise_text ILIKE :term_{idx}" for idx, _ in enumerate(SQL_TERMS)]) sql = text(f""" WITH task_context AS ( SELECT t.id::text, t.created_at, t.action_code, t.route, t.priority, t.status, t.conversation_id, t.contact_id, t.opportunity_id::text, COALESCE(t.note, '') AS note, COALESCE(t.action, '') AS action, COALESCE(re.payload->'sender'->>'name', '') AS sender_name, COALESCE(re.payload->'sender'->>'email', '') AS sender_email, COALESCE( re.payload->'conversation'->'additional_attributes'->>'mail_subject', re.payload->'content_attributes'->'email'->>'subject', re.payload->'conversation'->'messages'->0->'content_attributes'->'email'->>'subject', '' ) AS subject, COALESCE(m.clean_body, m.raw_body, re.payload->>'content', '') AS body, lower( COALESCE(t.action_code, '') || ' ' || COALESCE(t.route, '') || ' ' || COALESCE(t.action, '') || ' ' || COALESCE(t.note, '') || ' ' || COALESCE(re.payload->'sender'->>'name', '') || ' ' || COALESCE(re.payload->'sender'->>'email', '') || ' ' || COALESCE(re.payload->'conversation'->'additional_attributes'->>'mail_subject', '') || ' ' || COALESCE(re.payload->'content_attributes'->'email'->>'subject', '') || ' ' || COALESCE(re.payload->'conversation'->'messages'->0->'content_attributes'->'email'->>'subject', '') || ' ' || COALESCE(m.clean_body, '') || ' ' || COALESCE(m.raw_body, '') || ' ' || COALESCE(re.payload->>'content', '') ) AS noise_text FROM tasks t LEFT JOIN messages m ON m.id = t.message_id LEFT JOIN raw_events re ON re.id = t.raw_event_id WHERE t.status = 'pending' ) SELECT id, created_at, action_code, route, priority, status, conversation_id, contact_id, opportunity_id, sender_name, sender_email, subject, action, note FROM task_context WHERE ({term_sql} OR upper(coalesce(action_code,'')) = ANY(:noise_action_codes)) ORDER BY created_at DESC LIMIT :limit """) params = dict(term_params) params["noise_action_codes"] = NOISE_ACTION_CODES params["limit"] = int(limit) with engine.begin() as conn: rows = conn.execute(sql, params).mappings().all() return [dict(row) for row in rows] def apply_cleanup(candidates: list[dict]) -> int: if not candidates: return 0 ids = [row["id"] for row in candidates] payload = _json({ "cleanup_reason": "operations_noise_bounce_or_system_message", "cleanup_by": "cleanup_operations_noise", "clientflow_version": "v4.9.0", }) with engine.begin() as conn: for task_id in ids: conn.execute(text(""" UPDATE tasks SET status = 'skipped', updated_at = now(), done_at = COALESCE(done_at, now()), done_by = 'cleanup_operations_noise', metadata = COALESCE(metadata, '{}'::jsonb) || CAST(:payload AS JSONB) WHERE id = CAST(:task_id AS UUID) AND status = 'pending' """), {"task_id": task_id, "payload": payload}) conn.execute(text(""" INSERT INTO task_events (task_id, event_type, payload, created_by) VALUES (CAST(:task_id AS UUID), 'task_skipped_noise_cleanup', CAST(:payload AS JSONB), 'system') """), {"task_id": task_id, "payload": payload}) return len(ids) def main() -> None: parser = argparse.ArgumentParser(description="Clean pending Operations noise created from bounces/NDRs/system messages.") parser.add_argument("--limit", type=int, default=200) parser.add_argument("--apply", action="store_true", help="Apply cleanup. Without this, only prints candidates.") args = parser.parse_args() candidates = find_candidates(limit=args.limit) print(f"Found {len(candidates)} candidate task(s).") for row in candidates: subject = (row.get("subject") or row.get("note") or "")[:90] sender = row.get("sender_email") or row.get("sender_name") or row.get("contact_id") or "sem remetente" print(f"- {row.get('id')} | {row.get('action_code')} | {sender} | {subject}") if args.apply: total = apply_cleanup(candidates) print(f"Marked {total} task(s) as skipped.") else: print("Dry-run only. Re-run with --apply to mark candidates as skipped.") if __name__ == "__main__": main()