306 lines
13 KiB
Python
306 lines
13 KiB
Python
#!/usr/bin/env python3
|
|
"""Transformacao da fatura normalizada (dict JSON) -> formato textual em blocos.
|
|
|
|
Formato economico em tokens (~72% menor que o JSON indentado), pensado para enviar
|
|
a uma LLM. API publica: `to_text(data)` e a flag `USE_TEXT_FORMAT`.
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
import logging
|
|
import os
|
|
import re
|
|
from typing import Any
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
# Liga o formato textual da fatura no prompt. Default conceitual: False;
|
|
# mantido True nesta fase de rollout. Override por env, sem edicao de codigo.
|
|
USE_TEXT_FORMAT = os.getenv("TIM_INVOICE_DETAIL_AS_TEXT", "true").lower() == "true"
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# secao -> (acao, classe). O cabecalho usa o proprio nome da secao (mantido como no JSON).
|
|
SEC_META = {
|
|
"SVA Detalhe Total": ("cancelar", "avulso"),
|
|
"Itens Eventuais": ("cancelar", "avulso"),
|
|
"Serviços Bundle Inclusos": ("falar sobre", "bundle"),
|
|
}
|
|
|
|
# Secoes que rendem a acao por ITEM (sem "| ação=... | classe" no header).
|
|
# Item estrategico (estrategico=True) usa "falar sobre"; avulso, "cancelar".
|
|
PER_ITEM_ACTION_SECTIONS = {
|
|
"SVA Detalhe Total",
|
|
"Itens Eventuais",
|
|
"Mensalidades Adicionais",
|
|
}
|
|
|
|
# Secoes cujo header recebe o rotulo "| cobrados a parte" (servicos avulsos
|
|
# contratados separadamente). A acao continua por item.
|
|
COBRADOS_A_PARTE_SECTIONS = {"SVA Detalhe Total", "Itens Eventuais"}
|
|
|
|
# secao do JSON -> rotulo no cabecalho do prompt. "SVA Detalhe Total" e "Itens
|
|
# Eventuais" sao a MESMA categoria (avulsos cobrados a parte) e saem sob um unico
|
|
# cabecalho "Itens Eventuais". "Serviços Bundle Inclusos" vira "Serviços Inclusos
|
|
# no Plano": a palavra "bundle" e tecnica/interna e nao deve chegar ao prompt (nem
|
|
# ao LLM). As chaves do JSON normalizado NAO mudam nos dois casos.
|
|
SECTION_LABELS = {
|
|
"SVA Detalhe Total": "Itens Eventuais",
|
|
"Serviços Bundle Inclusos": "Benefícios do Plano",
|
|
}
|
|
|
|
# classe interna -> rotulo exibido no cabecalho (ver SEC_META). Mesma logica do
|
|
# SECTION_LABELS: so a APRESENTACAO muda, a classe interna ("bundle") segue igual
|
|
# em todo o resto do backend (bill_processor, invoice_resolver, tools, etc.).
|
|
CLASSE_LABELS = {"bundle": "incluso"}
|
|
|
|
_RX_OUTROS = re.compile(r"^\s*([^:()]+?)\s*:\s*\((.+)\)\s*$")
|
|
|
|
|
|
def _fmt_money(value: Any) -> str:
|
|
if isinstance(value, (int, float)):
|
|
return f"{value:.2f}"
|
|
return str(value)
|
|
|
|
|
|
def _snake(desc: Any) -> str:
|
|
return re.sub(r"\s+", "_", str(desc).strip().lower())
|
|
|
|
|
|
def is_period_range(period: Any) -> bool:
|
|
"""True quando ``period`` é uma FAIXA (ciclo da fatura, ex.: "14/10 a 13/11"),
|
|
não uma data única de cobrança. Compartilhado com o resolver, que usa o mesmo
|
|
critério para decidir se uma entry expõe ``charge_date`` (data única) — assim o
|
|
que o resolver enxerga como cobrança datada casa exatamente o ``data=`` do render
|
|
lean do classificador."""
|
|
text = str(period)
|
|
return " a " in text or "~" in text
|
|
|
|
|
|
def _item_acao(item: dict[str, Any], section: str) -> str | None:
|
|
"""Verbo de acao por item: estrategico -> 'falar sobre'; senao, o default da secao."""
|
|
if item.get("estrategico"):
|
|
return "falar sobre"
|
|
acao, _ = SEC_META.get(section, (None, None))
|
|
return acao
|
|
|
|
|
|
def _render_item(item: dict[str, Any], section: str) -> str:
|
|
"""<nome solto> | campo=valor | ... (com tratamento especial p/ Outros Valores)."""
|
|
desc = str(item.get("desc", "")).strip()
|
|
value = item.get("value")
|
|
|
|
if section == "Outros Valores":
|
|
match = _RX_OUTROS.match(desc)
|
|
if match:
|
|
parts = [match.group(1).strip().lower()]
|
|
if value is not None and value != "Incluído":
|
|
parts.append(f"valor={_fmt_money(value)}")
|
|
parts.append(f'ref="{match.group(2).strip()}"')
|
|
return " | ".join(parts)
|
|
|
|
parts = [desc]
|
|
if value is not None and value != "Incluído":
|
|
parts.append(f"valor={_fmt_money(value)}")
|
|
period = item.get("period")
|
|
if period and not is_period_range(period):
|
|
parts.append(f"data={period}")
|
|
if item.get("franchise"):
|
|
parts.append(f"franquia={item['franchise']}")
|
|
if item.get("consumption"):
|
|
parts.append(f"consumo={item['consumption']}")
|
|
if item.get("installment"):
|
|
parts.append(f"parcela={item['installment']}")
|
|
return " | ".join(parts)
|
|
|
|
|
|
# Seções omitidas no modo enxuto (intent classifier): o classificador só precisa
|
|
# dos NOMES dos itens por linha/seção, não de totais, planos nem juros/multas.
|
|
_LEAN_SKIP_SECTIONS = frozenset({"Fatura Resumo", "Planos", "Outros Valores"})
|
|
|
|
|
|
def to_text(data: dict[str, Any], *, lean: bool = False) -> str:
|
|
"""Converte o JSON normalizado de uma fatura no formato textual em blocos.
|
|
|
|
``lean=True`` (formato do intent classifier): mantém a estrutura LINHA/seção,
|
|
o NOME dos itens e o ``valor``/``data`` por item — descarta "Fatura Resumo",
|
|
"Planos", "Outros Valores" e os demais campos por-item (ação, type, franquia,
|
|
consumo). O ``valor``/``data`` distinguem cobranças duplicadas do mesmo nome na
|
|
mesma linha (desambiguação de cobrança). O orquestrador usa ``lean=False``
|
|
(formato completo, com ações)."""
|
|
lines: list[str] = []
|
|
msisdn_count = 0
|
|
|
|
# FATURA (achatada) — omitida no modo enxuto.
|
|
resumo = data.get("Fatura Resumo")
|
|
if not lean and isinstance(resumo, list):
|
|
lines.append("Fatura Resumo")
|
|
for item in resumo:
|
|
desc = item.get("desc", "?")
|
|
if "period" in item:
|
|
lines.append(f" {_snake(desc)}={item['period']}")
|
|
elif "emissao" in item:
|
|
lines.append(f" {_snake(desc)}={item['emissao']}")
|
|
elif "value" in item:
|
|
lines.append(f" {_snake(desc)}={_fmt_money(item['value'])}")
|
|
lines.append("")
|
|
|
|
# LINHAS (MSISDN)
|
|
for key, block in data.items():
|
|
if key in ("Fatura Resumo", "vocalized_msisdn"):
|
|
continue
|
|
if not isinstance(block, dict):
|
|
continue
|
|
|
|
msisdn_count += 1
|
|
lines.append(f"LINHA {key}")
|
|
|
|
# Planos — omitido no modo enxuto.
|
|
planos = block.get("Planos")
|
|
if not lean and isinstance(planos, dict) and planos:
|
|
lines.append("Planos")
|
|
for nome, plano in planos.items():
|
|
bits = [f"nome={nome}"]
|
|
if plano.get("period"):
|
|
bits.append(f"período={plano['period']}")
|
|
if plano.get("days") is not None:
|
|
bits.append(f"dias={plano['days']}")
|
|
bits.append(f"valor_final={_fmt_money(plano.get('valor_final'))}")
|
|
if plano.get("valor_bruto") not in (None, plano.get("valor_final")):
|
|
bits.append(f"valor_bruto={_fmt_money(plano.get('valor_bruto'))}")
|
|
if plano.get("total_descontos"):
|
|
bits.append(f"total_desc={_fmt_money(plano.get('total_descontos'))}")
|
|
if plano.get("is_controle"):
|
|
bits.append("controle=sim")
|
|
lines.append(" " + " | ".join(bits))
|
|
descontos = plano.get("descontos") or []
|
|
if descontos:
|
|
lines.append(" descontos:")
|
|
for desconto in descontos:
|
|
dbits = [
|
|
f"nome={desconto.get('desc')}",
|
|
f"valor={_fmt_money(desconto.get('value'))}",
|
|
]
|
|
if desconto.get("installment"):
|
|
dbits.append(f"parcela={desconto['installment']}")
|
|
lines.append(" " + " | ".join(dbits))
|
|
|
|
# Seções agrupadas pelo RÓTULO do cabeçalho (``SECTION_LABELS``): as que
|
|
# compartilham rótulo saem sob um único cabeçalho, na posição da primeira
|
|
# delas. Cada item guarda a seção de ORIGEM, que é quem decide a ação/classe
|
|
# por item — o rótulo é só apresentação.
|
|
groups: dict[str, tuple[str, list[tuple[str, dict[str, Any]]]]] = {}
|
|
for section, items in block.items():
|
|
if section == "Planos" or not isinstance(items, list):
|
|
continue
|
|
if lean and section in _LEAN_SKIP_SECTIONS:
|
|
continue
|
|
label = SECTION_LABELS.get(section, section)
|
|
_, entries = groups.setdefault(label, (section, []))
|
|
entries.extend((section, item) for item in items)
|
|
|
|
for label, (first_section, entries) in groups.items():
|
|
acao, classe = SEC_META.get(first_section, (None, None))
|
|
header = label
|
|
# No modo enxuto o cabeçalho é só o nome da seção (sem ação/classe).
|
|
if not lean and first_section in COBRADOS_A_PARTE_SECTIONS:
|
|
header += " | cobrados a parte"
|
|
elif acao and first_section not in PER_ITEM_ACTION_SECTIONS and not lean:
|
|
header += f" | ação={acao} | {CLASSE_LABELS.get(classe, classe)}"
|
|
lines.append(header)
|
|
for section, item in entries:
|
|
if lean:
|
|
# Nome do item + valor + data da cobrança (sem ação/type/
|
|
# franquia/consumo). O valor/data por item permitem o classifier
|
|
# DISTINGUIR N cobranças do mesmo nome na mesma linha (mesmo
|
|
# desc/msisdn, períodos diferentes) na desambiguação de cobrança
|
|
# duplicada — o msisdn vem do cabeçalho ``LINHA``. Itens
|
|
# ``Incluído`` (bundle) e períodos em faixa (ciclo da fatura, não
|
|
# data de cobrança) seguem só com o nome, como antes.
|
|
desc = str(item.get("desc", "")).strip()
|
|
if not desc:
|
|
continue
|
|
parts = [desc]
|
|
value = item.get("value")
|
|
if value is not None and value != "Incluído":
|
|
parts.append(f"valor={_fmt_money(value)}")
|
|
period = item.get("period")
|
|
if period and not is_period_range(period):
|
|
parts.append(f"data={period}")
|
|
lines.append(" " + " | ".join(parts))
|
|
continue
|
|
line = " " + _render_item(item, section)
|
|
if section in PER_ITEM_ACTION_SECTIONS:
|
|
item_acao = _item_acao(item, section)
|
|
if item_acao:
|
|
line += f" | ação={item_acao}"
|
|
if item.get("estrategico"):
|
|
line += " | type=estrategico"
|
|
lines.append(line)
|
|
lines.append("")
|
|
|
|
text = "\n".join(lines).rstrip()
|
|
if msisdn_count > 1:
|
|
text += "\n\nmultiplas_linhas = true"
|
|
else:
|
|
text += "\n\nmultiplas_linhas = false"
|
|
return text + "\n"
|
|
|
|
|
|
# ordem pedida pelo produto: avulso -> estrategico -> bundle
|
|
# (difere da ordem informacional bundle/estrategico/avulso usada em backend.py)
|
|
_VAS_PRODUCT_ORDER = ("avulso", "estrategico", "bundle")
|
|
|
|
|
|
def get_vas_product_names(data: dict[str, Any]) -> str:
|
|
"""Nomes dos VAS (avulso, estrategico, bundle — nessa ordem) da fatura
|
|
normalizada, deduplicados e separados por vírgula. Ex.: "Focus Mensal,
|
|
Tamboro Mensal". Ignora plano e demais seções não-VAS. Retorna "" se não houver."""
|
|
buckets: dict[str, list[str]] = {c: [] for c in _VAS_PRODUCT_ORDER}
|
|
seen: set[str] = set()
|
|
for key, block in data.items():
|
|
if key in ("Fatura Resumo", "vocalized_msisdn") or not isinstance(block, dict):
|
|
continue
|
|
for section, items in block.items():
|
|
if section == "Planos" or not isinstance(items, list):
|
|
continue
|
|
for item in items:
|
|
classe = item.get("classe")
|
|
if not classe and item.get("estrategico"):
|
|
classe = "estrategico"
|
|
if classe not in buckets:
|
|
continue
|
|
name = str(item.get("desc", "")).strip()
|
|
if not name or name in seen:
|
|
continue
|
|
seen.add(name)
|
|
buckets[classe].append(name)
|
|
ordered = [n for c in _VAS_PRODUCT_ORDER for n in buckets[c]]
|
|
return ", ".join(ordered)
|
|
|
|
|
|
def render_for_prompt(invoice_detail: Any, *, lean: bool = False) -> str | None:
|
|
"""Renderiza ``invoice_detail`` (dict ou str JSON) no formato textual ``to_text``
|
|
para uso no prompt. Fonte única compartilhada pelo orquestrador (system prompt)
|
|
e pelo intent classifier — garante que ambos leiam a MESMA fatura, sem drift.
|
|
|
|
``lean=True`` produz o formato enxuto do intent classifier (nomes + valor/data
|
|
por item, por linha/seção; ver :func:`to_text`); o orquestrador usa ``lean=False``.
|
|
|
|
Retorna ``None`` quando o formato textual está desligado (``USE_TEXT_FORMAT``
|
|
false) ou em qualquer falha de parse/render — cabe ao chamador decidir o
|
|
fallback (JSON cru no orquestrador; renderizador compacto no classifier).
|
|
Nunca levanta exceção: o prompt não pode quebrar por causa da fatura."""
|
|
if not (USE_TEXT_FORMAT and invoice_detail):
|
|
return None
|
|
try:
|
|
parsed = (
|
|
json.loads(invoice_detail)
|
|
if isinstance(invoice_detail, str)
|
|
else invoice_detail
|
|
)
|
|
if isinstance(parsed, dict) and parsed:
|
|
return to_text(parsed, lean=lean)
|
|
except Exception: # noqa: BLE001 — nunca quebrar o prompt
|
|
logger.warning("invoice_to_text.render_for_prompt falhou", exc_info=True)
|
|
return None
|