""" Casa cada Individuo (extraído do arquivo da operadora) com a linha correspondente na planilha padrão do sistema. Duas estratégias, escolhidas pelo `chave_casamento` da operadora: - "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: - O valor do mês entra na coluna VALOR (desconto do empregado), não em VALOREMPRESA. - 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, Tuple from core.leiaute_sistema import formata_valor_br from core.modelos import Individuo, ItemAuditoria, LinhaSistema, normaliza_nome # ---------------------------------------------------------------------- # 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], **_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 linha_destino.valor = formata_valor_br(ind.valor_total) 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], 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": linha_titular.valor = formata_valor_br(m.valor_total) 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 linha_dep.valor = formata_valor_br(m.valor_total) 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 main.py # ---------------------------------------------------------------------- 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", **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, **kwargs)