1174 lines
54 KiB
Python
1174 lines
54 KiB
Python
"""invoice_resolver: resolve itens citados pelo cliente contra o JSON da fatura.
|
||
|
||
Cruza ``mentioned_items`` (extraídos pela camada de LLM em ``IntentExtractor``)
|
||
com o ``invoice_detail`` estruturado vindo do backend. Devolve
|
||
``ResolvedInvoiceItem`` canônicos, com ``tool_category``, ``item_type``,
|
||
``msisdn`` e ``value`` definidos deterministicamente.
|
||
|
||
Substitui três blocos do prompt do orquestrador
|
||
(``builtin/prompts/agent_orchestrator.yaml``):
|
||
|
||
- regra de localização do msisdn (Seção 2.1 — "CRÍTICA"): o msisdn vem da entrada
|
||
cujo ``desc`` casa com o item citado, nunca da linha titular por default.
|
||
- mapeamento seção → classe → tool (Seções 3.1 e 3.2 + matriz de roteamento).
|
||
- classificação ``bundle`` vs ``estrategico`` vs ``avulso`` (Subtipos de
|
||
``vas_estrategico`` + lista do Apêndice A).
|
||
|
||
**Matching híbrido (espelha o ``IntentExtractor``).** O casamento entre a menção
|
||
do cliente e o ``desc`` da fatura é determinístico (substring bidirecional) por
|
||
default. Um ``ItemMatcherLLM`` opcional, injetado no construtor, é acionado
|
||
**apenas quando o determinístico não casa nada** para uma menção — cobrindo erros
|
||
de grafia não previsíveis ("aia" → "Aya"). Sem matcher injetado, o resolver
|
||
permanece função pura/determinística (todos os contratos de ``resolve_items``
|
||
seguem inalterados).
|
||
|
||
**Ambiguidade.** Quando UMA menção casa com 2+ itens distintos (mesmo nome em
|
||
linhas diferentes OU nomes/descrições distintas), trata-se de alvo ambíguo: o
|
||
``resolve`` devolve esses matches em ``ResolutionOutcome.ambiguous`` em vez de
|
||
``resolved`` — o runtime devolve o turno ao orquestrador com uma dica para
|
||
perguntar a qual item o cliente se refere.
|
||
|
||
Estrutura esperada de ``invoice_detail`` (vinda de ``PDFBillingProcessor``):
|
||
|
||
{
|
||
"Fatura Resumo": {...},
|
||
"vocalized_msisdn": {...},
|
||
"<msisdn>": {
|
||
"Plano": [...],
|
||
"SVA Detalhe Total": [{"desc": "...", "value": 0.0, "msisdn": "...", "classe": "..."}],
|
||
"Serviços Bundle Inclusos": [...],
|
||
"Itens Eventuais": [...],
|
||
...
|
||
},
|
||
...
|
||
}
|
||
|
||
As seções de :data:`SECTION_DEFAULTS` são tratáveis (roteiam para uma tool); as
|
||
demais seções de serviço viram itens ``out_of_scope`` (encontrados mas não
|
||
tratáveis). ``_NON_SERVICE_SECTIONS`` (plano/desconto/consumo) são ignoradas.
|
||
Itens cuja entrada não casa com nenhum ``mentioned_item`` não são devolvidos.
|
||
"""
|
||
from __future__ import annotations
|
||
|
||
import logging
|
||
import re
|
||
import unicodedata
|
||
from concurrent.futures import ThreadPoolExecutor
|
||
from dataclasses import dataclass, field
|
||
from decimal import Decimal, InvalidOperation
|
||
from typing import Any, Iterable, NamedTuple, Protocol
|
||
|
||
from .invoice_models import MentionedItem, ResolvedInvoiceItem
|
||
from .invoice_models import is_period_range
|
||
|
||
logger = logging.getLogger(__name__)
|
||
|
||
# Teto de threads para as chamadas do matcher LLM por menção. Cada menção que
|
||
# fura o determinístico vira uma chamada LLM independente; elas são disparadas
|
||
# em paralelo (I/O-bound) até este limite para não estourar o pool com turnos
|
||
# que citam muitos serviços de uma vez.
|
||
_MATCHER_MAX_WORKERS = 8
|
||
|
||
# Piso de reconhecimento (determinístico, grafia OU som) ANTES de chamar o matcher
|
||
# LLM. Uma menção que fura o determinístico só vai ao LLM se algum candidato cruza
|
||
# este piso; senão é transcrição/ruído mal reconhecido → sem match (o runtime pede
|
||
# para o cliente repetir). Estrito: ``<= 0.6`` reprova (ruído de Jaro-Winkler bate
|
||
# 0.600 exato). Verificado: typo/pronúncia reais cruzam (Netflics 0.92, apou 0.64);
|
||
# basta UM canal alto (grafia OU som) — 'nei mar junior'×'Neymar Jr.' passa pela
|
||
# grafia (0.78) mesmo com som baixo.
|
||
_RECOGNITION_THRESHOLD = 0.6
|
||
|
||
# Piso do RESGATE de funil (``runtime._maybe_rescue_item_name``): mais alto que o
|
||
# piso normal porque ali a entrada é a fala CRUA do cliente, não o nome já extraído
|
||
# pelo classificador. Medido no repo: ruído real de fala inteira encosta em 0.60-0.64
|
||
# ("blarg zorp flim flam ploft" 0.640; "tem jogo hoje?" 0.603) enquanto o caso real
|
||
# de nome mal transcrito fica bem acima ("Game is done"×"Gamedom Mensal" 0.871).
|
||
# 0.75 separa os dois com folga dos dois lados.
|
||
_FUNNEL_RESCUE_THRESHOLD = 0.75
|
||
|
||
# Margem mínima entre o melhor e o segundo melhor candidato para o resgate aceitar
|
||
# um nome falado. Abaixo disso a fala aponta para 2+ itens parecidos ("Gamedom" com
|
||
# "Gamedom Mensal" e "Gamedom Semanal" na fatura) e o resgate falha FECHADO — quem
|
||
# desempata é o cliente, não o similarity.
|
||
_FUNNEL_RESCUE_MARGIN = 0.05
|
||
|
||
# Lista fixa do Apêndice A do agent_orchestrator.yaml: nomes conhecidos de
|
||
# serviços estratégicos. A lista é mantida como referência de produto, mas NÃO
|
||
# deve sobrepor a seção da fatura. O roteamento de ferramenta usa a seção como
|
||
# fonte de verdade para evitar que um VAS avulso com nome comercial conhecido
|
||
# caia indevidamente no fluxo VAS Estratégico Bundle.
|
||
STRATEGIC_NAMES: frozenset[str] = frozenset(
|
||
{
|
||
"apple music",
|
||
"deezer",
|
||
"disney",
|
||
"fuze",
|
||
"forge",
|
||
"hbo",
|
||
"looke",
|
||
"netflix",
|
||
"paramount",
|
||
"tim cloud gaming",
|
||
"youtube",
|
||
"globoplay",
|
||
"amazon prime"
|
||
}
|
||
)
|
||
|
||
# Mapeamento das seções TRATÁVEIS da fatura para (default_tool, default_type).
|
||
# O resolver enumera TODAS as seções do bucket (ver ``_iter_candidates``): seção
|
||
# aqui → sua tool/tipo; qualquer outra seção de serviço → item ``out_of_scope``
|
||
# (encontrado mas não tratável). ``_NON_SERVICE_SECTIONS`` filtra ruído.
|
||
#
|
||
# O default só é fonte de verdade quando a seção INTEIRA tem uma classe (avulso em
|
||
# SVA/Eventuais; estratégico/avulso nas seções do formato billing_analysis
|
||
# ``Serviços Contratados de Terceiros``/``Streamings``/``Serviços de valor
|
||
# adicionado``). "Mensalidades Adicionais" (formato PDF) é MISTA — estratégicos são
|
||
# carimbados pelo parser (flag ``estrategico``, honrada por ``_classify``) e o resto
|
||
# não é tratável — então NÃO entra aqui: sem flag → ``out_of_scope`` pelo default de
|
||
# ``_iter_candidates``. Estratégico = lista fixa por NOME (SPEC §5), não por seção.
|
||
SECTION_DEFAULTS: dict[str, tuple[str, str]] = {
|
||
"SVA Detalhe Total": ("cancelar_vas_avulso", "avulso"),
|
||
"Itens Eventuais": ("cancelar_vas_avulso", "avulso"),
|
||
"Serviços Contratados de Terceiros": ("vas_estrategico", "estrategico"),
|
||
"Serviços Bundle Inclusos": ("vas_estrategico", "bundle"),
|
||
"Serviços de valor adicionado": ("cancelar_vas_avulso", "avulso"),
|
||
"Streamings": ("vas_estrategico", "estrategico"),
|
||
}
|
||
|
||
# Seções que NÃO são serviços acionáveis pelo cliente: plano (fluxo pro_rata
|
||
# próprio), descontos/deduções e linhas de consumo. Suas entradas são ignoradas
|
||
# pelo sweep — evita falso ``out_of_scope`` e preserva o pro_rata. Exceção: uma
|
||
# entry com flag tratável (``estrategico``/``classe``) carimbada pelo parser é
|
||
# sempre candidata, mesmo aqui (um estratégico pode cair em "Outros Valores").
|
||
_NON_SERVICE_SECTIONS: frozenset[str] = frozenset(
|
||
{
|
||
"Plano",
|
||
"Planos",
|
||
"Descontos",
|
||
"Deduções",
|
||
"TIM Viagem",
|
||
"Chamadas Rede TIM",
|
||
"Roaming Internacional",
|
||
"Débitos de outras operadoras",
|
||
"Outros Valores",
|
||
}
|
||
)
|
||
|
||
# Tipo dos itens encontrados na fatura mas SEM tool de ação (nenhuma das 3
|
||
# classes tratáveis). O resolver os separa em ``ResolutionOutcome.out_of_scope``.
|
||
_OUT_OF_SCOPE_TYPE = "out_of_scope"
|
||
|
||
# Prefixos de linhas de crédito/ajuste financeiro (ex.: "CRÉDITO: PAGAMENTO",
|
||
# valor negativo). NÃO são serviços — mesmo dentro de uma seção de serviço e
|
||
# carimbadas ``classe=avulso`` pela seção, o resolver as ignora (como multas/juros
|
||
# em "Outros Valores"). Comparados por prefixo do desc, sem acento e sem caixa.
|
||
_NON_SERVICE_DESC_PREFIXES: tuple[str, ...] = ("credito:",)
|
||
|
||
# Chaves top-level do invoice_detail que NÃO são buckets por msisdn.
|
||
_NON_MSISDN_KEYS: frozenset[str] = frozenset(
|
||
{"Fatura Resumo", "DANFE-COM", "vocalized_msisdn"}
|
||
)
|
||
|
||
# Tokens de conexão falados descartados ao montar a CHAVE de comparação
|
||
# (a fala "VOD mais Canais" e o desc "VOD +Canais" convergem). Removidos dos
|
||
# dois lados — simétrico, nunca altera o nome canônico nem o payload.
|
||
_CONNECTOR_TOKENS: frozenset[str] = frozenset({"mais", "e"})
|
||
|
||
|
||
class _Candidate(NamedTuple):
|
||
"""Uma entrada candidata da fatura (uma linha de uma seção suportada),
|
||
com os defaults da seção já resolvidos. Reutilizada pelos caminhos
|
||
determinístico e LLM."""
|
||
|
||
desc: str
|
||
section: str
|
||
default_tool: str | None
|
||
default_type: str
|
||
parent_msisdn: str
|
||
entry: dict[str, Any]
|
||
|
||
|
||
class ItemMatcherError(Exception):
|
||
"""Erro do matcher de itens via LLM (timeout, parse, schema)."""
|
||
|
||
|
||
class ItemMatcherLLM(Protocol):
|
||
"""Casa a menção do cliente com os ``desc`` candidatos da fatura, tolerando
|
||
grafia/semântica. É a única dependência de LLM do resolver, injetada via
|
||
Protocol (espelha ``IntentClassifierLLM``); o resolver não importa cliente
|
||
concreto e permanece testável sem mock pesado."""
|
||
|
||
def match(
|
||
self,
|
||
mention: str,
|
||
candidates: list[str],
|
||
*,
|
||
callbacks: list[Any] | None = None,
|
||
) -> list[str]:
|
||
"""Devolve o subconjunto de ``candidates`` (descs exatos) que ``mention``
|
||
referencia. ``[]`` quando nada casa; múltiplos quando genuinamente
|
||
ambíguo. Em falha, levanta ``ItemMatcherError``."""
|
||
...
|
||
|
||
def best_similarity(self, mention: str, candidates: list[str]) -> float:
|
||
"""Maior ``max(grafia, som)`` entre os candidatos (determinístico, sem
|
||
LLM). O resolver usa como gate: ``<= _RECOGNITION_THRESHOLD`` → menção mal
|
||
reconhecida, não vale acionar o LLM. Sem candidatos → ``0.0``."""
|
||
...
|
||
|
||
|
||
@dataclass(frozen=True, slots=True)
|
||
class ItemResolution:
|
||
"""Resultado da resolução de UMA menção: a fala e os itens da fatura que
|
||
casaram com ela. ``is_ambiguous`` quando há 2+ itens distintos."""
|
||
|
||
mention: str
|
||
matches: list[ResolvedInvoiceItem] = field(default_factory=list)
|
||
|
||
@property
|
||
def is_ambiguous(self) -> bool:
|
||
"""Distinto por ``(canonical_name, msisdn, charge_date)``: nomes diferentes,
|
||
o mesmo nome em linhas diferentes (``msisdn``) E o mesmo nome/linha cobrado
|
||
em datas diferentes (``charge_date``) contam como itens distintos. Em fatura
|
||
SEM cobrança duplicada ``charge_date`` é ``None`` em todos → a tupla reduz a
|
||
``(canonical_name, msisdn)`` e o comportamento é o de antes."""
|
||
distinct = {(m.canonical_name, m.msisdn, m.charge_date) for m in self.matches}
|
||
return len(distinct) > 1
|
||
|
||
|
||
class RecognitionScore(NamedTuple):
|
||
"""Score determinístico da fala crua contra a fatura (ver
|
||
:meth:`InvoiceResolver.recognition_score`). ``margin`` separa o melhor do
|
||
segundo colocado — margem pequena = a fala aponta para 2+ itens parecidos."""
|
||
|
||
best: float
|
||
runner_up: float
|
||
desc: str
|
||
candidate_count: int
|
||
|
||
@property
|
||
def margin(self) -> float:
|
||
return self.best - self.runner_up
|
||
|
||
|
||
@dataclass(frozen=True, slots=True)
|
||
class ResolutionOutcome:
|
||
"""Saída de :meth:`InvoiceResolver.resolve`. ``resolved`` é o flat dedupado
|
||
das menções **não-ambíguas** (o que vai para o ``ActionQueueBuilder``);
|
||
``ambiguous`` carrega as menções com 2+ candidatos (o runtime devolve ao
|
||
orquestrador com dica); ``out_of_scope`` são menções que casaram um item da
|
||
fatura **não tratável** (nenhuma das 3 classes — ex.: seguro em "Cobranças de
|
||
Terceiros"); ``not_found`` lista as menções sem nenhum match."""
|
||
|
||
resolved: list[ResolvedInvoiceItem] = field(default_factory=list)
|
||
ambiguous: list[ItemResolution] = field(default_factory=list)
|
||
not_found: list[str] = field(default_factory=list)
|
||
out_of_scope: list[ItemResolution] = field(default_factory=list)
|
||
|
||
|
||
class InvoiceResolver:
|
||
"""Resolve itens citados pelo cliente contra o ``invoice_detail`` estruturado.
|
||
|
||
Stateless por turno (o único colaborador é o ``matcher`` opcional, também
|
||
stateless): instâncias podem ser compartilhadas. As operações públicas são
|
||
:meth:`resolve_items` (determinístico, back-compat) e :meth:`resolve`
|
||
(ambiguidade + matcher LLM). Os métodos com prefixo ``_`` são helpers,
|
||
expostos para teste unitário direto."""
|
||
|
||
def __init__(self, *, matcher: ItemMatcherLLM | None = None) -> None:
|
||
self._matcher = matcher
|
||
|
||
def resolve_items(
|
||
self,
|
||
mentioned_items: list[str],
|
||
invoice_detail: dict[str, Any],
|
||
) -> list[ResolvedInvoiceItem]:
|
||
"""Caminho determinístico/back-compat: itera ``mentioned_items`` na ordem
|
||
recebida e devolve TODOS os matches (inclusive os ambíguos), preservando
|
||
a ordem da fala. Saída deduplicada por ``(canonical_name, msisdn,
|
||
section)``. Mantido para os consumidores/testes existentes; o runtime usa
|
||
:meth:`resolve`."""
|
||
if not mentioned_items or not isinstance(invoice_detail, dict):
|
||
return []
|
||
|
||
resolved: list[ResolvedInvoiceItem] = []
|
||
for mention in mentioned_items:
|
||
if not isinstance(mention, str) or not mention.strip():
|
||
continue
|
||
resolved.extend(self._resolve_one_mention(mention, invoice_detail, include_identity_oos=False))
|
||
return self._dedupe_exact_matches(resolved)
|
||
|
||
def resolve_each(
|
||
self,
|
||
mentions: "list[str | MentionedItem]",
|
||
invoice_detail: dict[str, Any],
|
||
*,
|
||
callbacks: list[Any] | None = None,
|
||
) -> list[ItemResolution]:
|
||
"""Motor de matching por menção: devolve uma :class:`ItemResolution` para
|
||
cada menção VÁLIDA (não-vazia), 1:1 e na ordem da fala.
|
||
|
||
Cada menção é uma ``str`` (só o nome) OU um :class:`MentionedItem` (nome +
|
||
``msisdn``/``date``-DICA). Duas passadas: (1) determinística por NOME; (2) as
|
||
menções que não casaram nada determinísticamente vão para o ``matcher`` LLM
|
||
(tolerante a grafia), cada uma em seu prompt, em paralelo. Quando a menção traz
|
||
``msisdn`` e/ou ``date``, os matches são FILTRADOS — primeiro pela LINHA
|
||
(:meth:`_filter_by_line`), depois pela COBRANÇA (:meth:`_filter_by_charge`),
|
||
nessa ordem: a data só afunila DENTRO da linha já escolhida. Desambigua o mesmo
|
||
nome em 2 linhas ou 2 cobranças → 1 item. É o miolo compartilhado por
|
||
:meth:`resolve` (que classifica em resolved/ambiguous/not_found) e pelo runtime
|
||
na expansão de RAG (só strings, sem filtro). Não classifica nem dedupa entre
|
||
menções — só resolve cada uma (dedupe interno por cobrança)."""
|
||
if not mentions or not isinstance(invoice_detail, dict):
|
||
return []
|
||
|
||
candidates = list(self._iter_candidates(invoice_detail))
|
||
unique_descs = self._unique_descs(candidates)
|
||
|
||
# Passada 1 — determinística por NOME. Guarda os matches por menção na ordem
|
||
# da fala + o msisdn-dica de cada uma (``lines``); acumula em ``pending`` as
|
||
# menções (válidas) que não casaram nada e precisam do LLM (só quando há
|
||
# matcher e candidatos). O gate de reconhecimento (grafia E som) barra antes
|
||
# do LLM as menções mal reconhecidas (transcrição/ruído).
|
||
deterministic: dict[int, list[ResolvedInvoiceItem]] = {}
|
||
ordered: list[tuple[int, str]] = []
|
||
lines: dict[int, str | None] = {}
|
||
dates: dict[int, str | None] = {}
|
||
pending: list[tuple[int, str]] = []
|
||
for mention in mentions:
|
||
name, msisdn, date = self._mention_fields(mention)
|
||
if not name.strip():
|
||
continue
|
||
idx = len(ordered)
|
||
ordered.append((idx, name))
|
||
lines[idx] = msisdn
|
||
dates[idx] = date
|
||
matches = self._resolve_one_mention(name, invoice_detail, include_identity_oos=True)
|
||
if matches:
|
||
deterministic[idx] = matches
|
||
elif (
|
||
self._matcher is not None
|
||
and candidates
|
||
and self._passes_recognition_gate(name, unique_descs)
|
||
):
|
||
pending.append((idx, name))
|
||
|
||
# Passada 2 — uma chamada LLM por menção pendente, em paralelo.
|
||
llm_results = self._match_pending_parallel(pending, candidates, callbacks)
|
||
|
||
resolutions: list[ItemResolution] = []
|
||
for idx, name in ordered:
|
||
if idx in deterministic:
|
||
matches = deterministic[idx]
|
||
else:
|
||
matches = llm_results.get(idx, [])
|
||
matches = self._filter_by_line(matches, lines.get(idx))
|
||
matches = self._filter_by_charge(matches, dates.get(idx))
|
||
resolutions.append(
|
||
ItemResolution(
|
||
mention=name,
|
||
matches=self._dedupe_exact_matches(matches, by_charge=True),
|
||
)
|
||
)
|
||
return resolutions
|
||
|
||
def resolve(
|
||
self,
|
||
mentioned_items: "list[str | MentionedItem]",
|
||
invoice_detail: dict[str, Any],
|
||
*,
|
||
callbacks: list[Any] | None = None,
|
||
) -> ResolutionOutcome:
|
||
"""Resolve as menções com consciência de ambiguidade e matcher LLM.
|
||
|
||
As menções são ``str`` (só nome) OU :class:`MentionedItem` (nome + msisdn-dica
|
||
para desambiguar a linha). Delega o matching a :meth:`resolve_each` e classifica
|
||
cada :class:`ItemResolution` em resolved / ambiguous / out_of_scope / not_found.
|
||
``resolved`` é o flat dedupado das menções não-ambíguas, na ordem da fala.
|
||
Uma menção que só casou item(s) não tratável(is) vai para ``out_of_scope``
|
||
(não é ``not_found`` — ela casou algo na fatura); matches tratáveis têm
|
||
precedência (menção mista segue o fluxo de ação)."""
|
||
resolutions = self.resolve_each(
|
||
mentioned_items, invoice_detail, callbacks=callbacks
|
||
)
|
||
resolved: list[ResolvedInvoiceItem] = []
|
||
ambiguous: list[ItemResolution] = []
|
||
not_found: list[str] = []
|
||
out_of_scope: list[ItemResolution] = []
|
||
for resolution in resolutions:
|
||
treatable = [
|
||
m for m in resolution.matches if m.item_type != _OUT_OF_SCOPE_TYPE
|
||
]
|
||
oos = [m for m in resolution.matches if m.item_type == _OUT_OF_SCOPE_TYPE]
|
||
if treatable:
|
||
treatable_res = ItemResolution(
|
||
mention=resolution.mention, matches=treatable
|
||
)
|
||
if treatable_res.is_ambiguous:
|
||
ambiguous.append(treatable_res)
|
||
else:
|
||
resolved.extend(treatable)
|
||
elif oos:
|
||
out_of_scope.append(
|
||
ItemResolution(mention=resolution.mention, matches=oos)
|
||
)
|
||
else:
|
||
not_found.append(resolution.mention)
|
||
return ResolutionOutcome(
|
||
resolved=self._dedupe_exact_matches(resolved, by_charge=True),
|
||
ambiguous=ambiguous,
|
||
not_found=not_found,
|
||
out_of_scope=out_of_scope,
|
||
)
|
||
|
||
def build_snapshot(
|
||
self,
|
||
invoice_detail: dict[str, Any],
|
||
*,
|
||
by_charge: bool = False,
|
||
) -> tuple[ResolvedInvoiceItem, ...]:
|
||
"""Enumera todos os itens da fatura como ``ResolvedInvoiceItem`` canônicos,
|
||
para uso como ``state.invoice_snapshot``.
|
||
|
||
Reusa ``_iter_candidates`` + ``_build_item`` da mesma máquina que serve
|
||
``resolve``: a classificação por seção e a ordem
|
||
``bucket → SECTION_DEFAULTS → entry`` são preservadas. Sem ambiguidade
|
||
envolvida (esta API é orientada a snapshot completo, não a matching de
|
||
menção), apenas dedupe por ``(canonical_name, msisdn, section)``.
|
||
|
||
``by_charge`` repassa ao ``_dedupe_exact_matches``: ``False`` (default)
|
||
colapsa cobranças duplicadas do mesmo serviço/linha (contrato dos consumers
|
||
atuais: snapshot do classificador e ``_has_retention_vas``); ``True`` inclui
|
||
``charge_date`` na chave, preservando cobranças datadas distintas (ex.: mesmo
|
||
VAS avulso cobrado em 2 ciclos) — usado ao montar a fila de cancelamento de
|
||
todos os avulsos, alinhado ao ``resolve``/``resolve_each``.
|
||
|
||
Retorna tupla (imutável) para uso como contexto estruturado no
|
||
classificador via ``state.invoice_snapshot``. O filtro por msisdn ativo
|
||
é responsabilidade do consumer (render do classifier), permitindo que
|
||
cenários família multi-msisdn vejam apenas a linha em discussão.
|
||
|
||
Devolve tupla vazia em payload inválido (silencioso, como o resto do
|
||
resolver): o consumer trata snapshot vazio como ausência de contexto."""
|
||
if not isinstance(invoice_detail, dict):
|
||
return ()
|
||
items: list[ResolvedInvoiceItem] = []
|
||
for cand in self._iter_candidates(invoice_detail):
|
||
items.append(self._build_item(cand))
|
||
return tuple(self._dedupe_exact_matches(items, by_charge=by_charge))
|
||
|
||
# ----- normalização de menção + filtro de linha ------------------------
|
||
|
||
@staticmethod
|
||
def _mention_fields(mention: Any) -> tuple[str, str | None, str | None]:
|
||
"""Normaliza uma menção em ``(nome, msisdn-dica, date-dica)``. ``str`` →
|
||
(nome, None, None); :class:`MentionedItem` → (desc, msisdn, date). O msisdn é
|
||
DICA de LINHA e a date é DICA de COBRANÇA: ambos são casados contra a fatura
|
||
real (ver :meth:`_filter_by_line`/:meth:`_filter_by_charge`), nunca fonte da
|
||
verdade. ``value`` do :class:`MentionedItem` é ignorado (o LLM ecoa ``14.9``
|
||
de ``14.90``; não serve como chave)."""
|
||
if isinstance(mention, MentionedItem):
|
||
msisdn = str(mention.msisdn).strip() if mention.msisdn else ""
|
||
date = str(mention.date).strip() if mention.date else ""
|
||
return str(mention.desc or ""), (msisdn or None), (date or None)
|
||
if isinstance(mention, str):
|
||
return mention, None, None
|
||
return "", None, None
|
||
|
||
def _filter_by_line(
|
||
self, matches: list[ResolvedInvoiceItem], msisdn: str | None
|
||
) -> list[ResolvedInvoiceItem]:
|
||
"""Restringe os matches à linha ``msisdn`` quando a menção a trouxe: o mesmo
|
||
nome em 2 linhas (ambíguo) vira 1 item na linha escolhida. O msisdn é DICA —
|
||
se NÃO casar nenhuma linha real dos matches, NÃO filtra (dica inválida →
|
||
preserva o comportamento sem linha). ``None``/sem matches → devolve como está."""
|
||
if not msisdn or not matches:
|
||
return matches
|
||
on_line = [m for m in matches if self._same_line(m.msisdn, msisdn)]
|
||
return on_line or matches
|
||
|
||
@staticmethod
|
||
def _same_line(item_msisdn: str, wanted: str) -> bool:
|
||
"""Compara dois msisdns só por dígitos: igual, ou um é SUFIXO do outro com ≥4
|
||
dígitos (o cliente/classifier pode dar só os últimos, ex.: "8119")."""
|
||
a = "".join(ch for ch in str(item_msisdn) if ch.isdigit())
|
||
b = "".join(ch for ch in str(wanted) if ch.isdigit())
|
||
if not b:
|
||
return True
|
||
if a == b:
|
||
return True
|
||
if len(b) >= 4 and a.endswith(b):
|
||
return True
|
||
if len(a) >= 4 and b.endswith(a):
|
||
return True
|
||
return False
|
||
|
||
def _filter_by_charge(
|
||
self, matches: list[ResolvedInvoiceItem], date: str | None
|
||
) -> list[ResolvedInvoiceItem]:
|
||
"""Restringe os matches à COBRANÇA de ``date`` quando a menção a trouxe:
|
||
duas cobranças do mesmo nome na mesma linha (ambíguo por data) viram 1 na
|
||
data escolhida. A date é DICA — se NÃO casar nenhuma cobrança real dos
|
||
matches, NÃO filtra (dica inválida → preserva o ambíguo; nunca zera).
|
||
``None``/sem matches → devolve como está. Espelha :meth:`_filter_by_line`."""
|
||
if not date or not matches:
|
||
return matches
|
||
on_charge = [m for m in matches if self._same_charge(m.charge_date, date)]
|
||
return on_charge or matches
|
||
|
||
@staticmethod
|
||
def _same_charge(item_date: str | None, wanted: str) -> bool:
|
||
"""Compara duas datas de cobrança só por dígitos, com tolerância de PREFIXO
|
||
(não sufixo como ``_same_line``): ``dd/mm`` casa ``dd/mm/aa`` — o cliente/
|
||
classifier pode dar a data sem o ano. Item sem ``charge_date`` nunca casa uma
|
||
dica de data (não é uma cobrança datada)."""
|
||
a = "".join(ch for ch in str(item_date or "") if ch.isdigit())
|
||
b = "".join(ch for ch in str(wanted) if ch.isdigit())
|
||
if not b:
|
||
return True
|
||
if not a:
|
||
return False
|
||
return a == b or a.startswith(b) or b.startswith(a)
|
||
|
||
# ----- matching por menção ---------------------------------------------
|
||
|
||
def _resolve_one_mention(
|
||
self,
|
||
mention: str,
|
||
invoice_detail: dict[str, Any],
|
||
*,
|
||
include_identity_oos: bool = False,
|
||
) -> list[ResolvedInvoiceItem]:
|
||
"""Match determinístico de UMA menção contra todas as combinações
|
||
msisdn × seção × entry, com precedência de match exato sobre substring.
|
||
|
||
**Match exato** sobre a CHAVE normalizada (:meth:`_normalize_match_text`:
|
||
sem acentos, lower, ``+`` como separador, conectores falados ``mais``/
|
||
``e`` descartados, ``filmes``→``filme``) vence quando existe. Assim
|
||
variações rasas de grafia/fala do MESMO nome convergem — ``"VOD + Canais
|
||
Abertos + Fechados"`` (espaçado, como o agente lista), ``"VOD +Canais
|
||
Abertos +Fechados"`` (desc da fatura) e ``"VOD mais Canais Abertos mais
|
||
Fechados"`` (falado) casam o item correto sem cair no substring. Nomes
|
||
correlatos onde um é prefixo de outro (``"VOD + Canais"`` vs ``"VOD +
|
||
Canais Abertos"``) continuam itens distintos: cada um casa exato a sua
|
||
própria chave (CY0004 preservado). Múltiplos exatos só ocorrem com o
|
||
mesmo desc em linhas diferentes (família) — ambiguidade legítima.
|
||
|
||
**Fallback substring** (bidirecional via ``_was_mentioned``, também
|
||
normalizado) é usado quando não há exato: cobre menção parcial ("aya"
|
||
para "Aya Books") e o clássico "dois ayas". O resultado do substring
|
||
passa pelo **guard anti-prefixo** (:meth:`_apply_prefix_guard`): nunca
|
||
deixa um prefixo curto vencer silenciosamente quando a menção carrega os
|
||
tokens que distinguem um item mais específico.
|
||
|
||
(incidente CY0007: a menção espaçada ``"VOD + Canais Abertos + Fechados"``
|
||
casava por substring o prefixo ``"VOD + Canais Abertos"`` — item errado;
|
||
a normalização a torna exata ao item ``+Fechados``.)"""
|
||
mention_normalized = self._normalize_match_text(mention)
|
||
|
||
# Primeiro procure igualdade EXATA em todo o catálogo da fatura, inclusive
|
||
# seções não acionáveis (Plano, descontos etc.). Isso é um guard de identidade:
|
||
# se o cliente nomeou precisamente um item conhecido, esse item precisa vencer
|
||
# antes de qualquer fuzzy matching. Sem isso, um plano explicitamente citado
|
||
# pode ser removido do universo tratável e o matcher acabar autorizando outro
|
||
# VAS apenas por similaridade (ex.: TIM CTRL Redes Sociais -> TIM Fashion).
|
||
if include_identity_oos:
|
||
identity_candidates = list(self._iter_identity_candidates(invoice_detail))
|
||
exact_identity = [
|
||
cand
|
||
for cand in identity_candidates
|
||
if mention_normalized
|
||
and self._normalize_match_text(cand.desc) == mention_normalized
|
||
]
|
||
if exact_identity:
|
||
return [self._build_item(cand) for cand in exact_identity]
|
||
|
||
# Fuzzy/substring continua restrito ao universo de serviços acionáveis +
|
||
# out-of-scope de seções de serviço. Seções explicitamente não acionáveis
|
||
# jamais entram no matcher aproximado; elas só podem bloquear por igualdade
|
||
# exata acima. Isso mantém o comportamento conservador do fluxo.
|
||
candidates = list(self._iter_candidates(invoice_detail))
|
||
substring_cands: list[_Candidate] = []
|
||
for cand in candidates:
|
||
if self._was_mentioned(cand.desc, [mention]):
|
||
substring_cands.append(cand)
|
||
guarded = self._apply_prefix_guard(mention, substring_cands, candidates)
|
||
return [self._build_item(cand) for cand in guarded]
|
||
|
||
def _apply_prefix_guard(
|
||
self,
|
||
mention: str,
|
||
substring_cands: list[_Candidate],
|
||
all_candidates: list[_Candidate],
|
||
) -> list[_Candidate]:
|
||
"""Guard anti-prefixo sobre o resultado do substring (sem match exato).
|
||
|
||
Impede que um candidato curto (prefixo) seja escolhido silenciosamente
|
||
quando a menção carrega os tokens que identificam um candidato mais
|
||
específico. Duas regras, ambas sobre a chave normalizada:
|
||
|
||
- **Promoção (→ ambíguo):** se um candidato curto ``C`` casou e existe um
|
||
candidato ``L`` que ESTENDE ``C`` (prefixo de tokens) cujos tokens
|
||
extras estão TODOS na menção, mas ``L`` não casou contíguo, ``L`` é
|
||
adicionado. ``C`` e ``L`` distintos ⇒ ``is_ambiguous`` ⇒ o runtime
|
||
pergunta ao cliente (Policy de desambiguação já existente).
|
||
- **Colapso (→ determinístico):** quando a menção COBRE por completo o
|
||
item mais longo (``norm(L)`` é substring de ``norm(menção)``), o prefixo
|
||
curto ``C`` que ``L`` estende é descartado — resolve no item completo.
|
||
|
||
Sem candidatos de substring, é no-op. Devolve ``list[_Candidate]``; o
|
||
chamador constrói os ``ResolvedInvoiceItem``."""
|
||
if not substring_cands:
|
||
return substring_cands
|
||
mention_norm = self._normalize_match_text(mention)
|
||
mention_tokens = set(mention_norm.split())
|
||
matched_norms = {self._normalize_match_text(c.desc) for c in substring_cands}
|
||
|
||
promoted: list[_Candidate] = list(substring_cands)
|
||
for cand in all_candidates:
|
||
cand_norm = self._normalize_match_text(cand.desc)
|
||
if cand_norm in matched_norms:
|
||
continue
|
||
for matched in substring_cands:
|
||
base_norm = self._normalize_match_text(matched.desc)
|
||
if not self._is_token_prefix(base_norm, cand_norm):
|
||
continue
|
||
base_tokens = set(base_norm.split())
|
||
extra = [t for t in cand_norm.split() if t not in base_tokens]
|
||
if extra and all(t in mention_tokens for t in extra):
|
||
promoted.append(cand)
|
||
matched_norms.add(cand_norm)
|
||
break
|
||
|
||
norms = [(self._normalize_match_text(c.desc), c) for c in promoted]
|
||
kept: list[_Candidate] = []
|
||
for norm_i, cand in norms:
|
||
shadowed = any(
|
||
norm_j != norm_i
|
||
and self._is_token_prefix(norm_i, norm_j)
|
||
and norm_j in mention_norm
|
||
for norm_j, _ in norms
|
||
)
|
||
if not shadowed:
|
||
kept.append(cand)
|
||
return kept
|
||
|
||
@staticmethod
|
||
def _is_token_prefix(shorter: str, longer: str) -> bool:
|
||
"""True quando ``longer`` estende ``shorter`` no nível de TOKEN
|
||
(``shorter`` + pelo menos mais um token). Evita falso prefixo
|
||
intra-palavra (ex.: "vod cana" NÃO é prefixo de "vod canais")."""
|
||
return bool(shorter) and longer != shorter and longer.startswith(shorter + " ")
|
||
|
||
@staticmethod
|
||
def _unique_descs(candidates: list[_Candidate]) -> list[str]:
|
||
"""Descs únicos preservando a ordem de aparição na fatura (chave
|
||
case-insensitive). Conjunto candidato compartilhado pelo matcher LLM e
|
||
pelo gate de reconhecimento."""
|
||
unique_descs: list[str] = []
|
||
seen_desc: set[str] = set()
|
||
for cand in candidates:
|
||
key = cand.desc.lower()
|
||
if key not in seen_desc:
|
||
seen_desc.add(key)
|
||
unique_descs.append(cand.desc)
|
||
return unique_descs
|
||
|
||
def _passes_recognition_gate(self, mention: str, descs: list[str]) -> bool:
|
||
"""Gate determinístico antes do matcher LLM: só vale chamar o LLM se algum
|
||
candidato tem grafia OU som acima do piso (``best_similarity >
|
||
_RECOGNITION_THRESHOLD``). Menção mal reconhecida (transcrição/ruído) fica
|
||
sem match → o runtime pede para o cliente repetir. Se o matcher não expõe
|
||
``best_similarity`` (ex.: fake antigo), não bloqueia — degrada para o
|
||
comportamento legado (sempre chama o LLM)."""
|
||
best = getattr(self._matcher, "best_similarity", None)
|
||
if best is None:
|
||
return True
|
||
try:
|
||
return best(mention, descs) > _RECOGNITION_THRESHOLD
|
||
except Exception: # noqa: BLE001 — gate nunca quebra o turno; degrada p/ legado
|
||
logger.debug("conversation.invoice.recognition_gate_failed", exc_info=True)
|
||
return True
|
||
|
||
def recognition_score(
|
||
self, mention: str, invoice_detail: dict[str, Any]
|
||
) -> "RecognitionScore | None":
|
||
"""Score determinístico da fala CRUA contra os itens da fatura (sem LLM).
|
||
|
||
Usado pelo RESGATE de funil (``runtime._maybe_rescue_item_name``) para
|
||
decidir se vale tratar a fala como nome de serviço. Reusa exatamente o
|
||
scoring do matcher (``best_similarity``), aplicado candidato a candidato
|
||
para expor também o segundo colocado — a margem entre os dois é o que
|
||
detecta ambiguidade ("Gamedom" com "Gamedom Mensal" e "Gamedom Semanal"
|
||
na fatura).
|
||
|
||
**Fail-closed** (≠ ``_passes_recognition_gate``, que degrada para ``True``
|
||
quando o matcher não expõe ``best_similarity``): aqui, sem matcher, sem
|
||
``best_similarity``, sem candidatos ou em qualquer erro devolve ``None``
|
||
e o chamador NÃO resgata. A assimetria é deliberada: o gate legado protege
|
||
um caminho já autorizado; o resgate CRIA autorização a partir de uma fala
|
||
que o classificador não entendeu."""
|
||
best_fn = getattr(self._matcher, "best_similarity", None)
|
||
if best_fn is None:
|
||
return None
|
||
text = (mention or "").strip()
|
||
if not text:
|
||
return None
|
||
try:
|
||
candidates = list(self._iter_candidates(invoice_detail))
|
||
descs = self._unique_descs(candidates)
|
||
if not descs:
|
||
return None
|
||
scored = sorted(
|
||
((float(best_fn(text, [desc])), desc) for desc in descs),
|
||
key=lambda pair: pair[0],
|
||
reverse=True,
|
||
)
|
||
except Exception: # noqa: BLE001 — resgate nunca quebra o turno; fail-closed
|
||
logger.debug("conversation.invoice.recognition_score_failed", exc_info=True)
|
||
return None
|
||
best_score, best_desc = scored[0]
|
||
runner_up = scored[1][0] if len(scored) > 1 else 0.0
|
||
return RecognitionScore(
|
||
best=best_score,
|
||
runner_up=runner_up,
|
||
desc=best_desc,
|
||
candidate_count=len(descs),
|
||
)
|
||
|
||
def _match_pending_parallel(
|
||
self,
|
||
pending: list[tuple[int, str]],
|
||
candidates: list[_Candidate],
|
||
callbacks: list[Any] | None,
|
||
) -> dict[int, list[ResolvedInvoiceItem]]:
|
||
"""Aciona o matcher LLM para cada menção pendente (as que o determinístico
|
||
não casou), uma por prompt, em paralelo via threads (I/O-bound). Devolve
|
||
``{idx: [itens]}`` por menção. Falha de uma menção é engolida (log) → ela
|
||
degrada para sem match, sem afetar as demais. Sem matcher, pendentes ou
|
||
candidatos, devolve ``{}``."""
|
||
if not pending or not candidates or self._matcher is None:
|
||
return {}
|
||
# Descs únicos preservando a ordem de aparição na fatura; o mesmo conjunto
|
||
# candidato serve todas as menções do turno (e o gate de reconhecimento).
|
||
unique_descs = self._unique_descs(candidates)
|
||
|
||
# Propaga o contexto OTel ativo (o span pai do Langfuse, aberto pelo
|
||
# runtime em ``conversation.invoke.resolve``) para as worker threads.
|
||
# ``contextvars`` NÃO são herdados por threads novas — sem reanexar o
|
||
# contexto, as gerações do matcher abrem traces órfãos em vez de
|
||
# aninhar sob o span pai. Captura aqui (main thread, dentro do span).
|
||
run_in_context = self._otel_context_runner()
|
||
|
||
max_workers = min(len(pending), _MATCHER_MAX_WORKERS)
|
||
with ThreadPoolExecutor(max_workers=max_workers) as executor:
|
||
futures = {
|
||
executor.submit(
|
||
run_in_context,
|
||
self._match_one_llm,
|
||
mention,
|
||
unique_descs,
|
||
callbacks,
|
||
): idx
|
||
for idx, mention in pending
|
||
}
|
||
results: dict[int, list[ResolvedInvoiceItem]] = {}
|
||
for future, idx in futures.items():
|
||
results[idx] = self._items_from_descs(future.result(), candidates)
|
||
return results
|
||
|
||
@staticmethod
|
||
def _otel_context_runner() -> Any:
|
||
"""Devolve um wrapper ``run(fn, *args)`` que executa ``fn`` dentro do
|
||
contexto OTel ativo no momento desta chamada (main thread). Usado para
|
||
que as chamadas do matcher em worker threads aninhem sob o span pai do
|
||
Langfuse. Sem OpenTelemetry instalado, devolve um passthrough (degrada
|
||
sem tracing, nunca quebra)."""
|
||
try:
|
||
from opentelemetry import context as otel_context
|
||
except Exception: # noqa: BLE001 — sem otel: roda sem propagação de contexto
|
||
return lambda fn, *args, **kwargs: fn(*args, **kwargs)
|
||
|
||
parent_ctx = otel_context.get_current()
|
||
|
||
def _run(fn: Any, *args: Any, **kwargs: Any) -> Any:
|
||
token = otel_context.attach(parent_ctx)
|
||
try:
|
||
return fn(*args, **kwargs)
|
||
finally:
|
||
otel_context.detach(token)
|
||
|
||
return _run
|
||
|
||
def _match_one_llm(
|
||
self,
|
||
mention: str,
|
||
unique_descs: list[str],
|
||
callbacks: list[Any] | None,
|
||
) -> list[str]:
|
||
"""Chamada única do matcher LLM para UMA menção; roda dentro de uma
|
||
thread. Falha (``ItemMatcherError`` ou outra) é engolida (log) → ``[]``,
|
||
nunca quebra o turno nem as outras menções em paralelo."""
|
||
try:
|
||
return self._matcher.match(mention, unique_descs, callbacks=callbacks)
|
||
except Exception: # noqa: BLE001 — ItemMatcherError ou outra: nunca quebra o turno
|
||
logger.debug("conversation.invoice.matcher_failed", exc_info=True)
|
||
return []
|
||
|
||
def _items_from_descs(
|
||
self,
|
||
matched_descs: list[str],
|
||
candidates: list[_Candidate],
|
||
) -> list[ResolvedInvoiceItem]:
|
||
"""Mapeia os descs casados pelo LLM de volta para TODAS as entries com
|
||
aquele desc (desc em 2 linhas → 2 matches → ambíguo). Defensivo: só
|
||
aceita descs que estão de fato no conjunto candidato, comparando pela
|
||
CHAVE normalizada — assim uma grafia espaçada devolvida pelo LLM ("VOD +
|
||
Canais Abertos + Fechados") ainda mapeia para o candidato compacto da
|
||
fatura, em vez de ser descartada silenciosamente."""
|
||
wanted = {
|
||
self._normalize_match_text(d)
|
||
for d in (matched_descs or [])
|
||
if isinstance(d, str) and str(d).strip()
|
||
}
|
||
wanted.discard("")
|
||
if not wanted:
|
||
return []
|
||
out: list[ResolvedInvoiceItem] = []
|
||
for cand in candidates:
|
||
if self._normalize_match_text(cand.desc) in wanted:
|
||
out.append(self._build_item(cand))
|
||
return out
|
||
|
||
# ----- construção de candidatos e itens --------------------------------
|
||
|
||
def _iter_identity_candidates(
|
||
self, invoice_detail: dict[str, Any]
|
||
) -> Iterable[_Candidate]:
|
||
"""Itera itens nomeáveis de TODAS as seções da fatura para match exato.
|
||
|
||
Diferente de :meth:`_iter_candidates`, este catálogo inclui também seções
|
||
não acionáveis como ``Plano``/``Planos``. Esses itens são construídos como
|
||
``out_of_scope`` e servem somente para preservar a identidade explicitamente
|
||
citada pelo cliente. Eles nunca participam do fuzzy matcher.
|
||
"""
|
||
for parent_msisdn, sections in self._iter_msisdn_buckets(invoice_detail):
|
||
if not isinstance(sections, dict):
|
||
continue
|
||
for section, entries in sections.items():
|
||
if not isinstance(entries, list):
|
||
continue
|
||
section_tool, section_type = SECTION_DEFAULTS.get(
|
||
section, (None, _OUT_OF_SCOPE_TYPE)
|
||
)
|
||
for entry in entries:
|
||
if not isinstance(entry, dict) or self._is_non_service_item(entry):
|
||
continue
|
||
desc = str(entry.get("desc") or "").strip()
|
||
if not desc:
|
||
continue
|
||
default_tool, default_type = section_tool, section_type
|
||
# Se a seção é explicitamente não acionável e a entry não foi
|
||
# carimbada como tratável pelo parser, force out_of_scope.
|
||
if section in _NON_SERVICE_SECTIONS and not self._has_treatable_flag(entry):
|
||
default_tool, default_type = None, _OUT_OF_SCOPE_TYPE
|
||
yield _Candidate(
|
||
desc=desc,
|
||
section=section,
|
||
default_tool=default_tool,
|
||
default_type=default_type,
|
||
parent_msisdn=parent_msisdn,
|
||
entry=entry,
|
||
)
|
||
|
||
def _iter_candidates(
|
||
self, invoice_detail: dict[str, Any]
|
||
) -> Iterable[_Candidate]:
|
||
"""Itera as entradas de TODAS as seções de serviço em cada bucket por
|
||
msisdn. Seção em ``SECTION_DEFAULTS`` herda sua ``(tool, type)``; as
|
||
demais recaem em ``(None, out_of_scope)`` — item encontrado mas não
|
||
tratável. Ignora metadados, entradas sem ``desc`` e seções de
|
||
``_NON_SERVICE_SECTIONS`` (plano/desconto/consumo), salvo entry com flag
|
||
tratável carimbada pelo parser (que é sempre candidata)."""
|
||
for parent_msisdn, sections in self._iter_msisdn_buckets(invoice_detail):
|
||
if not isinstance(sections, dict):
|
||
continue
|
||
for section, entries in sections.items():
|
||
if not isinstance(entries, list):
|
||
continue
|
||
default_tool, default_type = SECTION_DEFAULTS.get(
|
||
section, (None, _OUT_OF_SCOPE_TYPE)
|
||
)
|
||
skip_section = section in _NON_SERVICE_SECTIONS
|
||
for entry in entries:
|
||
if not isinstance(entry, dict):
|
||
continue
|
||
if self._is_non_service_item(entry):
|
||
continue
|
||
if skip_section and not self._has_treatable_flag(entry):
|
||
continue
|
||
desc = str(entry.get("desc") or "").strip()
|
||
if not desc:
|
||
continue
|
||
yield _Candidate(
|
||
desc=desc,
|
||
section=section,
|
||
default_tool=default_tool,
|
||
default_type=default_type,
|
||
parent_msisdn=parent_msisdn,
|
||
entry=entry,
|
||
)
|
||
|
||
@staticmethod
|
||
def _has_treatable_flag(entry: dict[str, Any]) -> bool:
|
||
"""True se o parser carimbou uma classe tratável (``estrategico``/
|
||
``classe``) na entry — usada para não descartar um estratégico que caia
|
||
numa seção de ``_NON_SERVICE_SECTIONS``."""
|
||
if entry.get("estrategico") is True:
|
||
return True
|
||
classe = str(entry.get("classe") or "").strip().casefold()
|
||
return classe in {"estrategico", "estrategica", "strategic", "bundle", "avulso", "avulsa"}
|
||
|
||
@staticmethod
|
||
def _is_non_service_item(entry: dict[str, Any]) -> bool:
|
||
"""True para linhas de crédito/ajuste financeiro (ex.: "CRÉDITO: PAGAMENTO"),
|
||
que NÃO são serviços tratáveis mesmo carimbadas ``classe=avulso`` pela seção
|
||
("Itens Eventuais"). Casa o prefixo do ``desc`` sem acento/caixa. Exclusão
|
||
incondicional: um crédito nunca é candidato, independente de flag."""
|
||
desc = str(entry.get("desc") or "")
|
||
norm = (
|
||
unicodedata.normalize("NFKD", desc)
|
||
.encode("ascii", "ignore")
|
||
.decode()
|
||
.casefold()
|
||
.strip()
|
||
)
|
||
return norm.startswith(_NON_SERVICE_DESC_PREFIXES)
|
||
|
||
def _build_item(self, cand: _Candidate) -> ResolvedInvoiceItem:
|
||
"""Monta o ``ResolvedInvoiceItem`` canônico de um candidato: aplica a
|
||
classificação por seção, deriva ``tool_category``, resolve o msisdn
|
||
(entrada > bucket pai) e a ``charge_date`` (``period`` quando é data única)."""
|
||
item_type = self._classify(cand.desc, cand.section, cand.default_type, cand.entry)
|
||
if item_type == _OUT_OF_SCOPE_TYPE:
|
||
tool_category = None # não tratável: nunca entra na action_queue
|
||
elif item_type in {"bundle", "estrategico"}:
|
||
tool_category = "vas_estrategico"
|
||
else:
|
||
tool_category = cand.default_tool
|
||
entry_msisdn = (
|
||
str(cand.entry.get("msisdn") or "").strip() or cand.parent_msisdn
|
||
)
|
||
# ``charge_date`` só quando o ``period`` é uma data ÚNICA de cobrança — faixa
|
||
# (ciclo da fatura) ou ausente → None. Usa o MESMO critério do render lean do
|
||
# classificador (``is_period_range``), para que a noção de "cobrança datada"
|
||
# do resolver case o ``data=`` que o classificador viu.
|
||
period = str(cand.entry.get("period") or "").strip()
|
||
charge_date = period if (period and not is_period_range(period)) else None
|
||
return ResolvedInvoiceItem(
|
||
canonical_name=cand.desc,
|
||
tool_category=tool_category, # type: ignore[arg-type]
|
||
item_type=item_type, # type: ignore[arg-type]
|
||
msisdn=entry_msisdn,
|
||
value=self._parse_money(cand.entry.get("value")),
|
||
section=cand.section,
|
||
charge_date=charge_date,
|
||
raw_entry=dict(cand.entry),
|
||
)
|
||
|
||
def _classify(
|
||
self,
|
||
desc: str,
|
||
section: str,
|
||
default_type: str,
|
||
entry: dict[str, Any] | None = None,
|
||
) -> str:
|
||
"""Classifica o tipo do item pela seção da fatura, respeitando o flag
|
||
estratégico carimbado pelo parser na própria entry.
|
||
|
||
Fonte de verdade do roteamento: o flag explícito ``estrategico``/
|
||
``classe`` que o ``bill_processor`` carimba tem precedência. Sem flag, a
|
||
seção é a fonte de verdade (avulso por default em SVA/Eventuais).
|
||
"""
|
||
if isinstance(entry, dict):
|
||
classe = self._normalize_match_text(str(entry.get("classe") or ""))
|
||
if entry.get("estrategico") is True:
|
||
return "estrategico"
|
||
if classe in {"estrategico", "estrategica", "strategic"}:
|
||
return "estrategico"
|
||
if classe in {"bundle"}:
|
||
return "bundle"
|
||
if classe in {"avulso", "avulsa"}:
|
||
return "avulso"
|
||
if section == "Serviços Bundle Inclusos":
|
||
return "bundle"
|
||
if section == "Serviços Contratados de Terceiros":
|
||
return "estrategico"
|
||
if section == "Streamings":
|
||
return "estrategico"
|
||
if section == "Serviços de valor adicionado":
|
||
if isinstance(entry, dict) and entry.get("contestable") is False:
|
||
return "estrategico"
|
||
return "avulso"
|
||
return default_type
|
||
|
||
@staticmethod
|
||
def _normalize_match_text(value: str) -> str:
|
||
"""Monta a CHAVE de comparação de uma menção/``desc``.
|
||
|
||
Usada SOMENTE para casar menção × ``desc`` — NUNCA muta ``canonical_name``,
|
||
valores de payload, campos de API ou texto ao cliente (o item resolvido
|
||
sempre carrega o ``desc`` cru da fatura). Passos:
|
||
|
||
- NFKD + remoção de acentos;
|
||
- lower;
|
||
- ``+`` e qualquer não-alfanumérico viram separador de token;
|
||
- tokens de conexão falados (:data:`_CONNECTOR_TOKENS`: ``mais``/``e``)
|
||
são descartados (a palavra ``de`` é preservada — só o token isolado
|
||
``e`` sai);
|
||
- ``filmes`` → ``filme`` (plural raso comum em voz);
|
||
- colapso para tokens separados por um único espaço.
|
||
|
||
Ex.: ``"VOD + Canais Abertos + Fechados"``, ``"VOD +Canais Abertos
|
||
+Fechados"`` e ``"VOD mais Canais Abertos mais Fechados"`` →
|
||
``"vod canais abertos fechados"``; ``"Aluguel de Filmes 2"`` →
|
||
``"aluguel de filme 2"``."""
|
||
decomposed = unicodedata.normalize("NFKD", str(value or ""))
|
||
ascii_text = "".join(
|
||
ch for ch in decomposed if not unicodedata.combining(ch)
|
||
)
|
||
tokens = re.split(r"[^a-z0-9]+", ascii_text.lower())
|
||
out: list[str] = []
|
||
for tok in tokens:
|
||
if not tok or tok in _CONNECTOR_TOKENS:
|
||
continue
|
||
out.append("filme" if tok == "filmes" else tok)
|
||
return " ".join(out)
|
||
|
||
@staticmethod
|
||
def _was_mentioned(desc: str, mentioned_items: list[str]) -> bool:
|
||
"""Match bidirecional sobre a chave normalizada
|
||
(:meth:`_normalize_match_text`): o ``desc`` contém a menção OU a menção
|
||
contém o ``desc``. A normalização cobre variações de grafia/fala (``+`` vs
|
||
``mais`` vs espaçamento) sem mexer no nome cru."""
|
||
desc_norm = InvoiceResolver._normalize_match_text(desc)
|
||
if not desc_norm:
|
||
return False
|
||
for mention in mentioned_items:
|
||
if not isinstance(mention, str):
|
||
continue
|
||
m = InvoiceResolver._normalize_match_text(mention)
|
||
if not m:
|
||
continue
|
||
if m in desc_norm or desc_norm in m:
|
||
return True
|
||
return False
|
||
|
||
@staticmethod
|
||
def _parse_money(value: Any) -> Decimal | None:
|
||
"""Converte ``value`` (str/int/float/Decimal) em ``Decimal`` com 2
|
||
casas. Devolve ``None`` em qualquer falha — a Policy/Composer decide
|
||
como agir sem o valor."""
|
||
if value is None or value == "":
|
||
return None
|
||
if isinstance(value, Decimal):
|
||
return value.quantize(Decimal("0.01"))
|
||
try:
|
||
return Decimal(str(value)).quantize(Decimal("0.01"))
|
||
except (InvalidOperation, ValueError, TypeError):
|
||
return None
|
||
|
||
@staticmethod
|
||
def _dedupe_exact_matches(
|
||
items: list[ResolvedInvoiceItem],
|
||
*,
|
||
by_charge: bool = False,
|
||
) -> list[ResolvedInvoiceItem]:
|
||
"""Mantém apenas a primeira ocorrência de cada chave de identidade. Itens
|
||
parecidos mas com nomes diferentes (ex.: "Aluguel de Filme 1" vs "Aluguel
|
||
de Filme 2") permanecem ambos — só duplicatas exatas saem.
|
||
|
||
``by_charge=False`` (default): chave ``(canonical_name, msisdn, section)`` —
|
||
colapsa duas cobranças do mesmo serviço na mesma linha. Preserva o contrato
|
||
de ``resolve_items``/``build_snapshot`` e o comportamento em fatura sem
|
||
cobrança duplicada. ``by_charge=True``: chave inclui ``charge_date``, então
|
||
duas cobranças datadas distintas do mesmo ``(nome, msisdn, section)`` SOBREVIVEM
|
||
(desambiguação de cobrança). Cobranças sem data (``charge_date`` None) ainda
|
||
colapsam — indistinguíveis."""
|
||
seen: set[tuple[str, ...]] = set()
|
||
out: list[ResolvedInvoiceItem] = []
|
||
for item in items:
|
||
key: tuple[str, ...] = (item.canonical_name, item.msisdn, item.section)
|
||
if by_charge:
|
||
key = (*key, item.charge_date or "")
|
||
if key in seen:
|
||
continue
|
||
seen.add(key)
|
||
out.append(item)
|
||
return out
|
||
|
||
@staticmethod
|
||
def has_analyzable_buckets(invoice_detail: dict[str, Any]) -> bool:
|
||
"""O payload tem ao menos UM bucket de linha (msisdn) com conteúdo?
|
||
|
||
Separa "fatura sem VAS" de "não veio fatura". Um ``invoice_detail`` só com
|
||
chaves de metadado (``vocalized_msisdn``/``Fatura Resumo``/``DANFE-COM``) é
|
||
um dict NÃO-vazio, mas não tem nada para classificar — é o que sobra quando
|
||
o PDF não pôde ser processado. Sem esta distinção, quem consome o snapshot
|
||
lê "nenhum VAS" e decide como se a fatura estivesse completa (incidente
|
||
Neymar Jr.Experience: o ramo NÃO encerrou a ligação por falta de dado).
|
||
|
||
Reusa ``_iter_msisdn_buckets`` — a mesma fonte de verdade de quais chaves
|
||
são metadado — em vez de duplicar a lista."""
|
||
for _msisdn, sections in InvoiceResolver._iter_msisdn_buckets(invoice_detail):
|
||
if sections:
|
||
return True
|
||
return False
|
||
|
||
@staticmethod
|
||
def _iter_msisdn_buckets(
|
||
invoice_detail: dict[str, Any],
|
||
) -> Iterable[tuple[str, Any]]:
|
||
"""Itera apenas os buckets por msisdn do top-level do invoice_detail,
|
||
ignorando chaves de metadado (Fatura Resumo, DANFE-COM, vocalized_msisdn)."""
|
||
converted = InvoiceResolver._billing_analysis_sections(invoice_detail)
|
||
if converted:
|
||
yield "", converted
|
||
for key, value in invoice_detail.items():
|
||
if key in _NON_MSISDN_KEYS:
|
||
continue
|
||
if key in {"currentInvoice", "invoiceVariation"}:
|
||
continue
|
||
converted = InvoiceResolver._billing_analysis_sections(value)
|
||
yield str(key), converted or value
|
||
|
||
@staticmethod
|
||
def _billing_analysis_sections(value: Any) -> dict[str, list[Any]]:
|
||
if not isinstance(value, dict):
|
||
return {}
|
||
mapped: dict[str, list[Any]] = {}
|
||
for key in ("currentInvoice", "invoiceVariation"):
|
||
sections = value.get(key)
|
||
if not isinstance(sections, list):
|
||
continue
|
||
for section in sections:
|
||
if not isinstance(section, dict):
|
||
continue
|
||
section_name = str(section.get("desc") or section.get("type") or "")
|
||
items = section.get("items")
|
||
if section_name and isinstance(items, list):
|
||
mapped.setdefault(section_name, []).extend(items)
|
||
return mapped
|
||
|
||
|
||
__all__ = [
|
||
"STRATEGIC_NAMES",
|
||
"SECTION_DEFAULTS",
|
||
"ItemMatcherError",
|
||
"ItemMatcherLLM",
|
||
"ItemResolution",
|
||
"ResolutionOutcome",
|
||
"InvoiceResolver",
|
||
]
|