portal_publico/portal_api/planos_saude/matcher.py

463 lines
21 KiB
Python
Raw Blame History

This file contains ambiguous Unicode characters

This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.

"""
Casa cada Individuo (extraído do arquivo da operadora) com a linha
correspondente na planilha padrão do sistema.
Porta de projects/project/core/matcher.py — mesmas duas estratégias
("cpf"/"nome", ver docstring original de cada uma abaixo), com UM
acréscimo: `regra_custeio` decide como o valor do mês se divide entre
`valor_empresa` (custeado pela empresa) e `valor` (desconto do
empregado) — antes disso o valor cheio sempre ia pra `valor`, fixo.
Essa é a única diferença de comportamento em relação ao pipeline
original; quem decide a regra é o usuário na tela de nova importação,
por tipo de lançamento **e** por tipo de beneficiário (titular ou
dependente/agregado) — ex.: mensalidade do titular custeada pela
empresa, mensalidade do dependente descontada do empregado (ver
`_regra_para_pessoa`/`_calcula_valores` abaixo pro formato).
- "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:
- 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.
Segundo acréscimo: `regra_empresa_fn` (opcional, implementado nas duas
estratégias — `_casa_por_nome` acumula por FAMÍLIA, `_casa_por_cpf`
acumula tudo de uma vez já que essa estratégia não tem noção de família)
substitui `regra_custeio` por uma função de custeio especial cadastrada em
código pra regras que não cabem no desenho normal titular/dependente ×
empresa/empregado/específica (ver portal_api.planos_saude.regras_empresa)
— resolvida e validada em pipeline.processa_importacao antes de chegar
aqui, e recebe `tipo_lancamento` como segundo argumento (a mesma função
pode se comportar diferente pra mensalidade e coparticipação, ver
REGRAS_EMPRESA).
Terceiro acréscimo: `vinculos_por_nome` (opcional, só na estratégia
"nome") — um "DE/PARA" persistente (`portal_api.models.VinculoNomeOperadora`,
representado aqui sem ORM como `modelos.VinculoNome`) que reaproveita uma
confirmação humana anterior ("Vincular pessoa", ver
ImportacaoPlanoSaudeAuditoriaViewSet.resolver em views.py) pra resolver
automaticamente a MESMA divergência de nome numa importação futura, sem
precisar vincular de novo todo mês. Continua não sendo aproximação — só
existe depois que alguém confirmou explicitamente aquele nome específico
uma vez; sem vínculo salvo, o comportamento é idêntico a antes (auditoria).
Cada aplicação automática é registrada em `VinculoAplicado` e devolvida
pra quem chamou (pipeline.py/views.py) montar um `ImportacaoPlanoSaudeAlteracao`
("vínculo automático") depois que as linhas forem persistidas — ver
"Vínculos de nome salvos (DE/PARA)" no CLAUDE.md.
"""
from typing import Callable, Dict, List, Optional, Tuple
from portal_api.planos_saude.leiaute_sistema import formata_valor_br
from portal_api.planos_saude.modelos import (
Individuo,
ItemAuditoria,
LinhaSistema,
VinculoAplicado,
VinculoNome,
normaliza_nome,
)
REGRA_CUSTEIO_PADRAO = {"modo": "empregado"}
def _calcula_valores(valor_total: float, regra: Optional[dict]) -> Tuple[float, float]:
"""
Divide `valor_total` entre (valor_empresa, valor_empregado) conforme
`regra` (`{"modo": "empresa"|"empregado"|"especifica", "limite_valor":
float|None, "percentual": float|None, "limite_desconto_empregado":
float|None}`):
- "empresa" -> tudo em valor_empresa.
- "empregado" -> tudo em valor (desconto do empregado) — comportamento
padrão de antes desta funcionalidade existir.
- "especifica" -> duas famílias de critério, mutuamente exclusivas
(garantido por `_monta_regra_custeio`/`serializers.py`, nunca as duas
preenchidas ao mesmo tempo):
- `limite_valor`/`percentual` protegem o gasto da EMPRESA: `limite_valor`
é o teto de quanto ela cobre (excedente vira desconto do empregado);
`percentual` é a fração do valor do mês que ela cobre (o resto vira
desconto). Pode vir só um dos dois ou os dois — quando os dois vêm
juntos, prevalece o mais restritivo (o menor valor entre os dois
critérios), decisão explícita do usuário ("pode ser aplicado apenas
uma destas regras ou as duas").
- `limite_desconto_empregado` protege o gasto do EMPREGADO: teto de
quanto é descontado dele (o restante, sem limite, fica com a
empresa) — direção oposta da anterior, por isso não se combina com
ela.
"""
regra = regra or REGRA_CUSTEIO_PADRAO
modo = regra.get("modo", "empregado")
if modo == "empresa":
return valor_total, 0.0
if modo != "especifica":
return 0.0, valor_total
limite_desconto_empregado = regra.get("limite_desconto_empregado")
if limite_desconto_empregado is not None:
valor_empregado = round(max(0.0, min(valor_total, limite_desconto_empregado)), 2)
valor_empresa = round(valor_total - valor_empregado, 2)
return valor_empresa, valor_empregado
candidatos = [valor_total]
percentual = regra.get("percentual")
if percentual is not None:
candidatos.append(valor_total * percentual / 100)
limite_valor = regra.get("limite_valor")
if limite_valor is not None:
candidatos.append(limite_valor)
# Arredonda valor_empresa primeiro e deriva valor_empregado como o
# complemento exato (valor_total - valor_empresa) — arredondar os dois
# separadamente (ex: 50% de 51,69 = 25,845 dos dois lados) podia perder
# ou sobrar 1 centavo na soma final.
valor_empresa = round(max(0.0, min(candidatos)), 2)
valor_empregado = round(valor_total - valor_empresa, 2)
return valor_empresa, valor_empregado
def _regra_para_pessoa(regra_por_pessoa: Optional[dict], tipo_pessoa: str) -> Optional[dict]:
"""
`regra_por_pessoa` é `{"titular": {...regra...}, "dependente":
{...regra...}}` — cada tipo de lançamento pode ter uma regra de custeio
diferente para titular e para dependente (ex.: mensalidade do titular
custeada pela empresa, mensalidade do dependente descontada do
empregado). `tipo_pessoa` é o `Individuo.tipo`/`LinhaSistema` ('T' =
Titular, 'D'/'A' = Dependente/Agregado — 'D' e 'A' caem no mesmo balde
'dependente', mesmo critério já usado no leiaute do sistema).
"""
chave = "titular" if tipo_pessoa == "T" else "dependente"
return (regra_por_pessoa or {}).get(chave)
def _aplica_regra_custeio(linha: LinhaSistema, valor_total: float, regra: Optional[dict]) -> None:
valor_empresa, valor_empregado = _calcula_valores(valor_total, regra)
linha.valor_empresa = formata_valor_br(valor_empresa)
linha.valor = formata_valor_br(valor_empregado)
def valores_formatados_para_pessoa(
valor_total: float, regra_por_pessoa: Optional[dict], tipo_pessoa: str
) -> Tuple[str, str]:
"""
Único ponto de entrada público deste módulo pra fora do pipeline normal
de casamento — usado pela resolução manual de um item de auditoria (ver
`ImportacaoPlanoSaudeAuditoriaViewSet.resolver` em views.py): dado um
valor já extraído do arquivo da operadora e a regra de custeio salva pro
tipo de lançamento, calcula e já formata (`formata_valor_br`) a divisão
(valor_empresa, valor) pra gravar direto numa `ImportacaoPlanoSaudeLinha`
— mesma regra/critério titular-dependente de `_regra_para_pessoa`/
`_calcula_valores`, só que devolvendo o resultado em vez de gravar numa
`LinhaSistema`.
"""
regra = _regra_para_pessoa(regra_por_pessoa, tipo_pessoa)
valor_empresa, valor_empregado = _calcula_valores(valor_total, regra)
return formata_valor_br(valor_empresa), formata_valor_br(valor_empregado)
# ----------------------------------------------------------------------
# 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],
regra_custeio: Optional[dict] = None,
regra_empresa_fn: Optional[Callable[[List[Tuple[LinhaSistema, float]], str], None]] = None,
tipo_lancamento: str = "",
**_ignorado,
) -> Tuple[List[LinhaSistema], List[ItemAuditoria], List[VinculoAplicado]]:
"""
`regra_empresa_fn`, quando informada, substitui o custeio normal por
pessoa (`regra_custeio`/`_aplica_regra_custeio`) — mesmo acréscimo já
documentado em `_casa_por_nome`, só que sem noção de família (o
casamento por CPF já resolve cada indivíduo direto, sem precisar
agrupar titular+dependentes primeiro): todo par (linha, valor) já
casado é acumulado e passado de uma vez só pra `regra_empresa_fn` no
fim, depois de todo o casamento — não linha a linha, já que a função
pode precisar enxergar todos os pares juntos (ex.: uma regra por
família, mesmo sem a estrutura de família explícita desta estratégia).
"""
titulares, dependentes = _indexa_por_cpf(linhas_sistema)
auditoria: List[ItemAuditoria] = []
linhas_e_valores: List[Tuple[LinhaSistema, float]] = []
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
if regra_empresa_fn is not None:
linhas_e_valores.append((linha_destino, ind.valor_total))
else:
_aplica_regra_custeio(linha_destino, ind.valor_total, _regra_para_pessoa(regra_custeio, ind.tipo))
if regra_empresa_fn is not None and linhas_e_valores:
regra_empresa_fn(linhas_e_valores, tipo_lancamento)
# Casamento por CPF nunca precisa do DE/PARA de nomes divergentes (ver
# _casa_por_nome) — o CPF já é exato por natureza.
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],
regra_custeio: Optional[dict] = None,
nomes_titular_por_numero: Dict[str, str] = None,
regra_empresa_fn: Optional[Callable[[List[Tuple[LinhaSistema, float]]], None]] = None,
vinculos_por_nome: Optional[Dict[str, VinculoNome]] = None,
tipo_lancamento: str = "",
**_ignorado,
) -> Tuple[List[LinhaSistema], List[ItemAuditoria], List[VinculoAplicado]]:
"""
`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.
`regra_empresa_fn`, quando informada (pra qualquer tipo de lançamento
coberto pela regra — ver portal_api.planos_saude.regras_empresa),
substitui o custeio normal por pessoa (`regra_custeio`/
`_aplica_regra_custeio`): em vez de aplicar a regra linha a linha
assim que cada membro é resolvido, as linhas/valores da família
inteira são acumulados primeiro em `linhas_e_valores` e só depois
passados de uma vez pra `regra_empresa_fn` no fim do laço da família
— ela decide como dividir entre as linhas (normalmente um teto por
família, não por pessoa).
`vinculos_por_nome` (nome normalizado, como veio do arquivo da
operadora, -> VinculoNome) é o "DE/PARA" persistente: quando um titular
ou dependente não bate por nome exato, mas já existe um vínculo salvo
pra esse nome exato — confirmado por um humano numa importação anterior
via "Vincular pessoa" (ImportacaoPlanoSaudeAuditoriaViewSet.resolver) —
ele é aplicado automaticamente, sem ir pra auditoria. Cada aplicação
automática vira um `VinculoAplicado` (devolvido pra quem chamou montar
um ImportacaoPlanoSaudeAlteracao depois, já que as linhas ainda não
têm `id` neste ponto). Continua valendo a regra de nunca resolver por
aproximação: um vínculo só existe depois de uma confirmação humana
explícita da MESMA divergência — isso aqui só reaproveita essa decisão,
não inventa uma nova.
"""
titulares, dependentes_por_titular = _indexa_por_nome(linhas_sistema)
nomes_titular_por_numero = nomes_titular_por_numero or {}
vinculos_por_nome = vinculos_por_nome or {}
auditoria: List[ItemAuditoria] = []
vinculos_aplicados: List[VinculoAplicado] = []
indice_por_id = {id(l): i for i, l in enumerate(linhas_sistema)}
# 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:
vinculo_titular = vinculos_por_nome.get(nome_titular_norm)
if vinculo_titular is not None:
linha_titular = titulares.get(normaliza_nome(vinculo_titular.nome_func_destino))
if linha_titular is not None:
vinculos_aplicados.append(VinculoAplicado(
indice_linha=indice_por_id[id(linha_titular)],
tipo_lancamento=tipo_lancamento,
vinculo_id=vinculo_titular.id,
nome_arquivo_operadora=nome_titular,
))
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
# A partir daqui `linha_titular` está sempre resolvida — por nome
# exato ou por vínculo salvo. `dependentes_por_titular` foi indexado
# a partir da planilha, então a chave certa é o nome REAL do
# titular na planilha (`linha_titular.nome_func`), não o nome que
# veio do arquivo da operadora (que pode ser divergente).
dependentes_da_familia = dependentes_por_titular.get(normaliza_nome(linha_titular.nome_func), {})
linhas_e_valores_familia: List[Tuple[LinhaSistema, float]] = []
for m in membros:
if m.valor_total < 0:
auditoria.append(_item_valor_negativo(m))
continue
if m.tipo == "T":
if regra_empresa_fn is not None:
linhas_e_valores_familia.append((linha_titular, m.valor_total))
else:
_aplica_regra_custeio(linha_titular, m.valor_total, _regra_para_pessoa(regra_custeio, m.tipo))
continue
linha_dep = dependentes_da_familia.get(m.nome_normalizado)
if linha_dep is None:
vinculo_dep = vinculos_por_nome.get(m.nome_normalizado)
if vinculo_dep is not None and vinculo_dep.nome_dependente_destino:
linha_dep = dependentes_da_familia.get(normaliza_nome(vinculo_dep.nome_dependente_destino))
if linha_dep is not None:
vinculos_aplicados.append(VinculoAplicado(
indice_linha=indice_por_id[id(linha_dep)],
tipo_lancamento=tipo_lancamento,
vinculo_id=vinculo_dep.id,
nome_arquivo_operadora=m.nome,
))
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
if regra_empresa_fn is not None:
linhas_e_valores_familia.append((linha_dep, m.valor_total))
else:
_aplica_regra_custeio(linha_dep, m.valor_total, _regra_para_pessoa(regra_custeio, m.tipo))
if regra_empresa_fn is not None and linhas_e_valores_familia:
regra_empresa_fn(linhas_e_valores_familia, tipo_lancamento)
return linhas_sistema, auditoria, vinculos_aplicados
# ----------------------------------------------------------------------
# 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 pipeline
# ----------------------------------------------------------------------
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",
regra_custeio: Optional[dict] = None,
**kwargs,
) -> Tuple[List[LinhaSistema], List[ItemAuditoria], List[VinculoAplicado]]:
estrategia = ESTRATEGIAS.get(chave_casamento)
if estrategia is None:
raise ValueError(f"chave_casamento desconhecida: {chave_casamento!r}")
return estrategia(individuos, linhas_sistema, regra_custeio=regra_custeio, **kwargs)