""" 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// 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 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.regras_empresa import valida_regra_empresa 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": "3758 - Amil Odonto", "parser": AmilOdontoMensalidade, }, "unimed_saude": { "label": "5060 - Unimed Saúde", "parser": UnimedSaude, }, "itamed_saude": { "label": "3755 - Itamed Saúde", "parser": ItamedSaude, }, "dental_uni_odonto_mensalidade": { "label": "1723 - Dental Uni Odonto", "parser": DentalUniOdontoMensalidade, }, "unimed_oeste_pr_saude": { "label": "4709 - 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], regra_empresa_key: Optional[str] = None, ) -> 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. `regra_empresa_key`, quando informado, substitui o custeio configurável de "mensalidade" por uma regra especial cadastrada em portal_api.planos_saude.regras_empresa (ex.: teto de custeio por família) — `custeio_por_tipo["mensalidade"]` é ignorado nesse caso (ImportacaoPlanoSaudeCreateSerializer já garante que vem vazio). Levanta RegraEmpresaIncompativelError se a regra não servir pra esta operadora/planilha (propagada pra fora, não é um erro de arquivo). """ 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) regra_empresa_fn = None if regra_empresa_key: regra_empresa_fn = valida_regra_empresa( regra_empresa_key, parser_operadora.chave_casamento, linhas_sistema_template ) 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, regra_empresa_fn=regra_empresa_fn if tipo_lancamento == "mensalidade" else None, ) resultado.linhas_por_tipo[tipo_lancamento] = linhas_atualizadas resultado.auditoria.extend(itens_auditoria) return resultado