portal_publico/portal_api/planos_saude/pipeline.py

121 lines
5.0 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
from portal_api.planos_saude.leiaute_sistema import le_planilha_padrao
from portal_api.planos_saude.matcher import casa_individuos_com_planilha
from portal_api.planos_saude.modelos import ItemAuditoria, LinhaSistema
from portal_api.planos_saude.operadoras.amil.odonto_mensalidade import AmilOdontoMensalidade
from portal_api.planos_saude.operadoras.dental_uni.odonto_mensalidade import DentalUniOdontoMensalidade
from portal_api.planos_saude.operadoras.itamed.saude import ItamedSaude
from portal_api.planos_saude.operadoras.unimed.saude import UnimedSaude
from portal_api.planos_saude.operadoras.unimed_oeste_pr.saude import UnimedOestePrSaude
TIPOS_LANCAMENTO_VALIDOS = ("mensalidade", "coparticipacao")
TIPOS_PESSOA_VALIDOS = ("titular", "dependente")
CUSTEIOS_VALIDOS = ("empresa", "empregado", "especifica")
OPERADORAS = {
"amil_odonto_mensalidade": {
"label": "Amil Odonto",
"parser": AmilOdontoMensalidade,
},
"unimed_saude": {
"label": "Unimed Saúde",
"parser": UnimedSaude,
},
"itamed_saude": {
"label": "Itamed Saúde",
"parser": ItamedSaude,
},
"dental_uni_odonto_mensalidade": {
"label": "Dental Uni Odonto",
"parser": DentalUniOdontoMensalidade,
},
"unimed_oeste_pr_saude": {
"label": "Unimed Oeste do Paraná",
"parser": UnimedOestePrSaude,
},
# próximas operadoras entram aqui, ex:
# "bradesco_saude": {"label": "Bradesco Saúde", "parser": BradescoSaude},
}
def lista_operadoras() -> List[Dict[str, str]]:
return [{"key": chave, "label": dados["label"]} for chave, dados in OPERADORAS.items()]
@dataclass
class ResultadoProcessamento:
nome_operadora: str
linhas_por_tipo: Dict[str, List[LinhaSistema]] = field(default_factory=dict)
auditoria: List[ItemAuditoria] = field(default_factory=list)
def processa_importacao(
operadora_key: str,
caminho_arquivo_operadora: str,
caminho_planilha_padrao: str,
tipos_selecionados: List[str],
custeio_por_tipo: Dict[str, dict],
) -> ResultadoProcessamento:
"""
Extrai o arquivo da operadora, 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.
"""
operadora_info = OPERADORAS[operadora_key]
parser_operadora = operadora_info["parser"]()
individuos, auditoria_extracao = parser_operadora.extrai(caminho_arquivo_operadora)
linhas_sistema_template = le_planilha_padrao(caminho_planilha_padrao)
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 = casa_individuos_com_planilha(
individuos_do_tipo,
linhas_copia,
parser_operadora.chave_casamento,
regra_custeio=custeio_por_tipo[tipo_lancamento],
nomes_titular_por_numero=nomes_titular_por_numero,
)
resultado.linhas_por_tipo[tipo_lancamento] = linhas_atualizadas
resultado.auditoria.extend(itens_auditoria)
return resultado