Metodologia

from typing import Any

import polars as pl

from pipeline import formatacao as fmt
from pipeline.regras.base import Contexto, Regra, intensidade_por_razao, na_lista, resultado


def aplicar(despesas: pl.DataFrame, params: dict[str, Any], contexto: Contexto) -> pl.DataFrame:
    limite = params["limite"]
    df = despesas.filter(na_lista("categoria", params["categorias"]) & (pl.col("valor") > limite))
    return resultado(
        df.select(
            pl.col("id").alias("despesa_id"),
            intensidade_por_razao(pl.col("valor") / limite).alias("intensidade"),
            pl.col("valor")
            .map_elements(
                lambda v: (
                    f"Refeição de {fmt.reais(v)}; o limite configurado é {fmt.reais(limite)} "
                    f"({fmt.numero(v / limite)} vezes o limite)."
                ),
                pl.String,
            )
            .alias("detalhe"),
        )
    )


REGRA = Regra(
    id="R01_refeicao_alta",
    nome="Refeição acima do limite",
    descricao=(
        "Marca refeições do próprio parlamentar com valor acima de um limite fixo, igual "
        "para todos. Só se aplica às categorias que identificam refeição (na Câmara, "
        "'Fornecimento de alimentação do parlamentar'; o Senado publica alimentação junto "
        "com locomoção e hospedagem, por isso não há como separar)."
    ),
    aplicar=aplicar,
    modulo=__name__,
)

from typing import Any

import polars as pl

from pipeline import formatacao as fmt
from pipeline.regras.base import Contexto, Regra, na_lista, resultado

CHAVE = ["parlamentar_id", "fornecedor", "valor", "data"]


def aplicar(despesas: pl.DataFrame, params: dict[str, Any], contexto: Contexto) -> pl.DataFrame:
    df = despesas.filter(
        pl.col("data").is_not_null()
        & (pl.col("valor") >= params["valor_minimo"])
        & ~na_lista("categoria", params.get("categorias_excluidas", []))
    ).with_columns(
        pl.coalesce("fornecedor_cnpj_cpf", pl.col("fornecedor_nome").str.to_uppercase()).alias(
            "fornecedor"
        )
    )
    # A primeira despesa do grupo (pela ordem do id) é a referência; as demais disparam.
    df = (
        df.sort("id")
        .with_columns(
            pl.len().over(CHAVE).alias("ocorrencias"),
            pl.int_range(1, pl.len() + 1).over(CHAVE).alias("ordem"),
        )
        .filter(pl.col("ordem") > 1)
    )
    return resultado(
        df.select(
            pl.col("id").alias("despesa_id"),
            pl.lit(1.0).alias("intensidade"),
            pl.struct("ocorrencias", "valor", "data")
            .map_elements(
                lambda s: (
                    f"Há {s['ocorrencias']} despesas deste parlamentar com o mesmo fornecedor, "
                    f"o mesmo valor ({fmt.reais(s['valor'])}) e a mesma data "
                    f"({fmt.data_curta(s['data'])})."
                ),
                pl.String,
            )
            .alias("detalhe"),
        )
    )


REGRA = Regra(
    id="R02_duplicidade",
    nome="Possível duplicidade",
    descricao=(
        "Marca despesas do mesmo parlamentar com o mesmo fornecedor, o mesmo valor e a mesma "
        "data de outra despesa já registrada. A primeira delas não é marcada; as repetições "
        "sim. Valores pequenos e categorias em que a repetição é comum (como várias diárias "
        "de hotel na mesma noite) ficam de fora, conforme a configuração."
    ),
    aplicar=aplicar,
    modulo=__name__,
)

from typing import Any

import polars as pl

from pipeline import formatacao as fmt
from pipeline.regras.base import Contexto, Regra, intensidade_por_razao, resultado

GRUPO = ["categoria", "ano"]


def aplicar(despesas: pl.DataFrame, params: dict[str, Any], contexto: Contexto) -> pl.DataFrame:
    percentil = params["percentil"]
    df = (
        despesas.filter(pl.col("valor") > 0)
        .with_columns(pl.col("competencia").str.slice(0, 4).alias("ano"))
        .with_columns(
            pl.len().over(GRUPO).alias("amostra"),
            pl.col("valor").quantile(percentil, "linear").over(GRUPO).alias("limiar"),
        )
        .filter(
            (pl.col("amostra") >= params["amostra_minima"]) & (pl.col("valor") > pl.col("limiar"))
        )
    )
    rotulo = f"percentil {fmt.numero(percentil * 100, 0)}"
    return resultado(
        df.select(
            pl.col("id").alias("despesa_id"),
            intensidade_por_razao(pl.col("valor") / pl.col("limiar")).alias("intensidade"),
            pl.struct("valor", "limiar", "ano", "amostra")
            .map_elements(
                lambda s: (
                    f"Valor de {fmt.reais(s['valor'])}, acima do {rotulo} da categoria em "
                    f"{s['ano']} ({fmt.reais(s['limiar'])}, calculado sobre "
                    f"{fmt.numero(s['amostra'], 0)} despesas)."
                ),
                pl.String,
            )
            .alias("detalhe"),
        )
    )


