portal_publico/projects/project/main.py

95 lines
4.0 KiB
Python

#!/usr/bin/env python3
"""
CLI do pipeline de importação de planos de saúde/odontológicos.
Uso:
python3 main.py --operadora amil_odonto_mensalidade \
--arquivo-operadora /caminho/relatorio.pdf \
--planilha-padrao /caminho/leiaute_vazio.csv \
--saida-dir /caminho/saida/ \
--auditoria /caminho/auditoria.csv
Uma única execução gera UM arquivo de saída para cada tipo de
lançamento (mensalidade / coparticipação) que aparecer nos dados da
operadora — não precisa rodar duas vezes. Os arquivos saem em
--saida-dir nomeados automaticamente: mensalidade.csv, coparticipacao.csv.
Para adicionar uma nova operadora: crie a classe em operadoras/<nome>/ e
registre-a no dicionário OPERADORAS abaixo. Nada mais no pipeline precisa
mudar.
"""
import argparse
import copy
import os
from core.auditoria import escreve_auditoria
from core.leiaute_sistema import escreve_planilha_padrao, le_planilha_padrao
from core.matcher import casa_individuos_com_planilha
from operadoras.amil.odonto_mensalidade import AmilOdontoMensalidade
from operadoras.unimed.saude import UnimedSaude
OPERADORAS = {
"amil_odonto_mensalidade": AmilOdontoMensalidade,
"unimed_saude": UnimedSaude,
# próximas operadoras entram aqui, ex:
# "bradesco_saude": BradescoSaude,
}
def main():
parser = argparse.ArgumentParser(description=__doc__)
parser.add_argument("--operadora", required=True, choices=OPERADORAS.keys())
parser.add_argument("--arquivo-operadora", required=True)
parser.add_argument("--planilha-padrao", required=True)
parser.add_argument("--saida-dir", required=True)
parser.add_argument("--auditoria", required=True)
args = parser.parse_args()
os.makedirs(args.saida_dir, exist_ok=True)
parser_operadora = OPERADORAS[args.operadora]()
individuos, auditoria_extracao = parser_operadora.extrai(args.arquivo_operadora)
linhas_sistema_template = le_planilha_padrao(args.planilha_padrao)
# separa os indivíduos por tipo de lançamento (mensalidade /
# coparticipação) — cada tipo gera seu próprio arquivo de saída,
# partindo de uma cópia independente da planilha padrão.
individuos_por_tipo = {}
for ind in individuos:
individuos_por_tipo.setdefault(ind.tipo_lancamento, []).append(ind)
# mapa global (todos os tipos juntos) de numero_titular -> nome do
# titular. Precisa ser global porque um titular pode não ter nenhuma
# linha de coparticipação num mês em que só o dependente foi ao
# médico — sem isso, a família "perderia" o titular ao filtrar por
# tipo antes de casar com a planilha padrão.
nomes_titular_por_numero = {
ind.numero_beneficiario: ind.nome for ind in individuos if ind.tipo == "T"
}
auditoria_total = list(auditoria_extracao)
resumo = []
for tipo_lancamento, individuos_do_tipo in individuos_por_tipo.items():
linhas_copia = copy.deepcopy(linhas_sistema_template)
linhas_atualizadas, itens_auditoria = casa_individuos_com_planilha(
individuos_do_tipo, linhas_copia, parser_operadora.chave_casamento,
nomes_titular_por_numero=nomes_titular_por_numero,
)
caminho_saida = os.path.join(args.saida_dir, f"{tipo_lancamento}.csv")
escreve_planilha_padrao(caminho_saida, linhas_atualizadas)
auditoria_total.extend(itens_auditoria)
resumo.append((tipo_lancamento, len(individuos_do_tipo), len(itens_auditoria), caminho_saida))
escreve_auditoria(args.auditoria, auditoria_total)
print(f"[{parser_operadora.nome_operadora}] (casamento por: {parser_operadora.chave_casamento})")
for tipo_lancamento, total, n_auditoria, caminho in resumo:
print(f" {tipo_lancamento}: {total} beneficiários extraídos, "
f"{total - n_auditoria} lançados, {n_auditoria} em auditoria -> {caminho}")
print(f" Auditoria (todos os tipos): {len(auditoria_total)} itens -> {args.auditoria}")
if __name__ == "__main__":
main()