302 lines
12 KiB
Python
302 lines
12 KiB
Python
"""Recipient and greeting helpers for customer reply drafts.
|
|
|
|
The reply assistant must greet the person who wrote the latest message, not the
|
|
fiscal company attached to the opportunity. These helpers deliberately use
|
|
small deterministic rules before the LLM sees the context, so the UI can keep a
|
|
stable, auditable greeting.
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import re
|
|
from typing import Any, Dict, Iterable, Tuple
|
|
|
|
|
|
COMPANY_MARKERS = {
|
|
"lda", "l.da", "ltd", "sa", "s.a", "unipessoal", "limitada", "empresa",
|
|
"engenharia", "condomínio", "condominio", "sociedade", "grupo",
|
|
}
|
|
FEMALE_FIRST_NAMES = {
|
|
"alexandra", "ana", "barbara", "bárbara", "beatriz", "catarina", "claudia", "cláudia",
|
|
"cristina", "daniela", "filipa", "ines", "inês", "joana", "maria", "mariana", "marta",
|
|
"patricia", "patrícia", "rita", "sara", "sofia", "susana", "teresa", "vera",
|
|
}
|
|
MALE_FIRST_NAMES = {
|
|
"antonio", "antónio", "bruno", "carlos", "diogo", "duarte", "fernando", "francisco",
|
|
"joao", "joão", "jose", "josé", "luis", "luís", "manuel", "miguel", "nuno",
|
|
"paulo", "pedro", "ricardo", "rui", "sergio", "sérgio", "tiago", "vasco",
|
|
}
|
|
EMAIL_LOCAL_GENERIC = {
|
|
"admin", "administracao", "administração", "apoio", "atendimento", "billing", "comercial",
|
|
"compras", "contact", "contacto", "contabilidade", "email", "encomendas", "escritorio",
|
|
"escritório", "faturacao", "faturação", "financeiro", "geral", "hello", "info",
|
|
"mail", "marketing", "noreply", "no-reply", "office", "orders", "postmaster", "rh",
|
|
"sales", "secretaria", "suporte", "support", "vendas",
|
|
}
|
|
FIRST_NAME_CANONICAL = {
|
|
"antonio": "António", "antónio": "António", "barbara": "Bárbara", "bárbara": "Bárbara",
|
|
"claudia": "Cláudia", "cláudia": "Cláudia", "ines": "Inês", "inês": "Inês",
|
|
"joao": "João", "joão": "João", "jose": "José", "josé": "José", "luis": "Luís",
|
|
"luís": "Luís", "maria": "Maria", "nuno": "Nuno", "patricia": "Patrícia",
|
|
"patrícia": "Patrícia", "sergio": "Sérgio", "sérgio": "Sérgio",
|
|
}
|
|
VALEDICTION_RE = re.compile(
|
|
r"^(?:com os melhores cumprimentos|melhores cumprimentos|cumprimentos|cordiais cumprimentos|obrigad[ao]s?|atenciosamente|best regards|regards)\b",
|
|
flags=re.I,
|
|
)
|
|
BAD_PERSON_RE = re.compile(
|
|
r"^(?:enviado do meu|sent from my|assinado por|aviso|disclaimer|cid:|image|logo|telefone|telem|tel\.?|email|e-mail|website|www\.|http|rua|avenida|morada)\b",
|
|
flags=re.I,
|
|
)
|
|
EMAIL_RE = re.compile(r"[A-Z0-9._%+\-]+@[A-Z0-9.\-]+\.[A-Z]{2,}", flags=re.I)
|
|
PHONE_RE = re.compile(r"(?:\+351\s*)?(?:\d[\s.-]?){9,}")
|
|
POSTCODE_RE = re.compile(r"\b\d{4}-\d{3}\b")
|
|
|
|
|
|
def _clean(value: Any) -> str:
|
|
return re.sub(r"\s+", " ", str(value or "")).strip()
|
|
|
|
|
|
def _candidate_texts(task_or_payload: Dict[str, Any]) -> Iterable[str]:
|
|
data = task_or_payload or {}
|
|
metadata = data.get("metadata") if isinstance(data.get("metadata"), dict) else {}
|
|
message_metadata = data.get("message_metadata") if isinstance(data.get("message_metadata"), dict) else {}
|
|
for candidate in [
|
|
data.get("request_text"),
|
|
data.get("raw_body"),
|
|
data.get("clean_body"),
|
|
metadata.get("request_text"),
|
|
metadata.get("customer_message"),
|
|
metadata.get("raw_body"),
|
|
message_metadata.get("raw_body"),
|
|
message_metadata.get("clean_body"),
|
|
data.get("note"),
|
|
]:
|
|
text = str(candidate or "").replace("\r\n", "\n").replace("\r", "\n").strip()
|
|
if text:
|
|
yield text
|
|
|
|
|
|
def _cut_quoted_history(text: str) -> str:
|
|
markers = [
|
|
"\nDe:", "\nDe ", "\nEnviada:", "\nEnviado:", "\nAssunto:", "\nPara:", "\nCc:",
|
|
"\nOn ", "\nFrom:", "\nSent:", "\nSubject:", "\nTo:",
|
|
"\n-----Original Message-----", "\n------ Mensagem original ------", "\n-----Mensagem original-----",
|
|
"\n________________________________", "\n[Quoted text hidden]",
|
|
]
|
|
cut_at = len(text)
|
|
for marker in markers:
|
|
idx = text.find(marker)
|
|
if idx != -1:
|
|
cut_at = min(cut_at, idx)
|
|
return text[:cut_at].strip()
|
|
|
|
|
|
def raw_latest_customer_text(task_or_payload: Dict[str, Any]) -> str:
|
|
"""Return latest customer text preserving signature when available."""
|
|
for text in _candidate_texts(task_or_payload):
|
|
raw = _cut_quoted_history(text)
|
|
if raw:
|
|
return raw
|
|
return ""
|
|
|
|
|
|
def looks_like_company_name(value: Any) -> bool:
|
|
name = _clean(value).lower().replace(",", " ").replace(".", " ")
|
|
tokens = set(name.split())
|
|
if tokens & COMPANY_MARKERS:
|
|
return True
|
|
if re.search(r"\b(?:l\s*d\s*a|s\s*a|unipessoal|limitada)\b", name, flags=re.I):
|
|
return True
|
|
return False
|
|
|
|
|
|
def _sanitize_person_name(value: Any) -> str:
|
|
v = _clean(value).strip(" ,;:-")
|
|
if not v or "@" in v or any(ch.isdigit() for ch in v):
|
|
return ""
|
|
if EMAIL_RE.search(v) or PHONE_RE.search(v) or POSTCODE_RE.search(v):
|
|
return ""
|
|
if BAD_PERSON_RE.search(v) or VALEDICTION_RE.search(v):
|
|
return ""
|
|
if looks_like_company_name(v):
|
|
return ""
|
|
# Remove common trailing punctuation but preserve Portuguese names.
|
|
v = re.sub(r"\s*[|•].*$", "", v).strip(" ,;:-")
|
|
if len(v) > 80:
|
|
return ""
|
|
return v
|
|
|
|
|
|
def looks_like_person_name(value: Any) -> bool:
|
|
v = _sanitize_person_name(value)
|
|
if not v:
|
|
return False
|
|
words = v.split()
|
|
if not (2 <= len(words) <= 5):
|
|
return False
|
|
capitalized = [
|
|
w for w in words
|
|
if re.match(r"^[A-ZÁÀÂÃÉÈÊÍÌÎÓÒÔÕÚÙÛÇ][A-Za-zÁÀÂÃÉÈÊÍÌÎÓÒÔÕÚÙÛÇáàâãéèêíìîóòôõúùûç'.-]+$", w)
|
|
]
|
|
return len(capitalized) >= 2
|
|
|
|
|
|
def extract_signature_person_name(text: str) -> str:
|
|
"""Extract a personal name immediately after a sign-off in the latest email."""
|
|
raw = _cut_quoted_history(str(text or ""))
|
|
lines = [_clean(line) for line in raw.splitlines() if _clean(line)]
|
|
for idx, line in enumerate(lines[:45]):
|
|
if VALEDICTION_RE.search(line):
|
|
for candidate in lines[idx + 1: idx + 8]:
|
|
if looks_like_person_name(candidate):
|
|
return _sanitize_person_name(candidate)
|
|
break
|
|
return ""
|
|
|
|
|
|
def _base_greeting(text: str) -> str:
|
|
low = str(text or "").lower()
|
|
if "boa tarde" in low[:200]:
|
|
return "Boa tarde"
|
|
if "boa noite" in low[:200]:
|
|
return "Boa noite"
|
|
return "Bom dia"
|
|
|
|
|
|
def _canonical_name_token(token: str) -> str:
|
|
token = re.sub(r"[^A-Za-zÁÀÂÃÉÈÊÍÌÎÓÒÔÕÚÙÛÇáàâãéèêíìîóòôõúùûç'-]", "", str(token or "").strip().lower())
|
|
if not token:
|
|
return ""
|
|
return FIRST_NAME_CANONICAL.get(token, token[:1].upper() + token[1:])
|
|
|
|
|
|
def _first_token(value: str) -> str:
|
|
parts = [p for p in re.split(r"\s+", str(value or "").strip()) if p]
|
|
return parts[0] if parts else ""
|
|
|
|
|
|
def infer_person_name_from_email(email: Any) -> Tuple[str, str]:
|
|
"""Infer a person name from a personal email address, never from generic inboxes.
|
|
|
|
Returns (full_name, confidence). Confidence is "alto" only when the local
|
|
part starts with a known Portuguese first name or with a two-token
|
|
first.last pattern where the first token is a known first name.
|
|
"""
|
|
raw = _clean(email).lower()
|
|
if not raw or "@" not in raw:
|
|
return "", ""
|
|
local = raw.split("@", 1)[0].strip(" ._-")
|
|
if not local or local in EMAIL_LOCAL_GENERIC:
|
|
return "", ""
|
|
local = re.sub(r"\+.*$", "", local)
|
|
tokens = [t for t in re.split(r"[._\-]+", local) if t and not t.isdigit()]
|
|
if not tokens or tokens[0] in EMAIL_LOCAL_GENERIC:
|
|
return "", ""
|
|
first_lower = tokens[0].lower()
|
|
known_first = first_lower in MALE_FIRST_NAMES or first_lower in FEMALE_FIRST_NAMES or first_lower in FIRST_NAME_CANONICAL
|
|
if not known_first:
|
|
return "", ""
|
|
clean_tokens = [_canonical_name_token(t) for t in tokens[:3]]
|
|
clean_tokens = [t for t in clean_tokens if t]
|
|
if not clean_tokens:
|
|
return "", ""
|
|
if len(clean_tokens) == 1:
|
|
return clean_tokens[0], "medio"
|
|
return " ".join(clean_tokens), "alto"
|
|
|
|
|
|
def preferred_greeting_for(person_name: str, customer_text: str, *, first_name_only: bool = False) -> str:
|
|
base = _base_greeting(customer_text)
|
|
raw_name = _clean(person_name)
|
|
first_display = _first_token(raw_name)
|
|
first_key = first_display.lower()
|
|
if not looks_like_person_name(raw_name):
|
|
if first_name_only and first_key in FEMALE_FIRST_NAMES:
|
|
return f"{base} Sra. {first_display},"
|
|
if first_name_only and first_key in MALE_FIRST_NAMES:
|
|
return f"{base} Sr. {first_display},"
|
|
return f"{base},"
|
|
greeting_name = first_display if first_name_only and first_display else raw_name
|
|
if first_key in FEMALE_FIRST_NAMES:
|
|
return f"{base} Sra. {greeting_name},"
|
|
if first_key in MALE_FIRST_NAMES:
|
|
return f"{base} Sr. {greeting_name},"
|
|
return f"{base} {greeting_name},"
|
|
|
|
|
|
def resolve_reply_recipient(task: Dict[str, Any], *, cleaned_customer_message: str = "") -> Dict[str, str]:
|
|
"""Resolve person/company/greeting for a reply draft.
|
|
|
|
Priority for greeting:
|
|
1. person name from latest email signature;
|
|
2. CRM/Chatwoot contact name only if it looks like a person;
|
|
3. generic greeting. Fiscal/company names are kept as company context but
|
|
never used as the greeting recipient.
|
|
"""
|
|
raw_text = raw_latest_customer_text(task)
|
|
signature_name = extract_signature_person_name(raw_text)
|
|
contact_name = _clean(task.get("customer_name") or task.get("contact_name") or "")
|
|
fiscal_name = _clean(task.get("linked_customer_name") or task.get("fiscal_customer_name") or "")
|
|
|
|
email_value = _clean(
|
|
task.get("customer_email")
|
|
or task.get("opportunity_customer_email")
|
|
or task.get("linked_customer_email")
|
|
or task.get("email")
|
|
or ""
|
|
)
|
|
email_name, email_confidence = infer_person_name_from_email(email_value)
|
|
|
|
if signature_name:
|
|
person_name = signature_name
|
|
source = "signature"
|
|
confidence = "alto"
|
|
first_name_only = False
|
|
elif looks_like_person_name(contact_name):
|
|
person_name = contact_name
|
|
source = "contact"
|
|
confidence = "alto"
|
|
first_name_only = False
|
|
elif email_name:
|
|
person_name = email_name
|
|
source = "email"
|
|
confidence = email_confidence or "medio"
|
|
first_name_only = True
|
|
else:
|
|
person_name = ""
|
|
source = "generic"
|
|
confidence = ""
|
|
first_name_only = False
|
|
|
|
company_name = fiscal_name or (contact_name if looks_like_company_name(contact_name) else "")
|
|
greeting_text = raw_text or cleaned_customer_message
|
|
preferred = preferred_greeting_for(person_name, greeting_text, first_name_only=first_name_only)
|
|
return {
|
|
"person_name": person_name,
|
|
"person_first_name": _first_token(person_name),
|
|
"person_confidence": confidence,
|
|
"company_name": company_name,
|
|
"contact_name": contact_name,
|
|
"fiscal_name": fiscal_name,
|
|
"email": email_value,
|
|
"preferred_greeting": preferred,
|
|
"greeting_source": source,
|
|
"raw_customer_text": raw_text,
|
|
}
|
|
|
|
|
|
def apply_preferred_greeting(message_body: str, preferred_greeting: str) -> str:
|
|
message_body = str(message_body or "").strip()
|
|
preferred_greeting = str(preferred_greeting or "").strip()
|
|
if not message_body or not preferred_greeting:
|
|
return message_body
|
|
lines = message_body.splitlines()
|
|
if not lines:
|
|
return message_body
|
|
first = lines[0].strip()
|
|
known_prefixes = ("Olá", "Bom dia", "Boa tarde", "Boa noite", "Exmo", "Exma", "Caro", "Cara")
|
|
if first.startswith(known_prefixes):
|
|
lines[0] = preferred_greeting
|
|
return "\n".join(lines).strip()
|
|
return f"{preferred_greeting}\n\n{message_body}".strip()
|