436 lines
17 KiB
Python
436 lines
17 KiB
Python
"""Cross-conversation company/opportunity linking helpers.
|
|
|
|
v4928.1.5.91: B2B customers often use separate inboxes for buying and
|
|
finance (e.g. geral@empresa.pt orders, financeiro@empresa.pt pays). A new
|
|
Chatwoot conversation must therefore not automatically create a new opportunity
|
|
when it references an existing document/process.
|
|
|
|
Rules implemented here:
|
|
- exact document references (ORC/FA/FT formats) are strong automatic evidence;
|
|
- a business domain is company evidence only, never enough when multiple open
|
|
opportunities exist;
|
|
- public/ISP domains are ignored for identity linking.
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
import re
|
|
from typing import Any, Dict, Iterable, List, Optional
|
|
|
|
from sqlalchemy import text
|
|
|
|
from app.db import engine
|
|
|
|
PUBLIC_EMAIL_DOMAINS = {
|
|
"gmail.com", "googlemail.com", "hotmail.com", "hotmail.pt", "outlook.com",
|
|
"outlook.pt", "live.com", "msn.com", "yahoo.com", "yahoo.pt", "icloud.com",
|
|
"me.com", "mac.com", "sapo.pt", "mail.telepac.pt", "netcabo.pt", "clix.pt",
|
|
"iol.pt", "vodafone.pt", "mail.com", "proton.me", "protonmail.com",
|
|
}
|
|
|
|
FINANCE_DEPARTMENT_PREFIXES = {
|
|
"financeiro", "financas", "finanças", "contabilidade", "accounts", "accounting",
|
|
"pagamentos", "payments", "billing", "facturacao", "faturacao", "faturação",
|
|
}
|
|
BUYING_DEPARTMENT_PREFIXES = {
|
|
"compras", "purchasing", "procurement", "encomendas", "orders", "geral", "info",
|
|
}
|
|
|
|
# Explicit stored formats: ORC.ORC2026.193, ORC2026.193, FA.FA2026.131, FT.FT2026.1
|
|
_EXPLICIT_DOC_RE = re.compile(
|
|
r"\b(?P<prefix>ORC|FA|FT|FR|NC|ND)[.\-/ ]?(?P=prefix)?\s*(?P<year>20\d{2})[.\-/ ]?0*(?P<num>\d{1,6})\b",
|
|
re.IGNORECASE,
|
|
)
|
|
|
|
# Human formats common in email subjects: Orçamento 2026/193, Fatura 2026/131.
|
|
_HUMAN_DOC_RE = re.compile(
|
|
r"\b(?P<label>or[cç]amento|orcamento|orc|cot[aá]?[cç][aã]o|proposta|fatura|factura|fat|invoice|ft|fa)\s+(?P<year>20\d{2})\s*/\s*0*(?P<num>\d{1,6})\b",
|
|
re.IGNORECASE,
|
|
)
|
|
|
|
PAYMENT_INTENT_RE = re.compile(
|
|
r"\b(comprovativo|pagamento|pago|transfer[êe]ncia|liquida[cç][aã]o|iban|recibo)\b",
|
|
re.IGNORECASE,
|
|
)
|
|
|
|
|
|
def _json(value: Any) -> str:
|
|
return json.dumps(value or {}, ensure_ascii=False, default=str)
|
|
|
|
|
|
def email_domain(email: object) -> str:
|
|
value = str(email or "").strip().lower()
|
|
if "@" not in value:
|
|
return ""
|
|
domain = value.rsplit("@", 1)[-1].strip(" .")
|
|
return domain
|
|
|
|
|
|
def email_local_part(email: object) -> str:
|
|
value = str(email or "").strip().lower()
|
|
if "@" not in value:
|
|
return ""
|
|
return value.split("@", 1)[0].strip()
|
|
|
|
|
|
def is_public_email_domain(domain_or_email: object) -> bool:
|
|
value = str(domain_or_email or "").strip().lower()
|
|
domain = email_domain(value) if "@" in value else value.strip(" .")
|
|
return bool(domain and domain in PUBLIC_EMAIL_DOMAINS)
|
|
|
|
|
|
def is_department_email(email: object) -> bool:
|
|
local = email_local_part(email)
|
|
if not local:
|
|
return False
|
|
root = re.split(r"[.\-_+0-9]", local, maxsplit=1)[0]
|
|
return root in FINANCE_DEPARTMENT_PREFIXES or root in BUYING_DEPARTMENT_PREFIXES
|
|
|
|
|
|
def _kind_for_prefix(prefix: str) -> str:
|
|
prefix = str(prefix or "").upper()
|
|
if prefix == "ORC":
|
|
return "quotation"
|
|
if prefix in {"FA", "FT"}:
|
|
return "invoice"
|
|
return "document"
|
|
|
|
|
|
def _prefix_for_label(label: str) -> str:
|
|
value = str(label or "").lower()
|
|
if any(term in value for term in ["fatura", "factura", "invoice", "fat", "ft", "fa"]):
|
|
return "FA"
|
|
return "ORC"
|
|
|
|
|
|
def normalize_document_reference(value: object) -> str:
|
|
"""Return canonical Jasmin-like document number when possible."""
|
|
text_value = str(value or "").strip()
|
|
if not text_value:
|
|
return ""
|
|
m = _EXPLICIT_DOC_RE.search(text_value)
|
|
if m:
|
|
prefix = m.group("prefix").upper()
|
|
year = m.group("year")
|
|
seq = int(m.group("num"))
|
|
return f"{prefix}.{prefix}{year}.{seq:03d}"
|
|
m = _HUMAN_DOC_RE.search(text_value)
|
|
if m:
|
|
prefix = _prefix_for_label(m.group("label"))
|
|
year = m.group("year")
|
|
seq = int(m.group("num"))
|
|
return f"{prefix}.{prefix}{year}.{seq:03d}"
|
|
return text_value
|
|
|
|
|
|
def document_reference_variants(document_number: object) -> List[str]:
|
|
normalized = normalize_document_reference(document_number)
|
|
if not normalized:
|
|
return []
|
|
variants: List[str] = [normalized]
|
|
m = re.match(r"^(?P<prefix>ORC|FA|FT|FR|NC|ND)\.(?P=prefix)(?P<year>20\d{2})\.(?P<num>\d{1,6})$", normalized, re.I)
|
|
if m:
|
|
prefix = m.group("prefix").upper()
|
|
year = m.group("year")
|
|
seq = int(m.group("num"))
|
|
label = "Orçamento" if prefix == "ORC" else "Fatura"
|
|
variants.extend([
|
|
f"{prefix}{year}.{seq:03d}",
|
|
f"{prefix} {year}/{seq}",
|
|
f"{label} {year}/{seq}",
|
|
f"{year}/{seq}",
|
|
f"{year[-2:]}/{seq}",
|
|
])
|
|
out: List[str] = []
|
|
for v in variants:
|
|
if v and v not in out:
|
|
out.append(v)
|
|
return out
|
|
|
|
|
|
def extract_document_references(*values: object) -> List[str]:
|
|
joined = "\n".join(str(v or "") for v in values if v is not None)
|
|
refs: List[str] = []
|
|
for regex in (_EXPLICIT_DOC_RE, _HUMAN_DOC_RE):
|
|
for m in regex.finditer(joined):
|
|
ref = normalize_document_reference(m.group(0))
|
|
if ref and ref not in refs:
|
|
refs.append(ref)
|
|
return refs
|
|
|
|
|
|
def task_text_blob(task: Dict[str, Any]) -> str:
|
|
return "\n".join(
|
|
str(task.get(k) or "")
|
|
for k in ("message_subject", "request_text", "note", "action", "customer_name", "customer_email")
|
|
)
|
|
|
|
|
|
def task_document_references(task: Dict[str, Any]) -> List[str]:
|
|
return extract_document_references(task_text_blob(task))
|
|
|
|
|
|
def is_payment_intent_task(task: Dict[str, Any]) -> bool:
|
|
code = str(task.get("action_code") or "").strip().upper()
|
|
if code in {"CONFIRM_PAYMENT", "FOLLOW_UP_PAYMENT", "SEND_PROFORMA", "SEND_INVOICE"}:
|
|
return True
|
|
return bool(PAYMENT_INTENT_RE.search(task_text_blob(task)))
|
|
|
|
|
|
def _unique_opportunities(rows: Iterable[Dict[str, Any]]) -> List[Dict[str, Any]]:
|
|
out: List[Dict[str, Any]] = []
|
|
seen = set()
|
|
for row in rows:
|
|
opp_id = str(row.get("id") or row.get("opportunity_id") or "")
|
|
if not opp_id or opp_id in seen:
|
|
continue
|
|
seen.add(opp_id)
|
|
out.append(dict(row))
|
|
return out
|
|
|
|
|
|
def _candidate_payload(row: Dict[str, Any], *, reason: str) -> Dict[str, Any]:
|
|
return {
|
|
"opportunity_id": str(row.get("id") or row.get("opportunity_id") or ""),
|
|
"title": str(row.get("title") or row.get("opportunity_title") or ""),
|
|
"stage": str(row.get("stage") or ""),
|
|
"customer_name": str(row.get("customer_name") or ""),
|
|
"customer_email": str(row.get("customer_email") or ""),
|
|
"document_number": str(row.get("document_number") or ""),
|
|
"document_kind": str(row.get("document_kind") or ""),
|
|
"total_amount": str(row.get("total_amount") or row.get("value_amount") or ""),
|
|
"conversation_id": str(row.get("conversation_id") or ""),
|
|
"reason": reason,
|
|
}
|
|
|
|
|
|
def find_open_opportunities_by_document_references(refs: List[str]) -> List[Dict[str, Any]]:
|
|
refs = [normalize_document_reference(r) for r in refs if normalize_document_reference(r)]
|
|
if not refs:
|
|
return []
|
|
with engine.begin() as conn:
|
|
rows = conn.execute(text("""
|
|
WITH refs AS (
|
|
SELECT unnest(CAST(:refs AS TEXT[])) AS ref
|
|
), doc_matches AS (
|
|
SELECT
|
|
o.*, d.document_number, d.document_kind, d.total_amount, d.amount,
|
|
'commercial_document_reference'::text AS link_reason,
|
|
d.updated_at AS match_updated_at
|
|
FROM commercial_documents d
|
|
JOIN opportunities o ON o.id = d.opportunity_id
|
|
JOIN refs r ON upper(COALESCE(d.document_number, '')) = upper(r.ref)
|
|
WHERE o.status = 'open'
|
|
AND COALESCE(d.is_active, TRUE) = TRUE
|
|
AND d.opportunity_id IS NOT NULL
|
|
), recon_matches AS (
|
|
SELECT
|
|
o.*, ri.document_number, ri.external_type AS document_kind, ri.amount AS total_amount, ri.amount,
|
|
'reconciliation_document_reference'::text AS link_reason,
|
|
ri.updated_at AS match_updated_at
|
|
FROM reconciliation_items ri
|
|
JOIN opportunities o ON o.id = ri.opportunity_id
|
|
JOIN refs r ON upper(COALESCE(ri.document_number, '')) = upper(r.ref)
|
|
WHERE o.status = 'open'
|
|
AND ri.opportunity_id IS NOT NULL
|
|
)
|
|
SELECT * FROM doc_matches
|
|
UNION ALL
|
|
SELECT * FROM recon_matches
|
|
ORDER BY match_updated_at DESC NULLS LAST
|
|
LIMIT 20
|
|
"""), {"refs": refs}).mappings().all()
|
|
return _unique_opportunities([dict(r) for r in rows])
|
|
|
|
|
|
def find_document_owner_opportunity_for_task(task: Dict[str, Any]) -> Optional[Dict[str, Any]]:
|
|
refs = task_document_references(task)
|
|
if not refs:
|
|
return None
|
|
matches = find_open_opportunities_by_document_references(refs)
|
|
if len(matches) != 1:
|
|
return None
|
|
result = dict(matches[0])
|
|
result["_link_match_reason"] = "document_reference"
|
|
result["_document_reference_matches"] = refs
|
|
return result
|
|
|
|
|
|
def company_domain_for_task(task: Dict[str, Any]) -> str:
|
|
for key in ("customer_email", "opportunity_customer_email", "linked_customer_email"):
|
|
domain = email_domain(task.get(key))
|
|
if domain and not is_public_email_domain(domain):
|
|
return domain
|
|
return ""
|
|
|
|
|
|
def find_open_company_domain_candidates(task: Dict[str, Any], *, recent_days: int = 45, limit: int = 8) -> List[Dict[str, Any]]:
|
|
domain = company_domain_for_task(task)
|
|
if not domain:
|
|
return []
|
|
with engine.begin() as conn:
|
|
rows = conn.execute(text("""
|
|
SELECT DISTINCT ON (o.id)
|
|
o.id::text AS opportunity_id,
|
|
o.id,
|
|
o.title,
|
|
o.stage,
|
|
o.status,
|
|
o.customer_name,
|
|
COALESCE(NULLIF(o.customer_email, ''), NULLIF(c.email, '')) AS customer_email,
|
|
o.conversation_id,
|
|
o.value_amount,
|
|
d.document_number,
|
|
d.document_kind,
|
|
d.total_amount,
|
|
o.updated_at
|
|
FROM opportunities o
|
|
LEFT JOIN customers c ON c.id = o.local_customer_id
|
|
LEFT JOIN commercial_documents d ON d.opportunity_id = o.id AND COALESCE(d.is_active, TRUE) = TRUE
|
|
WHERE o.status = 'open'
|
|
AND o.updated_at >= now() - make_interval(days => :recent_days)
|
|
AND (
|
|
lower(COALESCE(o.customer_email, '')) LIKE :domain_like
|
|
OR lower(COALESCE(c.email, '')) LIKE :domain_like
|
|
OR lower(COALESCE(o.metadata::text, '')) LIKE :domain_text_like
|
|
)
|
|
ORDER BY o.id, d.updated_at DESC NULLS LAST, o.updated_at DESC
|
|
LIMIT :limit
|
|
"""), {
|
|
"domain_like": f"%@{domain}",
|
|
"domain_text_like": f"%{domain}%",
|
|
"recent_days": int(recent_days),
|
|
"limit": int(limit),
|
|
}).mappings().all()
|
|
return [dict(r) for r in rows]
|
|
|
|
|
|
def find_unique_company_domain_opportunity_for_task(task: Dict[str, Any]) -> Optional[Dict[str, Any]]:
|
|
"""Auto-link only when the domain match is unique and the task is process-like.
|
|
|
|
This is deliberately conservative: public domains are ignored, and if more
|
|
than one opportunity exists for the same business domain the operator must
|
|
choose.
|
|
"""
|
|
if not is_payment_intent_task(task) and not is_department_email(task.get("customer_email")):
|
|
return None
|
|
candidates = find_open_company_domain_candidates(task, limit=3)
|
|
if len(candidates) != 1:
|
|
return None
|
|
result = dict(candidates[0])
|
|
result["_link_match_reason"] = "company_domain_unique_recent"
|
|
result["_company_domain"] = company_domain_for_task(task)
|
|
return result
|
|
|
|
|
|
def manual_link_candidates_for_task(task: Dict[str, Any]) -> List[Dict[str, Any]]:
|
|
"""Return candidates to show to the operator when automatic linking is unsafe."""
|
|
rows: List[Dict[str, Any]] = []
|
|
refs = task_document_references(task)
|
|
if refs:
|
|
rows.extend(find_open_opportunities_by_document_references(refs))
|
|
rows.extend(find_open_company_domain_candidates(task, limit=8))
|
|
out: List[Dict[str, Any]] = []
|
|
seen = set()
|
|
for row in rows:
|
|
opp_id = str(row.get("id") or row.get("opportunity_id") or "")
|
|
if not opp_id or opp_id in seen:
|
|
continue
|
|
seen.add(opp_id)
|
|
reason = "referência documental" if str(row.get("document_number") or "") else "mesmo domínio empresarial"
|
|
out.append(_candidate_payload(row, reason=reason))
|
|
return out
|
|
|
|
|
|
def mark_task_ambiguous_with_candidates(task_id: str, *, reason: str, candidates: List[Dict[str, Any]], created_by: str = "system") -> None:
|
|
metadata = {
|
|
"opportunity_linking_status": "ambiguous",
|
|
"opportunity_linking_reason": reason,
|
|
"opportunity_linking_candidates": candidates[:8],
|
|
"opportunity_linking_checked_by": created_by,
|
|
}
|
|
with engine.begin() as conn:
|
|
conn.execute(text("""
|
|
UPDATE tasks
|
|
SET route = CASE WHEN status = 'pending' THEN 'rever' ELSE route END,
|
|
metadata = COALESCE(metadata, '{}'::jsonb) || CAST(:metadata AS JSONB),
|
|
updated_at = now()
|
|
WHERE id = CAST(:task_id AS UUID)
|
|
"""), {"task_id": task_id, "metadata": _json(metadata)})
|
|
|
|
|
|
def associate_task_to_opportunity(task_id: str, opportunity_id: str, *, actor: str = "operator") -> Dict[str, int]:
|
|
"""Manual operator action: attach a task/conversation evidence to an existing opportunity."""
|
|
metadata = {
|
|
"opportunity_linking_status": "resolved",
|
|
"opportunity_linking_resolved_by": actor,
|
|
"opportunity_linking_resolved_reason": "operator_associated_existing_opportunity",
|
|
"opportunity_id": str(opportunity_id),
|
|
}
|
|
with engine.begin() as conn:
|
|
task = conn.execute(text("""
|
|
SELECT id::text, communication_id::text, message_id::text, conversation_id, contact_id, metadata
|
|
FROM tasks
|
|
WHERE id = CAST(:task_id AS UUID)
|
|
LIMIT 1
|
|
"""), {"task_id": task_id}).mappings().first()
|
|
if not task:
|
|
raise ValueError(f"task_not_found: {task_id}")
|
|
|
|
opp = conn.execute(text("""
|
|
SELECT id::text, title, conversation_id
|
|
FROM opportunities
|
|
WHERE id = CAST(:opportunity_id AS UUID)
|
|
LIMIT 1
|
|
"""), {"opportunity_id": opportunity_id}).mappings().first()
|
|
if not opp:
|
|
raise ValueError(f"opportunity_not_found: {opportunity_id}")
|
|
|
|
task_result = conn.execute(text("""
|
|
UPDATE tasks
|
|
SET opportunity_id = CAST(:opportunity_id AS UUID),
|
|
metadata = COALESCE(metadata, '{}'::jsonb) || CAST(:metadata AS JSONB),
|
|
updated_at = now()
|
|
WHERE id = CAST(:task_id AS UUID)
|
|
"""), {"task_id": task_id, "opportunity_id": opportunity_id, "metadata": _json(metadata)})
|
|
|
|
comm_result = conn.execute(text("""
|
|
UPDATE communications
|
|
SET opportunity_id = CAST(:opportunity_id AS UUID),
|
|
metadata = COALESCE(metadata, '{}'::jsonb) || CAST(:metadata AS JSONB),
|
|
updated_at = now()
|
|
WHERE task_id = CAST(:task_id AS UUID)
|
|
OR id = NULLIF(:communication_id, '')::uuid
|
|
OR source_message_id = NULLIF(:message_id, '')
|
|
OR (conversation_id = NULLIF(:conversation_id, '') AND opportunity_id IS NULL)
|
|
"""), {
|
|
"task_id": task_id,
|
|
"opportunity_id": opportunity_id,
|
|
"communication_id": task.get("communication_id") or "",
|
|
"message_id": task.get("message_id") or "",
|
|
"conversation_id": task.get("conversation_id") or "",
|
|
"metadata": _json(metadata),
|
|
})
|
|
|
|
conn.execute(text("""
|
|
UPDATE opportunities
|
|
SET metadata = COALESCE(metadata, '{}'::jsonb) || CAST(:metadata AS JSONB),
|
|
updated_at = now(),
|
|
last_task_id = CAST(:task_id AS UUID)
|
|
WHERE id = CAST(:opportunity_id AS UUID)
|
|
"""), {
|
|
"opportunity_id": opportunity_id,
|
|
"task_id": task_id,
|
|
"metadata": _json({"last_manual_task_association": metadata}),
|
|
})
|
|
|
|
conn.execute(text("""
|
|
INSERT INTO opportunity_events (id, opportunity_id, event_type, task_id, action_code, note, payload, created_by)
|
|
SELECT gen_random_uuid(), CAST(:opportunity_id AS UUID), 'manual_task_association', CAST(:task_id AS UUID),
|
|
action_code, 'Operador associou tarefa/conversa a oportunidade existente.', CAST(:metadata AS JSONB), :actor
|
|
FROM tasks
|
|
WHERE id = CAST(:task_id AS UUID)
|
|
"""), {"opportunity_id": opportunity_id, "task_id": task_id, "metadata": _json(metadata), "actor": actor})
|
|
|
|
return {"tasks": int(task_result.rowcount or 0), "communications": int(comm_result.rowcount or 0)}
|