#!/usr/bin/env -S uv run --script # /// script # requires-python = ">=3.9" # dependencies = ["resend"] # /// """ Test du routage par POSTMASTER FILTERS (Zammad / ADVMFR) via Resend. Ce script exerce specifiquement les comportements que SEULS les Postmaster Filters savent faire (routage a l'ingestion, hors Elasticsearch, cf. config/routing/seed_postmaster_filters.rb) : - routage par ADRESSE (baseline) -> quote@ = Quote - RE-ROUTE MAL-ADRESSAGE (filtres 40/41/42) -> RFQ tombe dans accounting@/sales@/ support@ repart vers Quote + tag 'rerouted:auto' - EXCLUSION Re:/Fwd (nouvelles demandes seulement) -> 'Re: Quote ...' reste sur place - TRIAGE par SUJET (filtres 30x) -> contact@ + mot-cle sujet -> groupe - TRIAGE par CORPS (filtres 20x) *** nouveaute *** -> contact@ + sujet neutre, mot-cle dans le CORPS -> groupe - SUJET > CORPS (ordre alphabetique des noms) -> sujet 'quotation' l'emporte sur un corps 'invoice' - ENRICHISSEMENT (filtres 50/60) -> no-reply -> tag 'auto' + prio basse ; corps FR -> tag 'non-english' Chaque corps porte un marqueur "PMFLOW" pour que verify_routing.rb retrouve exactement ces tickets et affiche le flow (expediteur -> boite -> groupe atterri + tags + priorite). Aucun secret dans le repo : la cle API se passe par variable d'env RESEND_TEST. Usage (uv installe 'resend' via le bloc PEP 723) : export RESEND_TEST=re_xxx export RESEND_DOMAINS=elevaseo.com,ceoinsight.dev,seoinsight.app # defaut: ces 3 export ZAMMAD_DOMAIN=vivacoda.com # defaut: vivacoda.com uv run tests/routing-sim/send_postmaster_tests.py # envoie (batch) uv run tests/routing-sim/send_postmaster_tests.py --dry-run # affiche sans envoyer uv run tests/routing-sim/send_postmaster_tests.py --only accounting Apres envoi : laisser ~1 min (polling IMAP), puis me demander de lancer la verification (docker exec ... bin/rails r /tmp/verify_routing.rb) ou voir tests/routing-sim/README.md. """ import argparse import os import sys DEFAULT_DOMAINS = ["elevaseo.com", "ceoinsight.dev", "seoinsight.app"] MARKER = "PMFLOW" # marqueur discret ajoute au corps pour retrouver ces tickets (aucun mot-cle) # "to" = local part de la boite destinataire. "from_local" = local part expediteur (domaine par # rotation). "expected" = comportement Postmaster attendu (pour la verification). SCENARIOS = [ # --- Baseline : routage par adresse --- dict(to="quote", name="Claire Dubois", from_local="claire.dubois", subject="Quote request: 500 custom 14k gold rings", body="Hello, we would like a quotation and your MOQ for 500 custom 14k gold rings.", expected="Par ADRESSE -> Quote/Estimating (baseline, aucun filtre de re-route)."), # --- Re-route mal-adressage (40/41/42) : RFQ tombe dans la mauvaise boite --- dict(to="accounting", name="Marco Rossi", from_local="marco.rossi", subject="Quote request for 300 gold wedding rings", body="Hi, please send a quotation for 300 wedding rings in 18k gold. New enquiry.", expected="MAL-ADRESSE -> re-route Quote + tag 'rerouted:auto' (filtre 40 accounting->quote)."), dict(to="sales", name="Sophie Martin", from_local="sophie.martin", subject="RFQ: pricing for 200 18k pendants", body="Hello, could you quote pricing for 200 pendants in 18k gold? Thanks.", expected="MAL-ADRESSE -> re-route Quote + tag 'rerouted:auto' (filtre 41 sales->quote)."), dict(to="support", name="Paolo Greco", from_local="paolo.greco", subject="Request for quotation, bulk silver earrings", body="Hi, I need a quotation for a bulk order of 925 silver earrings.", expected="MAL-ADRESSE -> re-route Quote + tag 'rerouted:auto' (filtre 42 support->quote)."), # --- Exclusion Re: (nouvelles demandes seulement) --- dict(to="accounting", name="Nina Schmidt", from_local="nina.schmidt", subject="Re: Quote Q-123 - invoice attached for payment", body="Hello, following your quotation Q-123, here is the invoice to process the payment.", expected="RESTE Accounting (sujet 'Re:' exclu du re-route -> vrai suivi compta)."), # --- Triage contact@ par SUJET (30x) --- dict(to="contact", name="Lucas Moreau", from_local="lucas.moreau", subject="Quotation for 1000 silver earrings", body="Hi, could I get a quotation for 1000 pairs of 925 silver earrings?", expected="TRIAGE sujet -> Quote/Estimating (filtre 30x)."), dict(to="contact", name="Yuki Tanaka", from_local="yuki.tanaka", subject="Invoice question about my payment", body="Hello, I have a question about an invoice and a payment that did not go through.", expected="TRIAGE sujet -> Accounting (filtre 31x)."), # --- Triage contact@ par CORPS (20x) : sujet NEUTRE, mot-cle dans le CORPS *** nouveaute *** --- dict(to="contact", name="Tom Becker", from_local="tom.becker", subject="Question about a custom piece", body="Hi, I have a technical drawing and a STEP CAD file to review for feasibility.", expected="TRIAGE corps -> Engineering/CAD (filtre 23x) : sujet neutre, mot-cle dans le corps."), dict(to="contact", name="Emma Wilson", from_local="emma.wilson", subject="Following up on my order", body="Could you confirm the delivery lead time and shipping options to France?", expected="TRIAGE corps -> Logistics (filtre 22x) : sujet neutre, mot-cle dans le corps."), # --- Sujet > Corps (ordre alphabetique des noms de filtres) --- dict(to="contact", name="David Lee", from_local="david.lee", subject="Request for quotation on a new collection", body="We also have an invoice and a payment matter to sort out separately.", expected="SUJET > CORPS -> Quote (30x apres 21x) : sujet 'quotation' l'emporte sur corps 'invoice'."), # --- Enrichissement (50/60) --- dict(to="quote", name="Mailer Daemon", from_local="no-reply", subject="Automated notification", body="This is an automated message. Please do not reply.", expected="Quote (adresse) + TAG 'auto' + PRIORITE BASSE (filtre 50, from no-reply)."), dict(to="contact", name="Camille Petit", from_local="camille.petit", subject="Demande de renseignements", body="Bonjour, je voudrais un devis pour des bagues en or. Cordialement, Camille.", expected="RESTE Triage (regex EN ne matche pas 'devis') + TAG 'non-english' (filtre 60, corps FR)."), ] def main() -> None: ap = argparse.ArgumentParser(description="Test routage Postmaster Filters Zammad via Resend.") ap.add_argument("--dry-run", action="store_true", help="affiche sans envoyer") ap.add_argument("--only", help="ne traiter que les scenarios vers cette boite (local part)") args = ap.parse_args() zammad_domain = os.environ.get("ZAMMAD_DOMAIN", "vivacoda.com") domains = [d.strip() for d in os.environ.get( "RESEND_DOMAINS", ",".join(DEFAULT_DOMAINS)).split(",") if d.strip()] if not domains: sys.exit("RESEND_DOMAINS vide.") scenarios = [s for s in SCENARIOS if not args.only or s["to"] == args.only] if not scenarios: sys.exit(f"Aucun scenario pour --only {args.only}") params, meta = [], [] for i, s in enumerate(scenarios): domain = domains[i % len(domains)] from_addr = f"{s['name']} <{s['from_local']}@{domain}>" to_addr = f"{s['to']}@{zammad_domain}" body = f"{s['body']}\n\n-- ref {MARKER}" # marqueur pour la verification params.append({"from": from_addr, "to": [to_addr], "subject": s["subject"], "text": body}) meta.append((s, from_addr, to_addr)) print(f"Destinataire: @{zammad_domain} | expediteurs (rotation): {', '.join(domains)} | " f"{'DRY-RUN' if args.dry_run else 'ENVOI (batch)'} | {len(params)} email(s) | marqueur: {MARKER}\n") if args.dry_run: for s, from_addr, to_addr in meta: print(f"[DRY] {to_addr:26} <- {from_addr}\n subj: {s['subject']}") else: import resend # SDK officiel (installe par uv via PEP 723) api_key = os.environ.get("RESEND_TEST") if not api_key: sys.exit("RESEND_TEST manquant (export RESEND_TEST=re_xxx).") resend.api_key = api_key try: result = resend.Batch.send(params) except Exception as e: # noqa: BLE001 sys.exit(f"Echec du batch Resend : {e}") ids = [e.get("id", "(no id)") for e in result.get("data", [])] for idx, (s, from_addr, to_addr) in enumerate(meta): eid = ids[idx] if idx < len(ids) else "(no id)" print(f"[OK {eid}] {to_addr:26} <- {from_addr}\n subj: {s['subject']}") print("\n==== FLOW ATTENDU (verifier ~1 min apres, via verify_routing.rb) ====") for s, from_addr, to_addr in meta: print(f"- {to_addr:26} <- {s['from_local']}@...\n {s['subject'][:58]!r}\n => {s['expected']}") print(f"\nNote: filtre '52 internal' (@{zammad_domain}) non testable via Resend sauf domaine verifie.") if __name__ == "__main__": main()