233 lines
8.9 KiB
Python
233 lines
8.9 KiB
Python
"""
|
|
Casa cada Individuo (extraído do arquivo da operadora) com a linha
|
|
correspondente na planilha padrão do sistema.
|
|
|
|
Duas estratégias, escolhidas pelo `chave_casamento` da operadora:
|
|
|
|
- "cpf" -> casamento por CPF normalizado (Amil). Mais seguro, porque o
|
|
CPF não muda nunca, independente de erro de grafia no nome.
|
|
|
|
- "nome" -> casamento por nome normalizado, ESCOPADO à família (Unimed,
|
|
que não manda CPF nenhum). Primeiro acha o titular pelo nome; só
|
|
então procura o dependente dentro das linhas daquele titular
|
|
específico na planilha (isso evita ambiguidade quando o mesmo
|
|
dependente aparece cadastrado em mais de uma família, o que
|
|
encontramos de fato nos dados reais). Nome que não bate 100% exato
|
|
(ex: "SCHAEFER" vs "SCHAFER") NUNCA é resolvido por aproximação — vai
|
|
direto para auditoria, por decisão explícita do cliente.
|
|
|
|
Regras de negócio combinadas com o cliente:
|
|
- O valor do mês entra na coluna VALOR (desconto do empregado), não em
|
|
VALOREMPRESA.
|
|
- Beneficiário com valor final negativo no mês (ex: devolução retroativa
|
|
de inativo) NÃO é lançado na planilha — vai para a auditoria.
|
|
- Beneficiário sem cadastro correspondente na planilha padrão (plano
|
|
não cadastrado para ele) também NÃO é lançado — vai para a auditoria.
|
|
"""
|
|
from typing import Dict, List, Tuple
|
|
|
|
from core.leiaute_sistema import formata_valor_br
|
|
from core.modelos import Individuo, ItemAuditoria, LinhaSistema, normaliza_nome
|
|
|
|
|
|
# ----------------------------------------------------------------------
|
|
# Indexação da planilha padrão
|
|
# ----------------------------------------------------------------------
|
|
|
|
def _indexa_por_cpf(linhas: List[LinhaSistema]):
|
|
titulares, dependentes = {}, {}
|
|
for l in linhas:
|
|
if l.eh_linha_titular():
|
|
titulares[l.cpf_func_normalizado()] = l
|
|
else:
|
|
dependentes[l.cpf_dependente_normalizado()] = l
|
|
return titulares, dependentes
|
|
|
|
|
|
def _indexa_por_nome(linhas: List[LinhaSistema]):
|
|
"""
|
|
titulares: nome normalizado -> LinhaSistema (a linha "do titular")
|
|
dependentes_por_titular: nome normalizado do titular -> {nome
|
|
normalizado do dependente -> LinhaSistema}
|
|
"""
|
|
titulares: Dict[str, LinhaSistema] = {}
|
|
dependentes_por_titular: Dict[str, Dict[str, LinhaSistema]] = {}
|
|
for l in linhas:
|
|
nome_titular_norm = l.nome_func_normalizado()
|
|
if l.eh_linha_titular():
|
|
titulares[nome_titular_norm] = l
|
|
else:
|
|
dependentes_por_titular.setdefault(nome_titular_norm, {})[
|
|
l.nome_dependente_normalizado()
|
|
] = l
|
|
return titulares, dependentes_por_titular
|
|
|
|
|
|
# ----------------------------------------------------------------------
|
|
# Estratégia: CPF
|
|
# ----------------------------------------------------------------------
|
|
|
|
def _casa_por_cpf(
|
|
individuos: List[Individuo], linhas_sistema: List[LinhaSistema], **_ignorado
|
|
) -> Tuple[List[LinhaSistema], List[ItemAuditoria]]:
|
|
titulares, dependentes = _indexa_por_cpf(linhas_sistema)
|
|
auditoria: List[ItemAuditoria] = []
|
|
|
|
for ind in individuos:
|
|
if ind.valor_total < 0:
|
|
auditoria.append(_item_valor_negativo(ind))
|
|
continue
|
|
|
|
cpf_norm = ind.cpf_normalizado
|
|
linha_destino = (
|
|
titulares.get(cpf_norm) if ind.tipo == "T" else dependentes.get(cpf_norm)
|
|
)
|
|
if linha_destino is None:
|
|
auditoria.append(_item_nao_cadastrado(ind, "CPF não encontrado na planilha padrão do sistema."))
|
|
continue
|
|
|
|
linha_destino.valor = formata_valor_br(ind.valor_total)
|
|
|
|
return linhas_sistema, auditoria
|
|
|
|
|
|
# ----------------------------------------------------------------------
|
|
# Estratégia: Nome (escopado por família via numero_titular)
|
|
# ----------------------------------------------------------------------
|
|
|
|
def _casa_por_nome(
|
|
individuos: List[Individuo],
|
|
linhas_sistema: List[LinhaSistema],
|
|
nomes_titular_por_numero: Dict[str, str] = None,
|
|
**_ignorado,
|
|
) -> Tuple[List[LinhaSistema], List[ItemAuditoria]]:
|
|
"""
|
|
`nomes_titular_por_numero` mapeia numero_titular (id da família no
|
|
arquivo da operadora) -> nome do titular. É construído a partir de
|
|
TODOS os indivíduos extraídos (todos os tipos de lançamento juntos),
|
|
porque um titular pode não ter nenhuma linha de um tipo específico
|
|
naquele mês (ex: só o dependente foi ao médico, o titular não teve
|
|
coparticipação) — sem esse mapa global, a família "perderia" o
|
|
titular ao filtrar por tipo antes de casar.
|
|
"""
|
|
titulares, dependentes_por_titular = _indexa_por_nome(linhas_sistema)
|
|
nomes_titular_por_numero = nomes_titular_por_numero or {}
|
|
auditoria: List[ItemAuditoria] = []
|
|
|
|
# Agrupa os indivíduos por família (numero_titular do arquivo da
|
|
# operadora) para resolver o titular uma vez e escopar os dependentes.
|
|
familias: Dict[str, List[Individuo]] = {}
|
|
for ind in individuos:
|
|
familias.setdefault(ind.numero_titular or ind.numero_beneficiario, []).append(ind)
|
|
|
|
for chave_familia, membros in familias.items():
|
|
nome_titular = nomes_titular_por_numero.get(chave_familia)
|
|
if nome_titular is None:
|
|
# nenhum indivíduo (de nenhum tipo) foi identificado como
|
|
# titular dessa família no arquivo da operadora
|
|
for m in membros:
|
|
auditoria.append(_item_nao_cadastrado(
|
|
m, "Família sem titular identificado no arquivo da operadora."
|
|
))
|
|
continue
|
|
|
|
nome_titular_norm = normaliza_nome(nome_titular)
|
|
linha_titular = titulares.get(nome_titular_norm)
|
|
|
|
if linha_titular is None:
|
|
for m in membros:
|
|
auditoria.append(_item_nao_cadastrado(
|
|
m, f"Titular '{nome_titular}' não encontrado (por nome) na planilha padrão."
|
|
))
|
|
continue
|
|
|
|
dependentes_da_familia = dependentes_por_titular.get(nome_titular_norm, {})
|
|
|
|
for m in membros:
|
|
if m.valor_total < 0:
|
|
auditoria.append(_item_valor_negativo(m))
|
|
continue
|
|
|
|
if m.tipo == "T":
|
|
linha_titular.valor = formata_valor_br(m.valor_total)
|
|
continue
|
|
|
|
linha_dep = dependentes_da_familia.get(m.nome_normalizado)
|
|
if linha_dep is None:
|
|
auditoria.append(ItemAuditoria(
|
|
motivo="NOME_DIVERGENTE",
|
|
numero_beneficiario=m.numero_beneficiario,
|
|
nome=m.nome,
|
|
cpf="",
|
|
tipo=m.tipo,
|
|
valor=m.valor_total,
|
|
tipo_lancamento=m.tipo_lancamento,
|
|
detalhe=(
|
|
f"Dependente '{m.nome}' não encontrado (nome exato) "
|
|
f"entre os dependentes de '{nome_titular}' na planilha "
|
|
f"padrão. Verificar grafia do nome — não é lançado "
|
|
f"automaticamente por decisão do cliente."
|
|
),
|
|
))
|
|
continue
|
|
|
|
linha_dep.valor = formata_valor_br(m.valor_total)
|
|
|
|
return linhas_sistema, auditoria
|
|
|
|
|
|
# ----------------------------------------------------------------------
|
|
# Helpers de auditoria
|
|
# ----------------------------------------------------------------------
|
|
|
|
def _item_valor_negativo(ind: Individuo) -> ItemAuditoria:
|
|
return ItemAuditoria(
|
|
motivo="VALOR_NEGATIVO",
|
|
numero_beneficiario=ind.numero_beneficiario,
|
|
nome=ind.nome,
|
|
cpf=ind.cpf,
|
|
tipo=ind.tipo,
|
|
valor=ind.valor_total,
|
|
tipo_lancamento=ind.tipo_lancamento,
|
|
detalhe=(
|
|
"Valor final do mês é negativo (ex: devolução/estorno retroativo "
|
|
"maior que a cobrança normal). Não lançado automaticamente — "
|
|
"revisar manualmente. Rubricas: " + " | ".join(ind.rubricas)
|
|
),
|
|
)
|
|
|
|
|
|
def _item_nao_cadastrado(ind: Individuo, motivo_texto: str) -> ItemAuditoria:
|
|
return ItemAuditoria(
|
|
motivo="NAO_CADASTRADO",
|
|
numero_beneficiario=ind.numero_beneficiario,
|
|
nome=ind.nome,
|
|
cpf=ind.cpf,
|
|
tipo=ind.tipo,
|
|
valor=ind.valor_total,
|
|
tipo_lancamento=ind.tipo_lancamento,
|
|
detalhe=motivo_texto,
|
|
)
|
|
|
|
|
|
# ----------------------------------------------------------------------
|
|
# Ponto de entrada único usado pelo main.py
|
|
# ----------------------------------------------------------------------
|
|
|
|
ESTRATEGIAS = {
|
|
"cpf": _casa_por_cpf,
|
|
"nome": _casa_por_nome,
|
|
}
|
|
|
|
|
|
def casa_individuos_com_planilha(
|
|
individuos: List[Individuo],
|
|
linhas_sistema: List[LinhaSistema],
|
|
chave_casamento: str = "cpf",
|
|
**kwargs,
|
|
) -> Tuple[List[LinhaSistema], List[ItemAuditoria]]:
|
|
estrategia = ESTRATEGIAS.get(chave_casamento)
|
|
if estrategia is None:
|
|
raise ValueError(f"chave_casamento desconhecida: {chave_casamento!r}")
|
|
return estrategia(individuos, linhas_sistema, **kwargs)
|