portal_publico/portal_api/indicadores/pipeline.py

"""Orquestração do Indicador de Desempenho: lê as duas planilhas, deriva os
papéis/tipos por empresa e pré-calcula os 3 critérios automáticos por
colaborador. Só calcula em memória (dataclasses), sem tocar no ORM — mesmo
padrão de `portal_api.planos_saude.pipeline`; quem persiste é a view."""

from __future__ import annotations

from dataclasses import dataclass, field
from decimal import Decimal

from .entregas import calcula_percentual_entrega
from .leiaute import (
    SERVICO_CONCILIACAO_FINANCEIRA,
    SERVICO_LIBERACAO_FISCAL,
    SERVICOS_BALANCETE,
    le_honorarios,
    le_servicos_tareffa,
)
from .tipos import PapelPorEmpresa, deriva_tipos_por_empresa

CALCULO_BALANCETE = "balancete"
CALCULO_LIBERACAO_FISCAL = "liberacao_fiscal"
CALCULO_CONCILIACAO = "conciliacao"

SERVICOS_POR_CALCULO_AUTOMATICO = {
    CALCULO_BALANCETE: SERVICOS_BALANCETE,
    CALCULO_LIBERACAO_FISCAL: {SERVICO_LIBERACAO_FISCAL},
    CALCULO_CONCILIACAO: {SERVICO_CONCILIACAO_FINANCEIRA},
}


@dataclass
class CriterioAutomaticoConfig:
    criterio_id: int
    tipo_calculo: str  # "balancete" | "liberacao_fiscal" | "conciliacao"
    limiar_percentual: Decimal


@dataclass
class RespostaAutomatica:
    criterio_id: int
    valor: str  # "SIM" | "NAO" | "NAO_FAZ"
    percentual_calculado: Decimal | None


@dataclass
class EmpresaCalculada:
    codigo_empresa: str
    nome_empresa: str
    honorario: Decimal
    honorario_nao_encontrado: bool
    tipo: str


@dataclass
class ColaboradorCalculado:
    nome: str
    gerente: str
    departamento_id: int | None = None
    empresas: list[EmpresaCalculada] = field(default_factory=list)
    respostas_automaticas: list[RespostaAutomatica] = field(default_factory=list)


@dataclass
class ResultadoApuracao:
    colaboradores: list[ColaboradorCalculado]
    avisos: list[str]


def processa_apuracao(
    caminho_tareffa: str,
    caminho_honorarios: str,
    gerente_departamento: dict[str, int],
    criterios_automaticos_por_departamento: dict[int, list[CriterioAutomaticoConfig]],
) -> ResultadoApuracao:
    """`gerente_departamento` ({nome do gerente: id do IndicadorDepartamento})
    e `criterios_automaticos_por_departamento` ({id do departamento: lista de
    critérios automáticos daquele departamento}) vêm prontos de fora (ORM
    fica só na view) — cada colaborador é resolvido pro departamento do seu
    `gerente` antes de decidir quais critérios automáticos calcular pra ele,
    já que cada departamento pode ter critérios/pesos/limiares diferentes.
    Colaborador cujo gerente não está em `gerente_departamento` fica com
    `departamento_id=None` e sem nenhum critério automático calculado (vira
    aviso, ver `avisos` abaixo)."""
    linhas = le_servicos_tareffa(caminho_tareffa)
    honorarios = le_honorarios(caminho_honorarios)

    derivacao = deriva_tipos_por_empresa(linhas)
    avisos = list(derivacao.avisos)

    gerente_por_responsavel: dict[str, str] = {}
    for linha in linhas:
        if linha.responsavel and linha.responsavel not in gerente_por_responsavel:
            gerente_por_responsavel[linha.responsavel] = linha.gerente

    papeis_por_responsavel: dict[str, list[PapelPorEmpresa]] = {}
    for papel in derivacao.papeis:
        papeis_por_responsavel.setdefault(papel.responsavel, []).append(papel)

    colaboradores: list[ColaboradorCalculado] = []
    for responsavel, papeis in sorted(papeis_por_responsavel.items()):
        gerente = gerente_por_responsavel.get(responsavel, "")
        departamento_id = gerente_departamento.get(gerente)
        if departamento_id is None:
            avisos.append(
                f'Gerente "{gerente or "(em branco)"}" (colaborador "{responsavel}") não está associado a '
                "nenhum departamento — cadastre essa relação em Configurações → Departamentos; até lá, este "
                "colaborador fica sem critérios nem percentuais calculados."
            )
        colaborador = ColaboradorCalculado(nome=responsavel, gerente=gerente, departamento_id=departamento_id)

        for papel in papeis:
            honorario = honorarios.get(papel.codigo_empresa)
            colaborador.empresas.append(
                EmpresaCalculada(
                    codigo_empresa=papel.codigo_empresa,
                    nome_empresa=papel.nome_empresa,
                    honorario=honorario if honorario is not None else Decimal("0"),
                    honorario_nao_encontrado=honorario is None,
                    tipo=papel.tipo,
                )
            )

        criterios_automaticos = criterios_automaticos_por_departamento.get(departamento_id, [])
        for criterio in criterios_automaticos:
            servicos = SERVICOS_POR_CALCULO_AUTOMATICO.get(criterio.tipo_calculo)
            if not servicos:
                continue
            resultado_entrega = calcula_percentual_entrega(linhas, responsavel, servicos)
            if resultado_entrega.percentual is None:
                # Responsável nunca prestou esse tipo de serviço na competência —
                # o critério não se aplica a ele, não é uma falha (isso é o que
                # permite deixar `papel_aplicavel` em branco nos 3 critérios
                # automáticos: quem não faz aquele serviço fica de fora do
                # cálculo por conta própria, sem precisar restringir por papel).
                valor = "NAO_SE_APLICA"
            elif resultado_entrega.percentual >= criterio.limiar_percentual:
                valor = "SIM"
            else:
                valor = "NAO"
            colaborador.respostas_automaticas.append(
                RespostaAutomatica(
                    criterio_id=criterio.criterio_id,
                    valor=valor,
                    percentual_calculado=resultado_entrega.percentual,
                )
            )

        colaboradores.append(colaborador)

    return ResultadoApuracao(colaboradores=colaboradores, avisos=avisos)