348 lines
13 KiB
Python
Executable File
348 lines
13 KiB
Python
Executable File
#!/usr/bin/env python3
|
|
"""Batch validation for email identity extraction.
|
|
|
|
Safe for LLM runs: prints progress, truncates long bodies, applies a per-message
|
|
process timeout, and writes JSONL incrementally so partial results are kept even
|
|
if a provider call stalls.
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import argparse
|
|
import csv
|
|
import json
|
|
import multiprocessing as mp
|
|
import os
|
|
import re
|
|
from datetime import datetime
|
|
from pathlib import Path
|
|
from typing import Any, Dict, List
|
|
|
|
from sqlalchemy import text
|
|
|
|
from app.commercial_service import normalize_fiscal_name
|
|
from app.db import engine
|
|
|
|
|
|
LEGAL_SUFFIX_TOKENS = {
|
|
"lda", "limitada", "unipessoal", "sa", "s", "a", "sociedade",
|
|
"mediação", "mediacao", "seguro", "seguros", "importação", "importacao",
|
|
"exportação", "exportacao", "fabricação", "fabricacao", "representação",
|
|
"representacao", "soluções", "solucoes", "metálicas", "metalicas",
|
|
}
|
|
|
|
|
|
def clean(value: Any) -> str:
|
|
return re.sub(r"\s+", " ", str(value or "")).strip()
|
|
|
|
|
|
def norm(value: Any) -> str:
|
|
return normalize_fiscal_name(value or "") or clean(value).casefold()
|
|
|
|
|
|
def meaningful_tokens(value: Any) -> set[str]:
|
|
n = norm(value)
|
|
tokens = {t for t in re.split(r"[^a-z0-9áàâãéèêíìîóòôõúùûç]+", n) if len(t) >= 3}
|
|
return {t for t in tokens if t not in LEGAL_SUFFIX_TOKENS}
|
|
|
|
|
|
def mentions_match_fiscal(mentions: List[str], fiscal_name: str | None) -> bool:
|
|
if not mentions or not fiscal_name:
|
|
return False
|
|
|
|
nf = norm(fiscal_name)
|
|
fiscal_tokens = meaningful_tokens(fiscal_name)
|
|
|
|
for mention in mentions:
|
|
nm = norm(mention)
|
|
if not nm:
|
|
continue
|
|
if nm == nf:
|
|
return True
|
|
if len(nm) >= 4 and (nm in nf or nf in nm):
|
|
return True
|
|
mention_tokens = meaningful_tokens(mention)
|
|
if not mention_tokens or not fiscal_tokens:
|
|
continue
|
|
overlap = mention_tokens & fiscal_tokens
|
|
if len(overlap) >= 2:
|
|
return True
|
|
if len(mention_tokens) <= 2 and overlap:
|
|
return True
|
|
return False
|
|
|
|
|
|
def classify(identity: Dict[str, Any], fiscal_customer: str | None) -> str:
|
|
if identity.get("_error"):
|
|
return "EXTRACTION_ERROR"
|
|
if identity.get("_timeout"):
|
|
return "EXTRACTION_TIMEOUT"
|
|
|
|
mentions = identity.get("company_mentions") or []
|
|
domain = identity.get("domain") or ""
|
|
person = identity.get("person_name") or ""
|
|
|
|
if mentions and fiscal_customer:
|
|
if mentions_match_fiscal(mentions, fiscal_customer):
|
|
return "OK_MENTION_COMPATIBLE_WITH_FISCAL"
|
|
return "CONFLICT_MENTION_DIFFERS_FROM_FISCAL"
|
|
if mentions and not fiscal_customer:
|
|
return "OK_MENTION_AVAILABLE_NO_FISCAL"
|
|
if not mentions and fiscal_customer:
|
|
return "WEAK_NO_COMPANY_MENTION_HAS_FISCAL"
|
|
if domain and person:
|
|
return "WEAK_PERSON_AND_DOMAIN_ONLY"
|
|
if domain:
|
|
return "WEAK_DOMAIN_ONLY"
|
|
return "NO_USEFUL_IDENTITY"
|
|
|
|
|
|
def fetch_cases(limit: int, offset: int) -> List[Dict[str, Any]]:
|
|
with engine.begin() as conn:
|
|
rows = conn.execute(text("""
|
|
WITH ranked AS (
|
|
SELECT
|
|
o.id::text AS opportunity_id,
|
|
o.title AS opportunity_title,
|
|
o.customer_name,
|
|
o.customer_email,
|
|
c.name AS fiscal_customer,
|
|
c.tax_id AS fiscal_tax_id,
|
|
c.email AS fiscal_email,
|
|
t.id::text AS task_id,
|
|
t.action_code,
|
|
t.route,
|
|
t.created_at AS task_created_at,
|
|
m.id::text AS message_id,
|
|
COALESCE(NULLIF(m.clean_body, ''), NULLIF(m.raw_body, '')) AS body,
|
|
COALESCE(
|
|
NULLIF(m.metadata->>'subject', ''),
|
|
NULLIF(re.payload->'conversation'->'additional_attributes'->>'mail_subject', ''),
|
|
NULLIF(re.payload->'content_attributes'->'email'->>'subject', ''),
|
|
NULLIF(re.payload->'conversation'->'messages'->0->'content_attributes'->'email'->>'subject', '')
|
|
) AS subject,
|
|
COALESCE(
|
|
NULLIF(re.payload->'sender'->>'email', ''),
|
|
NULLIF(re.payload->'conversation'->'meta'->'sender'->>'email', ''),
|
|
NULLIF(re.payload->'conversation'->'contact_inbox'->>'source_id', ''),
|
|
NULLIF(o.customer_email, '')
|
|
) AS sender_email,
|
|
ROW_NUMBER() OVER (
|
|
PARTITION BY o.id
|
|
ORDER BY t.created_at DESC
|
|
) AS rn
|
|
FROM opportunities o
|
|
JOIN tasks t ON t.opportunity_id = o.id
|
|
LEFT JOIN messages m ON m.id = t.message_id
|
|
LEFT JOIN raw_events re ON re.id = t.raw_event_id
|
|
LEFT JOIN customers c ON c.id = o.local_customer_id
|
|
WHERE COALESCE(NULLIF(m.clean_body, ''), NULLIF(m.raw_body, '')) IS NOT NULL
|
|
)
|
|
SELECT *
|
|
FROM ranked
|
|
WHERE rn = 1
|
|
ORDER BY task_created_at DESC
|
|
LIMIT :limit
|
|
OFFSET :offset
|
|
"""), {"limit": limit, "offset": offset}).mappings().all()
|
|
return [dict(r) for r in rows]
|
|
|
|
|
|
def _domain_from_email(email: str) -> str:
|
|
if "@" not in (email or ""):
|
|
return ""
|
|
return email.rsplit("@", 1)[1].lower().strip()
|
|
|
|
|
|
def worker_extract(queue: mp.Queue, body: str, email: str, subject: str, use_llm: bool) -> None:
|
|
try:
|
|
from app.email_identity_extraction_service import extract_email_identity
|
|
|
|
identity = extract_email_identity(
|
|
body or "",
|
|
email=email or "",
|
|
subject=subject or "",
|
|
use_llm=use_llm,
|
|
)
|
|
queue.put({"ok": True, "identity": identity})
|
|
except Exception as exc: # noqa: BLE001 validation tool should keep going
|
|
queue.put({"ok": False, "error": f"{type(exc).__name__}: {exc}"})
|
|
|
|
|
|
def extract_with_timeout(body: str, email: str, subject: str, use_llm: bool, timeout_seconds: int) -> Dict[str, Any]:
|
|
if not use_llm:
|
|
from app.email_identity_extraction_service import extract_email_identity
|
|
return extract_email_identity(body or "", email=email or "", subject=subject or "", use_llm=False)
|
|
|
|
queue: mp.Queue = mp.Queue()
|
|
proc = mp.Process(target=worker_extract, args=(queue, body, email, subject, use_llm))
|
|
proc.start()
|
|
proc.join(timeout_seconds)
|
|
|
|
if proc.is_alive():
|
|
proc.terminate()
|
|
proc.join(5)
|
|
return {
|
|
"_timeout": True,
|
|
"method": "timeout",
|
|
"confidence": 0,
|
|
"email": email,
|
|
"domain": _domain_from_email(email),
|
|
"person_name": "",
|
|
"company_mentions": [],
|
|
"address": "",
|
|
"phones": [],
|
|
"websites": [],
|
|
"evidence": [f"timeout após {timeout_seconds}s"],
|
|
}
|
|
|
|
if queue.empty():
|
|
return {
|
|
"_error": True,
|
|
"method": "error",
|
|
"confidence": 0,
|
|
"email": email,
|
|
"domain": _domain_from_email(email),
|
|
"person_name": "",
|
|
"company_mentions": [],
|
|
"address": "",
|
|
"phones": [],
|
|
"websites": [],
|
|
"evidence": ["processo terminou sem resultado"],
|
|
}
|
|
|
|
result = queue.get()
|
|
if result.get("ok"):
|
|
return result["identity"]
|
|
|
|
return {
|
|
"_error": True,
|
|
"method": "error",
|
|
"confidence": 0,
|
|
"email": email,
|
|
"domain": _domain_from_email(email),
|
|
"person_name": "",
|
|
"company_mentions": [],
|
|
"address": "",
|
|
"phones": [],
|
|
"websites": [],
|
|
"evidence": [result.get("error")],
|
|
}
|
|
|
|
|
|
def main() -> None:
|
|
parser = argparse.ArgumentParser()
|
|
parser.add_argument("--limit", type=int, default=20)
|
|
parser.add_argument("--offset", type=int, default=0)
|
|
parser.add_argument("--use-llm", action="store_true")
|
|
parser.add_argument("--model", default="", help="Override EMAIL_IDENTITY_LLM_MODEL for this validation run")
|
|
parser.add_argument("--fallback-model", default="", help="Override EMAIL_IDENTITY_LLM_FALLBACK_MODEL for this validation run")
|
|
parser.add_argument("--timeout-seconds", type=int, default=30)
|
|
parser.add_argument("--max-body-chars", type=int, default=3500)
|
|
parser.add_argument("--out-dir", default="reports")
|
|
args = parser.parse_args()
|
|
if args.model:
|
|
os.environ["EMAIL_IDENTITY_LLM_MODEL"] = args.model
|
|
if args.fallback_model:
|
|
os.environ["EMAIL_IDENTITY_LLM_FALLBACK_MODEL"] = args.fallback_model
|
|
|
|
cases = fetch_cases(args.limit, args.offset)
|
|
out_dir = Path(args.out_dir)
|
|
out_dir.mkdir(parents=True, exist_ok=True)
|
|
|
|
stamp = datetime.now().strftime("%Y%m%d_%H%M%S")
|
|
mode = "llm" if args.use_llm else "regex"
|
|
jsonl_path = out_dir / f"email_identity_validation_{mode}_{stamp}.jsonl"
|
|
csv_path = out_dir / f"email_identity_validation_{mode}_{stamp}.csv"
|
|
|
|
results: List[Dict[str, Any]] = []
|
|
counts: Dict[str, int] = {}
|
|
|
|
model_label = os.getenv("EMAIL_IDENTITY_LLM_MODEL") or os.getenv("OPENROUTER_MODEL", "")
|
|
fallback_label = os.getenv("EMAIL_IDENTITY_LLM_FALLBACK_MODEL", "")
|
|
print(
|
|
f"Casos: {len(cases)} | mode={mode} | offset={args.offset} | "
|
|
f"timeout={args.timeout_seconds}s | max_body_chars={args.max_body_chars} | "
|
|
f"model={model_label if args.use_llm else '-'} | fallback={fallback_label if args.use_llm and fallback_label else '-'}",
|
|
flush=True,
|
|
)
|
|
print(f"JSONL incremental: {jsonl_path}", flush=True)
|
|
|
|
with jsonl_path.open("w", encoding="utf-8") as jf:
|
|
for idx, row in enumerate(cases, start=1):
|
|
email = row.get("sender_email") or row.get("customer_email") or ""
|
|
body = (row.get("body") or "")[: args.max_body_chars]
|
|
subject = row.get("subject") or ""
|
|
|
|
print(f"\n[{idx}/{len(cases)}] {row.get('opportunity_title')} | {email}", flush=True)
|
|
|
|
identity = extract_with_timeout(
|
|
body=body,
|
|
email=email,
|
|
subject=subject,
|
|
use_llm=args.use_llm,
|
|
timeout_seconds=args.timeout_seconds,
|
|
)
|
|
status = classify(identity, row.get("fiscal_customer"))
|
|
|
|
result = {
|
|
"status": status,
|
|
"opportunity_id": row.get("opportunity_id"),
|
|
"opportunity_title": row.get("opportunity_title"),
|
|
"task_id": row.get("task_id"),
|
|
"action_code": row.get("action_code"),
|
|
"sender_email": email,
|
|
"customer_name": row.get("customer_name"),
|
|
"customer_email": row.get("customer_email"),
|
|
"fiscal_customer": row.get("fiscal_customer"),
|
|
"fiscal_tax_id": row.get("fiscal_tax_id"),
|
|
"fiscal_email": row.get("fiscal_email"),
|
|
"person_name": identity.get("person_name"),
|
|
"company_mentions": identity.get("company_mentions") or [],
|
|
"domain": identity.get("domain"),
|
|
"address": identity.get("address"),
|
|
"phones": identity.get("phones") or [],
|
|
"confidence": identity.get("confidence"),
|
|
"method": identity.get("method"),
|
|
"llm_model": identity.get("llm_model"),
|
|
"fallback_used": identity.get("fallback_used"),
|
|
"evidence": identity.get("evidence") or [],
|
|
}
|
|
results.append(result)
|
|
counts[status] = counts.get(status, 0) + 1
|
|
jf.write(json.dumps(result, ensure_ascii=False, default=str) + "\n")
|
|
jf.flush()
|
|
|
|
print(
|
|
"status:", status,
|
|
"| person:", result["person_name"],
|
|
"| companies:", result["company_mentions"],
|
|
flush=True,
|
|
)
|
|
|
|
csv_fields = [
|
|
"status", "opportunity_id", "opportunity_title", "task_id", "action_code",
|
|
"sender_email", "customer_name", "customer_email", "fiscal_customer",
|
|
"fiscal_tax_id", "fiscal_email", "person_name", "company_mentions",
|
|
"domain", "address", "phones", "confidence", "method", "llm_model", "fallback_used", "evidence",
|
|
]
|
|
with csv_path.open("w", encoding="utf-8", newline="") as f:
|
|
writer = csv.DictWriter(f, fieldnames=csv_fields)
|
|
writer.writeheader()
|
|
for r in results:
|
|
csv_row = dict(r)
|
|
for key in ["company_mentions", "phones", "evidence"]:
|
|
csv_row[key] = " | ".join(str(x) for x in csv_row.get(key) or [])
|
|
writer.writerow(csv_row)
|
|
|
|
print("\nResumo:")
|
|
for status, count in sorted(counts.items()):
|
|
print(f"{status}: {count}")
|
|
|
|
print("\nFicheiros gerados:")
|
|
print(jsonl_path)
|
|
print(csv_path)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|