""" Casa cada Individuo (extraído do arquivo da operadora) com a linha correspondente na planilha padrão do sistema. Porta de projects/project/core/matcher.py — mesmas duas estratégias ("cpf"/"nome", ver docstring original de cada uma abaixo), com UM acréscimo: `regra_custeio` decide como o valor do mês se divide entre `valor_empresa` (custeado pela empresa) e `valor` (desconto do empregado) — antes disso o valor cheio sempre ia pra `valor`, fixo. Essa é a única diferença de comportamento em relação ao pipeline original; quem decide a regra é o usuário na tela de nova importação, por tipo de lançamento **e** por tipo de beneficiário (titular ou dependente/agregado) — ex.: mensalidade do titular custeada pela empresa, mensalidade do dependente descontada do empregado (ver `_regra_para_pessoa`/`_calcula_valores` abaixo pro formato). - "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: - 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. Segundo acréscimo: `regra_empresa_fn` (opcional, só implementado na estratégia "nome" — ver `_casa_por_nome`) substitui `regra_custeio` por uma função de custeio calculada por FAMÍLIA inteira em vez de por pessoa, pra regras especiais que não cabem no desenho normal (ver portal_api.planos_saude.regras_empresa) — resolvida e validada em pipeline.processa_importacao antes de chegar aqui. Terceiro acréscimo: `vinculos_por_nome` (opcional, só na estratégia "nome") — um "DE/PARA" persistente (`portal_api.models.VinculoNomeOperadora`, representado aqui sem ORM como `modelos.VinculoNome`) que reaproveita uma confirmação humana anterior ("Vincular pessoa", ver ImportacaoPlanoSaudeAuditoriaViewSet.resolver em views.py) pra resolver automaticamente a MESMA divergência de nome numa importação futura, sem precisar vincular de novo todo mês. Continua não sendo aproximação — só existe depois que alguém confirmou explicitamente aquele nome específico uma vez; sem vínculo salvo, o comportamento é idêntico a antes (auditoria). Cada aplicação automática é registrada em `VinculoAplicado` e devolvida pra quem chamou (pipeline.py/views.py) montar um `ImportacaoPlanoSaudeAlteracao` ("vínculo automático") depois que as linhas forem persistidas — ver "Vínculos de nome salvos (DE/PARA)" no CLAUDE.md. """ from typing import Callable, Dict, List, Optional, Tuple from portal_api.planos_saude.leiaute_sistema import formata_valor_br from portal_api.planos_saude.modelos import ( Individuo, ItemAuditoria, LinhaSistema, VinculoAplicado, VinculoNome, normaliza_nome, ) REGRA_CUSTEIO_PADRAO = {"modo": "empregado"} def _calcula_valores(valor_total: float, regra: Optional[dict]) -> Tuple[float, float]: """ Divide `valor_total` entre (valor_empresa, valor_empregado) conforme `regra` (`{"modo": "empresa"|"empregado"|"especifica", "limite_valor": float|None, "percentual": float|None}`): - "empresa" -> tudo em valor_empresa. - "empregado" -> tudo em valor (desconto do empregado) — comportamento padrão de antes desta funcionalidade existir. - "especifica" -> `limite_valor` é o teto de quanto a empresa cobre (excedente vira desconto do empregado); `percentual` é a fração do valor do mês que a empresa cobre (o resto vira desconto). Pode vir só um dos dois ou os dois — quando os dois vêm juntos, prevalece o mais restritivo (o menor valor entre os dois critérios), decisão explícita do usuário ("pode ser aplicado apenas uma destas regras ou as duas"). """ regra = regra or REGRA_CUSTEIO_PADRAO modo = regra.get("modo", "empregado") if modo == "empresa": return valor_total, 0.0 if modo != "especifica": return 0.0, valor_total candidatos = [valor_total] percentual = regra.get("percentual") if percentual is not None: candidatos.append(valor_total * percentual / 100) limite_valor = regra.get("limite_valor") if limite_valor is not None: candidatos.append(limite_valor) # Arredonda valor_empresa primeiro e deriva valor_empregado como o # complemento exato (valor_total - valor_empresa) — arredondar os dois # separadamente (ex: 50% de 51,69 = 25,845 dos dois lados) podia perder # ou sobrar 1 centavo na soma final. valor_empresa = round(max(0.0, min(candidatos)), 2) valor_empregado = round(valor_total - valor_empresa, 2) return valor_empresa, valor_empregado def _regra_para_pessoa(regra_por_pessoa: Optional[dict], tipo_pessoa: str) -> Optional[dict]: """ `regra_por_pessoa` é `{"titular": {...regra...}, "dependente": {...regra...}}` — cada tipo de lançamento pode ter uma regra de custeio diferente para titular e para dependente (ex.: mensalidade do titular custeada pela empresa, mensalidade do dependente descontada do empregado). `tipo_pessoa` é o `Individuo.tipo`/`LinhaSistema` ('T' = Titular, 'D'/'A' = Dependente/Agregado — 'D' e 'A' caem no mesmo balde 'dependente', mesmo critério já usado no leiaute do sistema). """ chave = "titular" if tipo_pessoa == "T" else "dependente" return (regra_por_pessoa or {}).get(chave) def _aplica_regra_custeio(linha: LinhaSistema, valor_total: float, regra: Optional[dict]) -> None: valor_empresa, valor_empregado = _calcula_valores(valor_total, regra) linha.valor_empresa = formata_valor_br(valor_empresa) linha.valor = formata_valor_br(valor_empregado) def valores_formatados_para_pessoa( valor_total: float, regra_por_pessoa: Optional[dict], tipo_pessoa: str ) -> Tuple[str, str]: """ Único ponto de entrada público deste módulo pra fora do pipeline normal de casamento — usado pela resolução manual de um item de auditoria (ver `ImportacaoPlanoSaudeAuditoriaViewSet.resolver` em views.py): dado um valor já extraído do arquivo da operadora e a regra de custeio salva pro tipo de lançamento, calcula e já formata (`formata_valor_br`) a divisão (valor_empresa, valor) pra gravar direto numa `ImportacaoPlanoSaudeLinha` — mesma regra/critério titular-dependente de `_regra_para_pessoa`/ `_calcula_valores`, só que devolvendo o resultado em vez de gravar numa `LinhaSistema`. """ regra = _regra_para_pessoa(regra_por_pessoa, tipo_pessoa) valor_empresa, valor_empregado = _calcula_valores(valor_total, regra) return formata_valor_br(valor_empresa), formata_valor_br(valor_empregado) # ---------------------------------------------------------------------- # 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], regra_custeio: Optional[dict] = None, **_ignorado, ) -> Tuple[List[LinhaSistema], List[ItemAuditoria], List[VinculoAplicado]]: 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 _aplica_regra_custeio(linha_destino, ind.valor_total, _regra_para_pessoa(regra_custeio, ind.tipo)) # Casamento por CPF nunca precisa do DE/PARA de nomes divergentes (ver # _casa_por_nome) — o CPF já é exato por natureza. 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], regra_custeio: Optional[dict] = None, nomes_titular_por_numero: Dict[str, str] = None, regra_empresa_fn: Optional[Callable[[List[Tuple[LinhaSistema, float]]], None]] = None, vinculos_por_nome: Optional[Dict[str, VinculoNome]] = None, tipo_lancamento: str = "", **_ignorado, ) -> Tuple[List[LinhaSistema], List[ItemAuditoria], List[VinculoAplicado]]: """ `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. `regra_empresa_fn`, quando informada (só usada para "mensalidade" — ver portal_api.planos_saude.regras_empresa), substitui o custeio normal por pessoa (`regra_custeio`/`_aplica_regra_custeio`): em vez de aplicar a regra linha a linha assim que cada membro é resolvido, as linhas/valores da família inteira são acumulados primeiro em `linhas_e_valores` e só then passados de uma vez pra `regra_empresa_fn` no fim do laço da família — ela decide como dividir entre as linhas (normalmente um teto por família, não por pessoa). `vinculos_por_nome` (nome normalizado, como veio do arquivo da operadora, -> VinculoNome) é o "DE/PARA" persistente: quando um titular ou dependente não bate por nome exato, mas já existe um vínculo salvo pra esse nome exato — confirmado por um humano numa importação anterior via "Vincular pessoa" (ImportacaoPlanoSaudeAuditoriaViewSet.resolver) — ele é aplicado automaticamente, sem ir pra auditoria. Cada aplicação automática vira um `VinculoAplicado` (devolvido pra quem chamou montar um ImportacaoPlanoSaudeAlteracao depois, já que as linhas ainda não têm `id` neste ponto). Continua valendo a regra de nunca resolver por aproximação: um vínculo só existe depois de uma confirmação humana explícita da MESMA divergência — isso aqui só reaproveita essa decisão, não inventa uma nova. """ titulares, dependentes_por_titular = _indexa_por_nome(linhas_sistema) nomes_titular_por_numero = nomes_titular_por_numero or {} vinculos_por_nome = vinculos_por_nome or {} auditoria: List[ItemAuditoria] = [] vinculos_aplicados: List[VinculoAplicado] = [] indice_por_id = {id(l): i for i, l in enumerate(linhas_sistema)} # 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: vinculo_titular = vinculos_por_nome.get(nome_titular_norm) if vinculo_titular is not None: linha_titular = titulares.get(normaliza_nome(vinculo_titular.nome_func_destino)) if linha_titular is not None: vinculos_aplicados.append(VinculoAplicado( indice_linha=indice_por_id[id(linha_titular)], tipo_lancamento=tipo_lancamento, vinculo_id=vinculo_titular.id, nome_arquivo_operadora=nome_titular, )) 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 # A partir daqui `linha_titular` está sempre resolvida — por nome # exato ou por vínculo salvo. `dependentes_por_titular` foi indexado # a partir da planilha, então a chave certa é o nome REAL do # titular na planilha (`linha_titular.nome_func`), não o nome que # veio do arquivo da operadora (que pode ser divergente). dependentes_da_familia = dependentes_por_titular.get(normaliza_nome(linha_titular.nome_func), {}) linhas_e_valores_familia: List[Tuple[LinhaSistema, float]] = [] for m in membros: if m.valor_total < 0: auditoria.append(_item_valor_negativo(m)) continue if m.tipo == "T": if regra_empresa_fn is not None: linhas_e_valores_familia.append((linha_titular, m.valor_total)) else: _aplica_regra_custeio(linha_titular, m.valor_total, _regra_para_pessoa(regra_custeio, m.tipo)) continue linha_dep = dependentes_da_familia.get(m.nome_normalizado) if linha_dep is None: vinculo_dep = vinculos_por_nome.get(m.nome_normalizado) if vinculo_dep is not None and vinculo_dep.nome_dependente_destino: linha_dep = dependentes_da_familia.get(normaliza_nome(vinculo_dep.nome_dependente_destino)) if linha_dep is not None: vinculos_aplicados.append(VinculoAplicado( indice_linha=indice_por_id[id(linha_dep)], tipo_lancamento=tipo_lancamento, vinculo_id=vinculo_dep.id, nome_arquivo_operadora=m.nome, )) 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 if regra_empresa_fn is not None: linhas_e_valores_familia.append((linha_dep, m.valor_total)) else: _aplica_regra_custeio(linha_dep, m.valor_total, _regra_para_pessoa(regra_custeio, m.tipo)) if regra_empresa_fn is not None and linhas_e_valores_familia: regra_empresa_fn(linhas_e_valores_familia) return linhas_sistema, auditoria, vinculos_aplicados # ---------------------------------------------------------------------- # 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 pipeline # ---------------------------------------------------------------------- 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", regra_custeio: Optional[dict] = None, **kwargs, ) -> Tuple[List[LinhaSistema], List[ItemAuditoria], List[VinculoAplicado]]: estrategia = ESTRATEGIAS.get(chave_casamento) if estrategia is None: raise ValueError(f"chave_casamento desconhecida: {chave_casamento!r}") return estrategia(individuos, linhas_sistema, regra_custeio=regra_custeio, **kwargs)