mirror of
https://github.com/hoshikawa2/compass_backoffice.git
synced 2026-07-09 13:54:20 +00:00
91 lines
3.2 KiB
Python
91 lines
3.2 KiB
Python
from __future__ import annotations
|
|
|
|
import re
|
|
from typing import Any
|
|
|
|
_CPF_RE = re.compile(r"(?i)\bcpf\b\s*[:=\-]?\s*(\d{3}\.\d{3}\.\d{3}-\d{2}|\d{11})\b")
|
|
_CNPJ_RE = re.compile(r"(?i)\bcnpj\b\s*[:=\-]?\s*(\d{2}\.\d{3}\.\d{3}/\d{4}-\d{2}|\d{14})\b")
|
|
_MSISDN_RE = re.compile(r"(?i)\b(?:msisdn|linha|telefone|celular)\b\s*[:=\-]?\s*(\+?55)?\s*\(?\d{2}\)?\s*9?\d{4}[-\s]?\d{4}\b")
|
|
_PROTOCOL_RE = re.compile(r"(?i)\b(?:protocolo|protocol_id|chamado|reclama[cç][aã]o|ticket)\b\s*[:=\-#]?\s*([A-Za-z0-9][A-Za-z0-9._\-/]{3,})\b")
|
|
|
|
|
|
def _digits(value: str | None) -> str | None:
|
|
if not value:
|
|
return None
|
|
digits = re.sub(r"\D+", "", str(value))
|
|
return digits or None
|
|
|
|
|
|
def extract_identity_from_text(text: str | None) -> dict[str, str]:
|
|
"""Extrai chaves de negócio de mensagens livres do usuário.
|
|
|
|
O IdentityResolver do framework mapeia campos estruturados. Esta função só
|
|
complementa payloads textuais como: "consultar dados do cliente cpf 123...".
|
|
"""
|
|
text = text or ""
|
|
found: dict[str, str] = {}
|
|
|
|
cpf = _CPF_RE.search(text)
|
|
if cpf:
|
|
value = _digits(cpf.group(1))
|
|
if value and len(value) == 11:
|
|
found["customer_key"] = value
|
|
found["cpf"] = value
|
|
found["document"] = value
|
|
found["document_type"] = "cpf"
|
|
|
|
cnpj = _CNPJ_RE.search(text)
|
|
if cnpj:
|
|
value = _digits(cnpj.group(1))
|
|
if value and len(value) == 14:
|
|
found["customer_key"] = value
|
|
found["cnpj"] = value
|
|
found["document"] = value
|
|
found["document_type"] = "cnpj"
|
|
|
|
protocol = _PROTOCOL_RE.search(text)
|
|
if protocol:
|
|
value = protocol.group(1).strip()
|
|
if value and not value.lower().startswith(("cpf", "cnpj")):
|
|
found["interaction_key"] = value
|
|
found["protocol_id"] = value
|
|
found["protocolo"] = value
|
|
|
|
# Só captura MSISDN quando há rótulo explícito para evitar confundir CPF/CNPJ.
|
|
msisdn = _MSISDN_RE.search(text)
|
|
if msisdn and "customer_key" not in found:
|
|
value = _digits(msisdn.group(0))
|
|
if value:
|
|
# Remove prefixo 55 se o usuário digitou com DDI.
|
|
if value.startswith("55") and len(value) in {12, 13}:
|
|
value = value[2:]
|
|
found["customer_key"] = value
|
|
found["msisdn"] = value
|
|
found["document_type"] = "msisdn"
|
|
|
|
return found
|
|
|
|
|
|
def enrich_payload_with_text_identity(payload: dict[str, Any] | None) -> dict[str, Any]:
|
|
payload = dict(payload or {})
|
|
text = (
|
|
payload.get("message")
|
|
or payload.get("text")
|
|
or payload.get("query")
|
|
or payload.get("content")
|
|
or ""
|
|
)
|
|
extracted = extract_identity_from_text(str(text))
|
|
# Payload estruturado sempre tem prioridade; extração só preenche lacunas.
|
|
for key, value in extracted.items():
|
|
payload.setdefault(key, value)
|
|
return payload
|
|
|
|
|
|
def enrich_context_with_text_identity(context: dict[str, Any] | None, text: str | None) -> dict[str, Any]:
|
|
context = dict(context or {})
|
|
extracted = extract_identity_from_text(text)
|
|
for key, value in extracted.items():
|
|
context.setdefault(key, value)
|
|
return context
|