328 lines
14 KiB
Python
328 lines
14 KiB
Python
"""
|
|
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.
|
|
"""
|
|
from typing import 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, 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}`):
|
|
|
|
- "empresa" -> tudo em valor_empresa.
|
|
- "empregado" -> tudo em valor (desconto do empregado) — comportamento
|
|
padrão de antes desta funcionalidade existir.
|
|
- "especifica" -> `limite_valor` é o teto de quanto a empresa cobre
|
|
(excedente vira desconto do empregado); `percentual` é a fração do
|
|
valor do mês que a empresa 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").
|
|
"""
|
|
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
|
|
|
|
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,
|
|
**_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
|
|
|
|
_aplica_regra_custeio(linha_destino, ind.valor_total, _regra_para_pessoa(regra_custeio, ind.tipo))
|
|
|
|
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,
|
|
**_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":
|
|
_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:
|
|
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
|
|
|
|
_aplica_regra_custeio(linha_dep, m.valor_total, _regra_para_pessoa(regra_custeio, m.tipo))
|
|
|
|
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 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]]:
|
|
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)
|