267 lines
12 KiB
Python
267 lines
12 KiB
Python
"""
|
|
Orquestração do pipeline de importação de plano de saúde — substitui o
|
|
`main.py` (CLI) de projects/project/ por uma função chamável a partir de
|
|
uma view do Portal. Ver projects/importacao-planos-saude.skill para todo
|
|
o histórico de decisões de negócio por trás deste pipeline.
|
|
|
|
Para adicionar uma operadora nova: crie a classe em operadoras/<nome>/
|
|
implementando OperadoraParser (ver operadoras/base.py) e registre-a em
|
|
OPERADORAS abaixo. Nada mais precisa mudar.
|
|
"""
|
|
from dataclasses import dataclass, field
|
|
from typing import Dict, List, Optional, Tuple
|
|
|
|
from portal_api.planos_saude.matcher import casa_individuos_com_planilha
|
|
from portal_api.planos_saude.modelos import Individuo, ItemAuditoria, LinhaSistema, VinculoAplicado, VinculoNome
|
|
from portal_api.planos_saude.regras_empresa import valida_regra_empresa
|
|
from portal_api.planos_saude.operadoras.amil.odonto_mensalidade import AmilOdontoMensalidade
|
|
from portal_api.planos_saude.operadoras.bradesco.odonto_mensalidade import BradescoDentalOdontoMensalidade
|
|
from portal_api.planos_saude.operadoras.bradesco.saude import BradescoSaude
|
|
from portal_api.planos_saude.operadoras.dental_uni.odonto_mensalidade import DentalUniOdontoMensalidade
|
|
from portal_api.planos_saude.operadoras.humana.saude import HumanaSaude
|
|
from portal_api.planos_saude.operadoras.itamed.saude import ItamedSaude
|
|
from portal_api.planos_saude.operadoras.sulamerica.odonto_mensalidade import SulAmericaOdontoMensalidade
|
|
from portal_api.planos_saude.operadoras.sulamerica.saude import SulAmericaSaude
|
|
from portal_api.planos_saude.operadoras.unimed.saude import UnimedSaude
|
|
from portal_api.planos_saude.operadoras.unimed_cascavel.saude import UnimedCascavelSaude
|
|
from portal_api.planos_saude.operadoras.unimed_oeste_pr.saude import UnimedOestePrSaude
|
|
from portal_api.planos_saude.operadoras.unimed_vitoria.saude import UnimedVitoriaSaude
|
|
|
|
TIPOS_LANCAMENTO_VALIDOS = ("mensalidade", "coparticipacao")
|
|
TIPOS_PESSOA_VALIDOS = ("titular", "dependente")
|
|
CUSTEIOS_VALIDOS = ("empresa", "empregado", "especifica")
|
|
|
|
OPERADORAS = {
|
|
"amil_odonto_mensalidade": {
|
|
"codigo_operadora": "3758",
|
|
"nome": "Amil Odonto",
|
|
"parser": AmilOdontoMensalidade,
|
|
},
|
|
"amil_odonto_mensalidade_898": {
|
|
"codigo_operadora": "898",
|
|
"nome": "Amil Odonto",
|
|
"parser": AmilOdontoMensalidade,
|
|
},
|
|
"unimed_saude": {
|
|
"codigo_operadora": "5060",
|
|
"nome": "Unimed Saúde",
|
|
"parser": UnimedSaude,
|
|
},
|
|
"itamed_saude": {
|
|
"codigo_operadora": "3755",
|
|
"nome": "Itamed Saúde",
|
|
"parser": ItamedSaude,
|
|
},
|
|
"dental_uni_odonto_mensalidade": {
|
|
"codigo_operadora": "4723",
|
|
"nome": "Dental Uni Odonto",
|
|
"parser": DentalUniOdontoMensalidade,
|
|
},
|
|
"humana_saude": {
|
|
"codigo_operadora": "5064",
|
|
"nome": "Humana Saúde",
|
|
"parser": HumanaSaude,
|
|
},
|
|
"unimed_oeste_pr_saude": {
|
|
"codigo_operadora": "4709",
|
|
"nome": "Unimed Oeste do Paraná",
|
|
"parser": UnimedOestePrSaude,
|
|
},
|
|
"bradesco_saude": {
|
|
"codigo_operadora": "1386",
|
|
"nome": "Bradesco Saúde",
|
|
"parser": BradescoSaude,
|
|
},
|
|
"bradesco_dental_odonto_mensalidade": {
|
|
"codigo_operadora": "3759",
|
|
"nome": "Bradesaude Odonto",
|
|
"parser": BradescoDentalOdontoMensalidade,
|
|
},
|
|
"unimed_vitoria_saude": {
|
|
"codigo_operadora": "4750",
|
|
"nome": "Unimed Vitória",
|
|
"parser": UnimedVitoriaSaude,
|
|
},
|
|
"sulamerica_odonto_mensalidade": {
|
|
"codigo_operadora": "4726",
|
|
"nome": "SulAmérica Odonto",
|
|
"parser": SulAmericaOdontoMensalidade,
|
|
},
|
|
"sulamerica_saude": {
|
|
"codigo_operadora": "5775",
|
|
"nome": "SulAmérica",
|
|
"parser": SulAmericaSaude,
|
|
},
|
|
"unimed_cascavel_saude": {
|
|
"codigo_operadora": "158",
|
|
"nome": "Unimed Cascavel",
|
|
"parser": UnimedCascavelSaude,
|
|
},
|
|
}
|
|
|
|
|
|
def label_operadora(operadora_key: str) -> str:
|
|
""""<código de cadastro da operadora no Questor> - <Nome>" (ex.:
|
|
"5060 - Unimed Saúde") — esse código é da OPERADORA, não do
|
|
`codigo_empresa` do cliente que contratou o plano. Fonte única pra
|
|
montar o label a partir dos campos separados de OPERADORAS, usada
|
|
tanto por `lista_operadoras()` quanto por quem precisava do `label`
|
|
bruto antes desta função existir."""
|
|
dados = OPERADORAS[operadora_key]
|
|
return f"{dados['codigo_operadora']} - {dados['nome']}"
|
|
|
|
|
|
def lista_operadoras() -> List[Dict[str, str]]:
|
|
chaves_ordenadas = sorted(OPERADORAS, key=lambda chave: int(OPERADORAS[chave]["codigo_operadora"]))
|
|
return [{"key": chave, "label": label_operadora(chave)} for chave in chaves_ordenadas]
|
|
|
|
|
|
@dataclass
|
|
class ResultadoProcessamento:
|
|
nome_operadora: str
|
|
linhas_por_tipo: Dict[str, List[LinhaSistema]] = field(default_factory=dict)
|
|
auditoria: List[ItemAuditoria] = field(default_factory=list)
|
|
# Nomes divergentes resolvidos automaticamente via um VinculoNomeOperadora
|
|
# já salvo (ver "Vínculos de nome salvos (DE/PARA)" no CLAUDE.md) — views.py
|
|
# usa isso pra gerar um ImportacaoPlanoSaudeAlteracao por vínculo aplicado,
|
|
# depois que `linhas_por_tipo` já estiver persistido (só aí as linhas têm id).
|
|
vinculos_aplicados: List[VinculoAplicado] = field(default_factory=list)
|
|
|
|
|
|
def _agrega_individuos_entre_arquivos(individuos: List[Individuo]) -> List[Individuo]:
|
|
"""Some os `Individuo` da MESMA pessoa e do MESMO `tipo_lancamento` que
|
|
vieram de arquivos diferentes (ex.: duas coparticipações separadas por
|
|
período) — sem isso, `casa_individuos_com_planilha`/`_aplica_regra_
|
|
custeio` (matcher.py) *grava* o valor por linha em vez de acumular, então
|
|
o segundo arquivo simplesmente sobrescreveria o valor do primeiro. Cada
|
|
`OperadoraParser.extrai()` já faz esse mesmo tipo de agregação dentro de
|
|
UM arquivo (ver `_agrega_por_individuo_e_tipo` nos parsers) — esta função
|
|
é o equivalente entre arquivos, chamada uma vez por importação depois de
|
|
somar os indivíduos de todos eles."""
|
|
agregados: Dict[Tuple[str, str], Individuo] = {}
|
|
ordem = []
|
|
for ind in individuos:
|
|
chave = (ind.numero_beneficiario, ind.tipo_lancamento)
|
|
if chave not in agregados:
|
|
agregados[chave] = Individuo(
|
|
numero_beneficiario=ind.numero_beneficiario,
|
|
nome=ind.nome,
|
|
cpf=ind.cpf,
|
|
tipo=ind.tipo,
|
|
tipo_lancamento=ind.tipo_lancamento,
|
|
numero_titular=ind.numero_titular,
|
|
)
|
|
ordem.append(chave)
|
|
agregados[chave].valor_total += ind.valor_total
|
|
agregados[chave].rubricas.extend(ind.rubricas)
|
|
return [agregados[chave] for chave in ordem]
|
|
|
|
|
|
def processa_importacao(
|
|
operadora_key: str,
|
|
caminhos_arquivo_operadora: List[str],
|
|
linhas_sistema_template: List[LinhaSistema],
|
|
tipos_selecionados: List[str],
|
|
custeio_por_tipo: Dict[str, dict],
|
|
regra_empresa_key: Optional[str] = None,
|
|
vinculos_por_nome: Optional[Dict[str, VinculoNome]] = None,
|
|
) -> ResultadoProcessamento:
|
|
"""
|
|
Extrai um ou mais arquivos da operadora (a maioria manda só um, mas
|
|
algumas — ex.: Unimed Saúde em PDF — mandam mensalidade e coparticipação
|
|
em arquivos separados; `OperadoraParser.extrai()` é chamado uma vez por
|
|
arquivo e os indivíduos/itens de auditoria resultantes são somados antes
|
|
de seguir), filtra pelos tipos de lançamento que o usuário selecionou na
|
|
tela (ignora qualquer outro tipo presente no arquivo) e casa cada tipo
|
|
com uma cópia própria da planilha padrão —
|
|
cada tipo divide o valor do mês entre `valor_empresa`/`valor` conforme a
|
|
regra de custeio escolhida para ele, uma para titular e outra para
|
|
dependente (ver `matcher._regra_para_pessoa`/`_calcula_valores` pro
|
|
formato de `custeio_por_tipo[tipo]`: `{"titular": {"modo": "empresa"|
|
|
"empregado"|"especifica", "limite_valor": float|None, "percentual":
|
|
float|None}, "dependente": {...}}`). Réplica do loop de main.py, com o
|
|
acréscimo do filtro por tipos_selecionados e do custeio configurável.
|
|
|
|
`linhas_sistema_template` já vem pronta (lida do arquivo via
|
|
`le_planilha_padrao` ou buscada no Questor via
|
|
`questor_planilha.busca_linhas_questor` — decisão de qual das duas
|
|
fontes usar é de quem chama, em views.py).
|
|
|
|
`regra_empresa_key`, quando informado, substitui o custeio configurável
|
|
dos tipos de lançamento que a regra cobre (`REGRAS_EMPRESA[chave]
|
|
["tipos_lancamento"]` — ver portal_api.planos_saude.regras_empresa) por
|
|
uma regra especial cadastrada em código (ex.: teto de custeio por
|
|
família, ou um critério fixo por tipo de beneficiário) —
|
|
`custeio_por_tipo[tipo]` desses tipos é ignorado nesse caso
|
|
(ImportacaoPlanoSaudeCreateSerializer já garante que vem vazio). Um
|
|
tipo selecionado que a regra NÃO cobre continua seguindo
|
|
`custeio_por_tipo[tipo]` normalmente. Levanta RegraEmpresaIncompativelError
|
|
se a regra não servir pra esta operadora/planilha (propagada pra fora,
|
|
não é um erro de arquivo).
|
|
|
|
`vinculos_por_nome` (opcional, só usado pela estratégia "nome" — ver
|
|
matcher._casa_por_nome) é o "DE/PARA" de nomes divergentes já
|
|
confirmados numa importação anterior; buscado no banco por views.py
|
|
(este pacote não tem ORM) e passado pra todo tipo_lancamento igual.
|
|
"""
|
|
operadora_info = OPERADORAS[operadora_key]
|
|
parser_operadora = operadora_info["parser"]()
|
|
|
|
individuos: List[Individuo] = []
|
|
auditoria_extracao: List[ItemAuditoria] = []
|
|
for caminho in caminhos_arquivo_operadora:
|
|
individuos_arquivo, auditoria_arquivo = parser_operadora.extrai(caminho)
|
|
individuos.extend(individuos_arquivo)
|
|
auditoria_extracao.extend(auditoria_arquivo)
|
|
# Chamado só depois que extrai() já rodou pra todos os arquivos —
|
|
# operadoras que precisam ver o conjunto completo antes de decidir algo
|
|
# sobrescrevem isto (ver OperadoraParser.finaliza()); a maioria não usa.
|
|
individuos_finais, auditoria_final = parser_operadora.finaliza()
|
|
individuos.extend(individuos_finais)
|
|
auditoria_extracao.extend(auditoria_final)
|
|
individuos = _agrega_individuos_entre_arquivos(individuos)
|
|
|
|
regra_empresa_fn = None
|
|
regra_empresa_tipos: Tuple[str, ...] = ()
|
|
if regra_empresa_key:
|
|
regra_empresa_fn, regra_empresa_tipos = valida_regra_empresa(
|
|
regra_empresa_key, parser_operadora.chave_casamento_para_tipo, linhas_sistema_template, tipos_selecionados
|
|
)
|
|
|
|
individuos_por_tipo: Dict[str, list] = {}
|
|
for ind in individuos:
|
|
if ind.tipo_lancamento in tipos_selecionados:
|
|
individuos_por_tipo.setdefault(ind.tipo_lancamento, []).append(ind)
|
|
|
|
# mapa global (todos os tipos juntos, mesmo os não selecionados) de
|
|
# numero_titular -> nome do titular — ver nota em main.py/matcher.py
|
|
# original: precisa ser global pra não perder o titular de uma
|
|
# família quando ele não teve nenhuma cobrança do tipo selecionado.
|
|
nomes_titular_por_numero = {
|
|
ind.numero_beneficiario: ind.nome for ind in individuos if ind.tipo == "T"
|
|
}
|
|
|
|
resultado = ResultadoProcessamento(nome_operadora=parser_operadora.nome_operadora)
|
|
resultado.auditoria.extend(auditoria_extracao)
|
|
|
|
for tipo_lancamento in tipos_selecionados:
|
|
individuos_do_tipo = individuos_por_tipo.get(tipo_lancamento, [])
|
|
linhas_copia = [
|
|
LinhaSistema(**vars(linha)) for linha in linhas_sistema_template
|
|
]
|
|
linhas_atualizadas, itens_auditoria, vinculos_aplicados = casa_individuos_com_planilha(
|
|
individuos_do_tipo,
|
|
linhas_copia,
|
|
parser_operadora.chave_casamento_para_tipo(tipo_lancamento),
|
|
regra_custeio=custeio_por_tipo[tipo_lancamento],
|
|
nomes_titular_por_numero=nomes_titular_por_numero,
|
|
regra_empresa_fn=regra_empresa_fn if tipo_lancamento in regra_empresa_tipos else None,
|
|
vinculos_por_nome=vinculos_por_nome,
|
|
tipo_lancamento=tipo_lancamento,
|
|
)
|
|
resultado.linhas_por_tipo[tipo_lancamento] = linhas_atualizadas
|
|
resultado.vinculos_aplicados.extend(vinculos_aplicados)
|
|
resultado.auditoria.extend(itens_auditoria)
|
|
|
|
return resultado
|