153 lines
4.9 KiB
Python
153 lines
4.9 KiB
Python
"""Utilities to extract the useful customer message from email/chat text.
|
|
|
|
This cleaner is intentionally conservative: it removes quoted history,
|
|
transport headers and common signatures before either deterministic guardrails
|
|
or the LLM see the text. The goal is not to classify by rules; it is to avoid
|
|
feeding the assistant with footers, legal disclaimers and old messages.
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import re
|
|
from typing import Any, Dict
|
|
|
|
|
|
def get_nested(data: Dict[str, Any], *keys: str) -> Any:
|
|
current: Any = data
|
|
|
|
for key in keys:
|
|
if not isinstance(current, dict):
|
|
return None
|
|
current = current.get(key)
|
|
|
|
return current
|
|
|
|
|
|
def _cut_quoted_history(text: str) -> str:
|
|
cut_markers = [
|
|
"\nDe:",
|
|
"\nDe ",
|
|
"\nEnviada:",
|
|
"\nEnviado:",
|
|
"\nAssunto:",
|
|
"\nPara:",
|
|
"\nCc:",
|
|
"\nÀs ",
|
|
"\nA ",
|
|
"\nOn ",
|
|
"\nFrom:",
|
|
"\nSent:",
|
|
"\nSubject:",
|
|
"\nTo:",
|
|
"\nCc:",
|
|
"\n-----Original Message-----",
|
|
"\n------ Mensagem original ------",
|
|
"\n-----Mensagem original-----",
|
|
"\n________________________________",
|
|
"\n---------------------------------------------------------------",
|
|
"\n[Quoted text hidden]",
|
|
]
|
|
cut_at = len(text)
|
|
for marker in cut_markers:
|
|
idx = text.find(marker)
|
|
if idx != -1:
|
|
cut_at = min(cut_at, idx)
|
|
return text[:cut_at]
|
|
|
|
|
|
def _looks_like_signature_start(line: str) -> bool:
|
|
normalized = re.sub(r"\s+", " ", line.strip().lower())
|
|
if not normalized:
|
|
return False
|
|
starts = (
|
|
"cumprimentos",
|
|
"com os melhores cumprimentos",
|
|
"melhores cumprimentos",
|
|
"best regards",
|
|
"regards",
|
|
"atenciosamente",
|
|
"cordiais cumprimentos",
|
|
"cmpt",
|
|
"cps",
|
|
"enviado do meu",
|
|
"sent from",
|
|
)
|
|
return any(normalized.startswith(item) for item in starts)
|
|
|
|
|
|
def _drop_signature_and_disclaimers(text: str) -> str:
|
|
lines = text.split("\n")
|
|
kept: list[str] = []
|
|
non_empty = 0
|
|
for line in lines:
|
|
stripped = line.strip()
|
|
low = stripped.lower()
|
|
if stripped.startswith(">"):
|
|
continue
|
|
# Keep greetings at the top, but stop when the actual signature starts
|
|
# after at least one meaningful line.
|
|
if non_empty >= 1 and _looks_like_signature_start(stripped):
|
|
break
|
|
if any(marker in low for marker in [
|
|
"aviso de confidencialidade",
|
|
"esta mensagem pode conter informação confidencial",
|
|
"this message may contain confidential",
|
|
"se recebeu esta mensagem por engano",
|
|
"before you print",
|
|
"por favor tenha em atenção a sua responsabilidade ambiental",
|
|
"por favor considere o impacto",
|
|
"chamada para rede fixa nacional",
|
|
"chamada para a rede móvel nacional",
|
|
]):
|
|
break
|
|
kept.append(line)
|
|
if stripped:
|
|
non_empty += 1
|
|
return "\n".join(kept).strip()
|
|
|
|
|
|
def clean_email_reply_text(content: str) -> str:
|
|
text = str(content or "").replace("\r\n", "\n").replace("\r", "\n").strip()
|
|
text = _cut_quoted_history(text).strip()
|
|
text = _drop_signature_and_disclaimers(text).strip()
|
|
# Collapse excessive blank lines but preserve paragraph breaks.
|
|
text = re.sub(r"\n{3,}", "\n\n", text)
|
|
return text.strip()
|
|
|
|
|
|
def extract_customer_reply_text(task_or_payload: Dict[str, Any]) -> str:
|
|
"""Return the best available customer-authored text for reply generation."""
|
|
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 {}
|
|
candidates = [
|
|
data.get("request_text"),
|
|
data.get("clean_body"),
|
|
data.get("raw_body"),
|
|
metadata.get("request_text"),
|
|
metadata.get("customer_message"),
|
|
message_metadata.get("clean_body"),
|
|
message_metadata.get("raw_body"),
|
|
data.get("message_subject"),
|
|
data.get("note"),
|
|
]
|
|
for candidate in candidates:
|
|
cleaned = clean_email_reply_text(str(candidate or ""))
|
|
if cleaned:
|
|
return cleaned
|
|
return ""
|
|
|
|
|
|
def extract_chatwoot_content(payload: Dict[str, Any], message: Dict[str, Any]) -> str:
|
|
content = (
|
|
get_nested(message, "content_attributes", "email", "html_content", "reply")
|
|
or get_nested(message, "content_attributes", "email", "text_content", "reply")
|
|
or get_nested(payload, "content_attributes", "email", "html_content", "reply")
|
|
or get_nested(payload, "content_attributes", "email", "text_content", "reply")
|
|
or message.get("processed_message_content")
|
|
or message.get("content")
|
|
or payload.get("content")
|
|
or ""
|
|
)
|
|
|
|
return clean_email_reply_text(str(content or ""))
|