REGRA = Regra(
    id="R03_outlier_categoria",
    nome="Valor muito acima do comum na categoria",
    descricao=(
        "Para cada categoria de despesa e cada ano, calcula o valor abaixo do qual estão 99% "
        "das despesas (percentil 99) de todos os parlamentares. Marca as despesas acima "
        "desse valor. Categorias com poucas despesas no ano não são avaliadas."
    ),
    aplicar=aplicar,
    modulo=__name__,
)

from typing import Any

import polars as pl

from pipeline import formatacao as fmt
from pipeline.regras.base import Contexto, Regra, resultado


def _detalhe(s: dict[str, Any]) -> str:
    if s["aberto_depois"]:
        return (
            f"Segundo a Receita Federal, o CNPJ do fornecedor começou a funcionar em "
            f"{fmt.data_curta(s['data_inicio'])}, depois da data da despesa "
            f"({fmt.data_curta(s['data'])})."
        )
    return (
        f"Segundo a Receita Federal, o CNPJ do fornecedor está com situação "
        f"'{s['situacao']}' desde {fmt.data_curta(s['data_situacao'])}, antes da data da "
        f"despesa ({fmt.data_curta(s['data'])})."
    )


def aplicar(despesas: pl.DataFrame, params: dict[str, Any], contexto: Contexto) -> pl.DataFrame:
    df = (
        despesas.filter(
            pl.col("data").is_not_null() & (pl.col("fornecedor_cnpj_cpf").str.len_chars() == 14)
        )
        .join(contexto.cnpjs, left_on="fornecedor_cnpj_cpf", right_on="cnpj", how="inner")
        .with_columns(
            (pl.col("data_inicio") > pl.col("data")).fill_null(False).alias("aberto_depois"),
            (
                pl.col("situacao").is_not_null()
                & (pl.col("situacao").str.to_uppercase() != "ATIVA")
                & (pl.col("data_situacao") <= pl.col("data"))
            )
            .fill_null(False)
            .alias("inativo"),
        )
        .filter(pl.col("aberto_depois") | pl.col("inativo"))
    )
    return resultado(
        df.select(
            pl.col("id").alias("despesa_id"),
            pl.lit(1.0).alias("intensidade"),
            pl.struct("aberto_depois", "data_inicio", "data", "situacao", "data_situacao")
            .map_elements(_detalhe, pl.String)
            .alias("detalhe"),
        )
    )


REGRA = Regra(
    id="R04_cnpj_situacao",
    nome="Fornecedor com CNPJ irregular na data",
    descricao=(
        "Consulta o cadastro do CNPJ do fornecedor na Receita Federal (via BrasilAPI). Marca a "
        "despesa se, na data dela, a empresa ainda não tinha sido aberta ou já estava com "
        "situação diferente de 'ativa' (baixada, inapta, suspensa ou nula). Empresas que "
        "fecharam depois da despesa não são marcadas."
    ),
    aplicar=aplicar,
    modulo=__name__,
)

from datetime import date
from typing import Any

import polars as pl

from pipeline import formatacao as fmt
from pipeline.regras.base import Contexto, Regra, na_lista, resultado
from pipeline.regras.feriados import feriado


def motivo(d: date) -> str | None:
    """'sábado', 'domingo', 'feriado nacional (Tiradentes)' ou None em dia útil."""
    if nome := feriado(d):
        return f"feriado nacional ({nome})"
    if d.weekday() >= 5:
        return fmt.DIAS_DA_SEMANA[d.weekday()]
    return None


def aplicar(despesas: pl.DataFrame, params: dict[str, Any], contexto: Contexto) -> pl.DataFrame:
    df = despesas.filter(
        pl.col("data").is_not_null()
        & (pl.col("valor") >= params["valor_minimo"])
        & na_lista("categoria", params["categorias"])
    ).with_columns(pl.col("data").map_elements(motivo, pl.String).alias("motivo"))
    df = df.filter(pl.col("motivo").is_not_null())
    return resultado(
        df.select(
            pl.col("id").alias("despesa_id"),
            pl.lit(1.0).alias("intensidade"),
            pl.struct("data", "motivo")
            .map_elements(
                lambda s: (
                    f"Documento de despesa de expediente emitido em "
                    f"{fmt.data_curta(s['data'])}, {s['motivo']}."
                ),
                pl.String,
            )
            .alias("detalhe"),
        )
    )


REGRA = Regra(
    id="R05_fim_de_semana",
    nome="Despesa de expediente em fim de semana ou feriado",
    descricao=(
        "Marca despesas de categorias típicas de expediente de escritório (manutenção de "
        "escritório, aluguel, material de consumo, consultorias, divulgação, publicações e "
        "serviços postais) cujo documento foi emitido num sábado, domingo ou feriado "
        "nacional."
    ),
    aplicar=aplicar,
    modulo=__name__,
)