"""Importação de dados por planilha (.xlsx), no formato do modelo gerado por `importacao_modelo.py`.

Fluxo: lê o arquivo, valida cada linha (tipos, obrigatórios, opções) e aplica na ordem de dependência dentro da sessão
(com `flush`, sem `commit` por linha). Os erros são **acumulados** (aba, linha, coluna, mensagem). No fim, `gravar=False`
(prévia) ou qualquer erro → `rollback` (nada persiste, e a prévia é fiel porque percorre o mesmo caminho do código);
`gravar=True` sem erros → `commit`. Cria código novo e atualiza código existente (custos pelo histórico "Atualizado em");
nunca apaga.
"""

import unicodedata
import zipfile
from collections.abc import Callable
from dataclasses import dataclass, field
from datetime import date, datetime
from decimal import ROUND_HALF_UP, Decimal, InvalidOperation
from io import BytesIO
from typing import Any

from sqlalchemy.exc import DBAPIError
from sqlalchemy.orm import Session

from custos_precificacao import arvore, despesas as despesas_dominio, grupos, historico, impostos, planejamento_schema, vinculos
from custos_precificacao.importacao_abas import (
    ABAS,
    ABAS_AUXILIARES,
    ABAS_POR_NOME,
    DECIMAL,
    INTEIRO,
    MES,
    OPCAO,
    PCT,
    SEMPRE,
    CRIAR,
    Aba,
    Coluna,
)
from custos_precificacao.models import (
    AliquotaImposto,
    Ambiente,
    Area,
    Atividade,
    CenarioPrecificacao,
    CentroDeCusto,
    Despesa,
    Direcionador,
    Funcionario,
    IcmsCompra,
    IcmsDesenvolve,
    IcmsDestino,
    INICIO_DOS_TEMPOS,
    Insumo,
    Macroprocesso,
    Maquina,
    Planejamento,
    PlanoDeContas,
    Premissa,
    Processo,
    Produto,
    RoteiroInsumo,
    VolumeProjetado,
)

LIMITE_BYTES = 5 * 1024 * 1024
LIMITE_LINHAS = 20_000
_MONEY = Decimal("0.000001")
_PCT = Decimal("0.0001")
_MAXIMO = Decimal("100000000")  # 1e8: acima disso é quase certamente erro de digitação
_MAX_TEXTO = {"codigo": 20, "codigo_fopag": 20, "cnpj": 20, "texto": 1000, "nome": 200}


_EM_ARVORE = (PlanoDeContas, CentroDeCusto, Ambiente)  # cadastros em árvore: código hierárquico, só nó analítico recebe lançamento

class ErroImportacao(ValueError):
    """Problema do arquivo como um todo (formato, tamanho, aba/coluna desconhecida)."""


class LinhaErro(Exception):
    """Erro de uma linha; `coluna` é a chave da coluna (ou None se vale pela linha toda)."""

    def __init__(self, mensagem: str, coluna: str | None = None):
        super().__init__(mensagem)
        self.coluna = coluna


@dataclass
class ErroCelula:
    aba: str
    linha: int
    coluna: str | None
    mensagem: str


@dataclass
class ResultadoAba:
    nome: str
    lidas: int = 0
    criados: int = 0
    atualizados: int = 0
    iguais: int = 0
    erros: list[ErroCelula] = field(default_factory=list)


@dataclass
class Resultado:
    abas: list[ResultadoAba] = field(default_factory=list)
    erros_gerais: list[str] = field(default_factory=list)
    gravado: bool = False

    @property
    def n_erros(self) -> int:
        return len(self.erros_gerais) + sum(len(a.erros) for a in self.abas)

    @property
    def ok(self) -> bool:
        return self.n_erros == 0


# --------------------------------------------------------------------------- leitura de células


def _norm(texto: Any) -> str:
    texto = unicodedata.normalize("NFKD", str(texto)).encode("ascii", "ignore").decode()
    return " ".join(texto.replace("*", " ").casefold().split())


def _vazio(valor: Any) -> bool:
    return valor is None or (isinstance(valor, str) and not valor.strip())


def _decimal(bruto: Any) -> Decimal:
    if isinstance(bruto, bool):
        raise ValueError("número inválido")
    if isinstance(bruto, (int, float, Decimal)):
        valor = Decimal(repr(bruto)) if isinstance(bruto, float) else Decimal(bruto)
    else:
        texto = str(bruto).strip().replace("R$", "").replace(" ", "").replace(" ", "")
        if "," in texto and "." in texto:
            texto = texto.replace(".", "").replace(",", ".")
        elif "," in texto:
            texto = texto.replace(",", ".")
        elif texto.count(".") > 1:  # 1.234.567 -> separador de milhar
            texto = texto.replace(".", "")
        try:
            valor = Decimal(texto)
        except InvalidOperation as exc:
            raise ValueError(f"{bruto!r} não é um número") from exc
    if not valor.is_finite():
        raise ValueError("número inválido")
    return valor


def _mes(bruto: Any) -> date:
    if isinstance(bruto, datetime):
        return bruto.date().replace(day=1)
    if isinstance(bruto, date):
        return bruto.replace(day=1)
    texto = str(bruto).strip()
    for formato in ("%Y-%m", "%m/%Y", "%Y-%m-%d", "%d/%m/%Y"):
        try:
            return datetime.strptime(texto, formato).date().replace(day=1)
        except ValueError:
            continue
    raise ValueError(f"{bruto!r} não é um mês válido (use AAAA-MM, por exemplo 2026-10)")


def _converter(coluna: Coluna, bruto: Any) -> Any:
    """Valor da célula já no tipo interno. ValueError com mensagem pronta se inválido."""
    if coluna.tipo == DECIMAL:
        valor = _decimal(bruto)
        if valor < 0:
            raise ValueError("não pode ser negativo")
        if valor > _MAXIMO:
            raise ValueError("valor grande demais — confira a digitação")
        return valor.quantize(_MONEY, ROUND_HALF_UP)
    if coluna.tipo == PCT:
        if isinstance(bruto, str) and bruto.strip().endswith("%"):
            valor = _decimal(bruto.strip()[:-1]) / 100
        else:
            valor = _decimal(bruto)
            if valor > 1:
                raise ValueError(f"{bruto!r} parece estar fora do formato de porcentagem — digite como 18% (ou formate a célula como %)")
        if valor < 0 or valor > 1:
            raise ValueError("a porcentagem deve ficar entre 0% e 100%")
        return valor.quantize(_PCT, ROUND_HALF_UP)
    if coluna.tipo == INTEIRO:
        valor = _decimal(bruto)
        if valor != valor.to_integral_value():
            raise ValueError("precisa ser um número inteiro")
        return int(valor)
    if coluna.tipo == MES:
        return _mes(bruto)
    if coluna.tipo == OPCAO:
        achado = coluna.opcoes.get(_norm(bruto))
        if achado is None:
            from custos_precificacao.importacao_abas import rotulos_da_coluna

            raise ValueError(f"{bruto!r} não é uma opção válida. Use: {', '.join(rotulos_da_coluna(coluna))}")
        return achado
    # texto
    if isinstance(bruto, float) and bruto.is_integer():
        bruto = int(bruto)
    texto = str(bruto).strip()
    maximo = _MAX_TEXTO.get(coluna.chave, 200)
    if len(texto) > maximo:
        raise ValueError(f"aceita até {maximo} caracteres (tem {len(texto)})")
    return texto


@dataclass
class _Linha:
    numero: int
    valores: dict[str, Any]


def _ler_planilha(dados: bytes, resultado: Resultado) -> dict[str, list[_Linha]]:
    """Lê as abas conhecidas. Problemas de leitura de células viram erros na aba; problemas do arquivo, ErroImportacao."""
    from defusedxml import ElementTree as _  # noqa: F401  (garante a proteção de XML do openpyxl)
    from openpyxl import load_workbook

    if len(dados) > LIMITE_BYTES:
        raise ErroImportacao(f"Arquivo grande demais ({len(dados) // 1024 // 1024} MB); o limite é {LIMITE_BYTES // 1024 // 1024} MB")
    if not dados.startswith(b"PK"):
        raise ErroImportacao("O arquivo não é uma planilha .xlsx válida")
    try:
        nomes = zipfile.ZipFile(BytesIO(dados)).namelist()
    except zipfile.BadZipFile as exc:
        raise ErroImportacao("O arquivo não é uma planilha .xlsx válida") from exc
    if any(n.lower().endswith("vbaproject.bin") for n in nomes):
        raise ErroImportacao("Planilhas com macros (.xlsm) não são aceitas — salve como .xlsx")
    try:
        livro = load_workbook(BytesIO(dados), read_only=True, data_only=True)
    except Exception as exc:  # openpyxl levanta vários tipos para arquivo corrompido
        raise ErroImportacao("Não foi possível abrir a planilha. Use um arquivo .xlsx baseado no modelo.") from exc

    por_aba: dict[str, list[_Linha]] = {}
    total_linhas = 0
    for folha in livro.worksheets:
        if folha.title in ABAS_AUXILIARES:
            continue
        aba = ABAS_POR_NOME.get(folha.title)
        if aba is None:
            resultado.erros_gerais.append(
                f"Aba desconhecida: {folha.title!r}. As abas aceitas são: {', '.join(a.nome for a in ABAS)} "
                f"(as abas {' e '.join(ABAS_AUXILIARES)} são ignoradas)."
            )
            continue
        res = ResultadoAba(aba.nome)
        resultado.abas.append(res)
        linhas = folha.iter_rows(values_only=True)
        cabecalho = next(linhas, None)
        if cabecalho is None:
            continue
        indice: dict[int, Coluna] = {}
        reconhecidos = {_norm(c.rotulo): c for c in aba.colunas} | {_norm(c.chave): c for c in aba.colunas}
        for posicao, celula in enumerate(cabecalho):
            if _vazio(celula):
                continue
            coluna = reconhecidos.get(_norm(celula))
            if coluna is None:
                res.erros.append(ErroCelula(aba.nome, 1, str(celula), f"Coluna desconhecida {str(celula)!r}. Use o cabeçalho do modelo."))
            elif coluna in indice.values():
                res.erros.append(ErroCelula(aba.nome, 1, coluna.rotulo, "Coluna repetida no cabeçalho."))
            else:
                indice[posicao] = coluna
        faltando = [c.rotulo for c in aba.colunas if c.obrigatoria == SEMPRE and c not in indice.values()]
        if faltando:
            res.erros.append(ErroCelula(aba.nome, 1, None, f"Faltam colunas obrigatórias no cabeçalho: {', '.join(faltando)}."))
        if res.erros:
            continue

        lidas: list[_Linha] = []
        for numero, celulas in enumerate(linhas, start=2):
            if celulas is None or all(_vazio(c) for c in celulas):
                continue
            primeira = next((c for c in celulas if not _vazio(c)), None)
            if isinstance(primeira, str) and primeira.strip().startswith("#"):
                continue  # linha de exemplo/comentário
            total_linhas += 1
            if total_linhas > LIMITE_LINHAS:
                raise ErroImportacao(f"A planilha tem linhas demais (limite de {LIMITE_LINHAS}).")
            valores: dict[str, Any] = {c.chave: None for c in aba.colunas}
            erro_linha = False
            for posicao, coluna in indice.items():
                bruto = celulas[posicao] if posicao < len(celulas) else None
                if _vazio(bruto):
                    continue
                try:
                    valores[coluna.chave] = _converter(coluna, bruto)
                except ValueError as exc:
                    res.erros.append(ErroCelula(aba.nome, numero, coluna.rotulo, str(exc)))
                    erro_linha = True
            res.lidas += 1
            if not erro_linha:
                lidas.append(_Linha(numero, valores))
        por_aba[aba.chave] = lidas
    return por_aba


# --------------------------------------------------------------------------- aplicação


class _Ctx:
    def __init__(self, session: Session):
        self.session = session
        self._mapas: dict[tuple[type, str], dict[str, Any]] = {}
        self.mix: dict[int, dict[str, Any]] = {}  # aba Mix de Destinos: {cenario_id: {"linha": 1ª linha, "mix": {UF: fração}}}, gravado ao fim da aba
        # Postgres: a sessão está apontada para o schema de um Planejamento (o selecionado); a planilha só trabalha nele.
        self.planejamento_em_uso: str | None = (
            session.info.get("planejamento_codigo") if planejamento_schema.e_postgres(session.get_bind()) else None
        )

    def exigir_planejamento_em_uso(self, codigo: str, coluna: str) -> None:
        if self.planejamento_em_uso and codigo != self.planejamento_em_uso:
            raise LinhaErro(
                f"Planejamento {codigo!r} não é o selecionado ({self.planejamento_em_uso!r}): selecione-o no cabeçalho antes de importar",
                coluna,
            )

    def mapa(self, modelo: type, campo: str) -> dict[str, Any]:
        chave = (modelo, campo)
        if chave not in self._mapas:
            self._mapas[chave] = {getattr(o, campo): o for o in self.session.query(modelo)}
        return self._mapas[chave]

    def achar(self, modelo: type, campo: str, codigo: str, coluna: str, rotulo: str):
        objeto = self.mapa(modelo, campo).get(codigo)
        if objeto is None and campo == "codigo" and modelo in _EM_ARVORE:
            # planilha antiga: o código antigo (chave) continua valendo
            objeto = {o.chave: o for o in self.mapa(modelo, campo).values() if o.chave}.get(codigo)
        if objeto is None:
            raise LinhaErro(f"{rotulo} {codigo!r} não existe no sistema nem na planilha", coluna)
        if modelo in _EM_ARVORE:
            try:
                arvore.exigir_lancavel(objeto, rotulo)
            except arvore.ErroDeArvore as exc:
                raise LinhaErro(str(exc), coluna) from exc
        return objeto

    def criado(self, modelo: type, campo: str, objeto: Any) -> None:
        self.session.add(objeto)
        self.session.flush()
        self.mapa(modelo, campo)[getattr(objeto, campo)] = objeto


def _exigir_ao_criar(aba: Aba, valores: dict[str, Any], *chaves: str) -> None:
    for chave in chaves:
        if valores.get(chave) is None:
            raise LinhaErro("obrigatório ao criar um registro novo", chave)


def _mes_do_registro(valores: dict[str, Any], novo: bool) -> date | None:
    mes = valores.get("atualizado_em")
    return mes if mes is not None else (INICIO_DOS_TEMPOS if novo else None)


def _atualizar(ctx: _Ctx, existente: Any, novos: dict[str, Any], mes: date | None) -> bool:
    """Aplica `novos` (só campos informados) num registro existente. Devolve True se algo mudou."""
    novos = {c: v for c, v in novos.items() if v is not None}
    mudou = any(getattr(existente, c) != v for c, v in novos.items())
    if not mudou:
        return False
    if historico.eh_versionada(type(existente)):
        try:
            historico.atualizar_com_historico(ctx.session, existente, novos, mes, commit=False)
        except historico.ErroDeHistorico as exc:
            raise LinhaErro(str(exc), "atualizado_em") from exc
    else:
        for campo, valor in novos.items():
            setattr(existente, campo, valor)
    ctx.session.flush()
    return True


def _upsert(
    ctx: _Ctx, aba: Aba, linha: _Linha, modelo: type, campo_chave: str, chave: str, campos: dict[str, Any],
    obrigatorios: tuple[str, ...] = (), *, versionado: bool = False,
) -> str:
    """Cria ou atualiza um registro por chave natural. Devolve 'criado' | 'atualizado' | 'igual'."""
    existente = ctx.mapa(modelo, campo_chave).get(chave)
    if existente is None:
        _exigir_ao_criar(aba, linha.valores, *obrigatorios)
        extras = {"atualizado_em": _mes_do_registro(linha.valores, True)} if versionado else {}
        informados = {c: v for c, v in campos.items() if v is not None}
        ctx.criado(modelo, campo_chave, modelo(**{campo_chave: chave}, **informados, **extras))
        return "criado"
    return "atualizado" if _atualizar(ctx, existente, campos, _mes_do_registro(linha.valores, False)) else "igual"


def _simples(modelo: type, campo_chave: str, mapeamento: dict[str, str], obrigatorios: tuple[str, ...], versionado: bool = False):
    def aplicar(ctx: _Ctx, aba: Aba, linha: _Linha) -> str:
        v = linha.valores
        campos = {destino: v.get(origem) for origem, destino in mapeamento.items()}
        return _upsert(ctx, aba, linha, modelo, campo_chave, v[campo_chave], campos, obrigatorios, versionado=versionado)

    return aplicar


def _ref(ctx: _Ctx, v: dict, chave: str, modelo: type, campo: str, rotulo: str) -> int | None:
    codigo = v.get(chave)
    return None if codigo is None else ctx.achar(modelo, campo, codigo, chave, rotulo).id


def _unidade(v: dict, coluna: str = "unidade") -> str | None:
    unidade = (v.get(coluna) or "").strip().lower() or None
    if unidade and len(unidade) > 10:
        raise LinhaErro("a unidade aceita até 10 caracteres", coluna)
    return unidade


_ROTULO_CLASSE = {"icms": "ICMS", "cbs": "CBS", "ir_csll": "IR/CSLL"}


def _tem_creditos(v: dict) -> bool:
    return any(v.get(f"{p}_{c}") is not None for c in impostos.CLASSES for p in ("credito", "aliquota")) or v.get("origem_icms") is not None


def _creditos_do_item(ctx: _Ctx, v: dict, origem: str, item_id: int) -> bool:
    """Aplica as colunas de crédito (ICMS, CBS, IR/CSLL, alíquota própria e origem da compra) sobre as classes já marcadas do item: Sim marca, Não
    desmarca, vazio não mexe. True se algo mudou."""
    if not _tem_creditos(v):
        return False
    atuais = impostos.classes_do_item(ctx.session, origem, item_id)
    novas = dict(atuais)
    for classe in impostos.CLASSES:
        marca, aliquota = v.get(f"credito_{classe}"), v.get(f"aliquota_{classe}")
        faixa_pedida = v.get("origem_icms") if classe == "icms" else None
        if marca == "nao":
            if aliquota is not None or faixa_pedida is not None:
                raise LinhaErro(f"alíquota própria/origem não valem com Crédito {_ROTULO_CLASSE[classe]} = Não", f"credito_{classe}")
            novas.pop(classe, None)
            continue
        if marca is None and aliquota is None and faixa_pedida is None:
            continue
        if marca is None and classe not in atuais:
            raise LinhaErro(f"informe Crédito {_ROTULO_CLASSE[classe]} = Sim para usar alíquota própria ou origem da compra", f"credito_{classe}")
        base = novas.get(classe, impostos.ClasseMarcada())
        faixa = base.faixa if faixa_pedida is None else (None if faixa_pedida == "BA" else faixa_pedida)
        novas[classe] = impostos.ClasseMarcada(base.aliquota if aliquota is None else aliquota, faixa)
    if novas == atuais:
        return False
    try:
        impostos.definir_classes_do_item(ctx.session, origem, item_id, novas)
    except ValueError as exc:
        raise LinhaErro(str(exc), "credito_icms") from exc
    return True


def _no_em_arvore(modelo: type, rotulo: str, obrigatorios: tuple[str, ...], campos_de, *, versionado: bool = False):
    """Aba de cadastro em árvore: o código é hierárquico (1, 1.01, 1.01.001); o pai é o código sem o último trecho e
    precisa existir (no sistema ou antes na planilha). Código existente (ou a chave antiga) atualiza."""

    def aplicar(ctx: _Ctx, aba: Aba, linha: _Linha) -> str:
        v = linha.valores
        codigo = v["codigo"]
        campos = {"nome": v["nome"], **campos_de(ctx, v)}
        existente = ctx.mapa(modelo, "codigo").get(codigo) or {
            o.chave: o for o in ctx.mapa(modelo, "codigo").values() if o.chave}.get(codigo)
        if existente is not None:
            return "atualizado" if _atualizar(ctx, existente, campos, _mes_do_registro(v, False)) else "igual"
        _exigir_ao_criar(aba, v, *obrigatorios)
        pai = None
        if "." in codigo:
            pai_codigo = codigo.rsplit(".", 1)[0]
            pai = ctx.mapa(modelo, "codigo").get(pai_codigo)
            if pai is None:
                raise LinhaErro(f"O pai {pai_codigo!r} não existe: cadastre-o antes (na planilha ou no sistema)", "codigo")
        extras = {c: x for c, x in campos.items() if x is not None and c != "nome"}
        if versionado:
            extras["atualizado_em"] = _mes_do_registro(v, True)
        try:
            no = arvore.criar_no(ctx.session, modelo, nome=campos["nome"], pai_id=pai.id if pai else None,
                                 sufixo=codigo.rsplit(".", 1)[-1], **extras)
        except arvore.ErroDeArvore as exc:
            raise LinhaErro(str(exc), "codigo") from exc
        ctx.mapa(modelo, "codigo")[no.codigo] = no
        return "criado"

    return aplicar


_aplicar_centro = _no_em_arvore(
    CentroDeCusto, "Centro de Custo", ("nome",),
    lambda ctx, v: {"produtivo": None if v.get("produtivo") is None else v["produtivo"] == "sim"},
)
_aplicar_ambiente = _no_em_arvore(
    Ambiente, "Instalação", ("nome", "classificacao"),
    lambda ctx, v: {
        "classificacao": v["classificacao"], "horas_uteis_mensais": v["horas_uteis_mensais"],
        "centro_de_custo_id": _ref(ctx, v, "centro_de_custo", CentroDeCusto, "codigo", "Centro de Custo"),
    },
    versionado=True,
)


def _maquina_da_categoria(categoria: str):
    """Aba Máquinas ('maquina') ou Outros ('outro'): mesma tabela e mesmos campos; o código não pode mudar de categoria."""
    rotulo = "Outros" if categoria == "outro" else "Máquina"

    def aplicar(ctx: _Ctx, aba: Aba, linha: _Linha) -> str:
        v = linha.valores
        existente = ctx.mapa(Maquina, "codigo").get(v["codigo"])
        if existente is not None and existente.categoria != categoria:
            outra = "Máquinas" if categoria == "outro" else "Outros"
            raise LinhaErro(f"o código {v['codigo']!r} já existe na aba {outra}; um elemento não muda de categoria", "codigo")
        campos = {
            "nome": v["nome"], "classificacao": v["classificacao"], "horas_uteis_mensais": v["horas_uteis_mensais"],
            "centro_de_custo_id": _ref(ctx, v, "centro_de_custo", CentroDeCusto, "codigo", "Centro de Custo"),
            "ambiente_id": _ref(ctx, v, "ambiente", Ambiente, "codigo", "Instalação"),
            **({"categoria": categoria} if existente is None else {}),
        }
        return _upsert(ctx, aba, linha, Maquina, "codigo", v["codigo"], campos, ("nome", "classificacao"), versionado=True)

    aplicar.__doc__ = f"Elemento {rotulo}."
    return aplicar


_aplicar_maquina = _maquina_da_categoria("maquina")
_aplicar_outro = _maquina_da_categoria("outro")


def _aplicar_funcionario(ctx: _Ctx, aba: Aba, linha: _Linha) -> str:
    v = linha.valores
    campos = {
        "nome": v["nome"], "cargo": v["cargo"], "classificacao": v["classificacao"], "horas_semanais": v["horas_semanais"],
        "horas_contratadas_mensais": v["horas_contratadas_mensais"], "absenteismo_pct": v["absenteismo_pct"],
        "centro_de_custo_id": _ref(ctx, v, "centro_de_custo", CentroDeCusto, "codigo", "Centro de Custo"),
    }
    return _upsert(
        ctx, aba, linha, Funcionario, "codigo_fopag", v["codigo_fopag"], campos,
        ("nome", "cargo", "centro_de_custo", "classificacao"), versionado=True,
    )


def _exigir_categoria(tipo: str, recurso) -> None:
    """Tipo Máquina só aceita máquina e tipo Outros só aceita elemento Outros (os dois dividem a tabela `maquina`)."""
    esperada = vinculos.CATEGORIA_DO_TIPO.get(tipo)
    if esperada is not None and recurso.categoria != esperada:
        raise LinhaErro(
            f"{recurso.codigo!r} é um elemento {'Outros' if recurso.categoria == 'outro' else 'Máquina'}: use o tipo "
            f"{'Outros' if recurso.categoria == 'outro' else 'Máquina'}", "recurso",
        )


def _aplicar_gasto(ctx: _Ctx, aba: Aba, linha: _Linha, vistos: set) -> str:
    v = linha.valores
    cfg = vinculos.GASTO_RECURSOS[vinculos.TIPO_ROTEIRO_DE.get(v["tipo"], v["tipo"])]
    recurso = ctx.achar(cfg.recurso_model, cfg.chave_recurso, v["recurso"], "recurso", "Recurso")
    _exigir_categoria(v["tipo"], recurso)
    plano = ctx.achar(PlanoDeContas, "codigo", v["plano_de_contas"], "plano_de_contas", "Plano de Contas")
    chave = (v["tipo"], recurso.id, plano.id)
    if chave in vistos:
        raise LinhaErro("linha repetida na planilha (mesmo recurso e Plano de Contas)")
    vistos.add(chave)
    existentes = ctx.session.query(cfg.gasto_model).filter_by(**{cfg.campo_fk: recurso.id, "plano_de_contas_id": plano.id}).all()
    if len(existentes) > 1:
        raise LinhaErro(f"há {len(existentes)} gastos com esse recurso e Plano de Contas no sistema — edite pela tela de Elementos")
    campos = {"desembolso": v["desembolso"], "retencao": v["retencao"]}
    if not existentes:
        _exigir_ao_criar(aba, v, "desembolso")
        gasto = cfg.gasto_model(
            plano_de_contas_id=plano.id, desembolso=v["desembolso"], retencao=v["retencao"] or Decimal(0),
            atualizado_em=_mes_do_registro(v, True), **{cfg.campo_fk: recurso.id},
        )
        ctx.session.add(gasto)
        ctx.session.flush()
        _creditos_do_item(ctx, v, impostos._ORIGEM_DO_MODELO[cfg.gasto_model], gasto.id)
        return "criado"
    estado = "atualizado" if _atualizar(ctx, existentes[0], campos, _mes_do_registro(v, False)) else "igual"
    if _creditos_do_item(ctx, v, impostos._ORIGEM_DO_MODELO[cfg.gasto_model], existentes[0].id) and estado == "igual":
        estado = "atualizado"
    return estado


def _aplicar_insumo(ctx: _Ctx, aba: Aba, linha: _Linha) -> str:
    v = linha.valores
    unidade = (v.get("unidade") or "").strip().lower() or None
    if unidade and len(unidade) > 10:
        raise LinhaErro("a unidade aceita até 10 caracteres", "unidade")
    campos = {
        "nome": v["nome"], "tipo": v["tipo"], "unidade": unidade, "valor_compra_unitario": v["valor_compra_unitario"], "frete": v["frete"],
        "ipi_pct": v["ipi_pct"], "icms_pct": v["icms_pct"],
    }
    estado = _upsert(ctx, aba, linha, Insumo, "codigo", v["codigo"], campos, ("nome", "tipo"), versionado=True)
    if _creditos_do_item(ctx, v, "insumo", ctx.mapa(Insumo, "codigo")[v["codigo"]].id) and estado == "igual":
        estado = "atualizado"
    return estado


def _aplicar_produto(ctx: _Ctx, aba: Aba, linha: _Linha) -> str:
    v = linha.valores
    existente = ctx.mapa(Produto, "codigo").get(v["codigo"])
    tipo = v.get("tipo") or (existente.tipo if existente is not None else None)
    if v.get("custo_compra") is not None and tipo != "revenda":
        raise LinhaErro("o custo de compra vale só para produto de revenda", "custo_compra")
    if _tem_creditos(v) and tipo != "revenda":
        raise LinhaErro("os créditos de imposto do produto valem só para revenda (os dos insumos e elementos ficam nas abas deles)", "credito_icms")
    if v.get("perda_pct") is not None and v["perda_pct"] >= 1:
        raise LinhaErro("a perda do processo tem de ser menor que 100%", "perda_pct")
    campos = {k: v[k] for k in ("nome", "tipo", "rendimento_kg", "rendimento_un", "qtd_estimada_producao_mes", "base_quantidade", "perda_pct", "custo_compra")}
    campos["unidade"] = _unidade(v)
    marcado = v.get("tabela_de_precos")
    if marcado is not None:
        campos["tabela_de_precos"] = marcado == "sim"
    elif existente is None:  # novo e sem a coluna: cabeça se for acabado ou revenda
        campos["tabela_de_precos"] = v.get("tipo") in ("acabado", "revenda")
    estado = _upsert(ctx, aba, linha, Produto, "codigo", v["codigo"], campos, ("nome", "tipo"), versionado=True)
    if _creditos_do_item(ctx, v, "produto_revenda", ctx.mapa(Produto, "codigo")[v["codigo"]].id) and estado == "igual":
        estado = "atualizado"
    return estado


def _aplicar_roteiro(ctx: _Ctx, aba: Aba, linha: _Linha, vistos: set) -> str:
    v = linha.valores
    tipo = v["tipo"]
    produto = ctx.achar(Produto, "codigo", v["produto"], "produto", "Produto")
    atividade = ctx.achar(Atividade, "codigo", v["atividade"], "atividade", "Atividade")
    tipo_elemento = vinculos.TIPO_ROTEIRO_DE.get(tipo, tipo)  # 'outros' usa a linha de roteiro da máquina
    if tipo_elemento in vinculos.ROTEIRO_RECURSOS:
        cfg = vinculos.ROTEIRO_RECURSOS[tipo_elemento]
        if tipo_elemento == "mao-de-obra":  # grupo Depto-Cargo (ou o código FOPAG de um colaborador, que vale como o grupo dele)
            ctx.session.flush()  # forma os grupos dos colaboradores criados nesta mesma planilha
            alvo = grupos.grupo_do_codigo(ctx.session, v["recurso"])
            if alvo is None:
                raise LinhaErro(f"Grupo de Mão de obra {v['recurso']!r} não existe no sistema (o código é o do grupo Depto-Cargo ou o FOPAG de um colaborador)", "recurso")
        else:
            alvo = ctx.achar(cfg.recurso_model, cfg.chave_recurso, v["recurso"], "recurso", "Recurso")
        _exigir_categoria(tipo, alvo)
        modelo, campo_fk, campo_qtd = cfg.roteiro_model, cfg.campo_fk, "tempo_horas"
    else:
        modelo, campo_qtd = RoteiroInsumo, "quantidade_kg"
        if tipo == "insumo":
            alvo = ctx.achar(Insumo, "codigo", v["recurso"], "recurso", "Insumo")
            campo_fk = "insumo_id"
        else:
            alvo = ctx.achar(Produto, "codigo", v["recurso"], "recurso", "Produto semiacabado")
            campo_fk = "insumo_produto_id"
            if alvo.id == produto.id:
                raise LinhaErro("um Produto não pode ser insumo de si mesmo", "recurso")
    direcionador = None
    if v.get("direcionador"):
        if tipo_elemento not in vinculos.ROTEIRO_RECURSOS:
            raise LinhaErro("o Direcionador vale só para mão de obra, instalação, máquina e outros", "direcionador")
        direcionador = ctx.achar(Direcionador, "codigo", v["direcionador"], "direcionador", "Direcionador")
        if direcionador.base != "tempo":
            raise LinhaErro(f"o Direcionador {direcionador.codigo!r} é de base {direcionador.base!r}; elementos usam base 'tempo'", "direcionador")
        if direcionador.atividade_id != atividade.id:
            raise LinhaErro(f"o Direcionador {direcionador.codigo!r} não pertence à Atividade {atividade.codigo!r}", "direcionador")
    if tipo_elemento in vinculos.ROTEIRO_RECURSOS and produto.tipo == "revenda":
        raise LinhaErro(f"o produto de revenda {produto.codigo!r} não recebe elementos de custo", "produto")
    chave = (tipo, produto.id, atividade.id, alvo.id)
    if chave in vistos:
        raise LinhaErro("linha repetida na planilha (mesmo Produto, Atividade e recurso)")
    vistos.add(chave)
    existentes = ctx.session.query(modelo).filter_by(produto_id=produto.id, atividade_id=atividade.id, **{campo_fk: alvo.id}).all()
    if len(existentes) > 1:
        raise LinhaErro(f"há {len(existentes)} linhas iguais no roteiro do sistema — edite pela tela de Fichas Técnicas")
    extras = {"direcionador_id": direcionador.id} if direcionador is not None else {}
    unidade_linha = _unidade(v)
    if unidade_linha is not None or v.get("perda_pct") is not None:
        if tipo_elemento in vinculos.ROTEIRO_RECURSOS:
            raise LinhaErro("a Unidade e a Perda valem só para insumo e semiacabado", "unidade" if unidade_linha is not None else "perda_pct")
        if v.get("perda_pct") is not None and v["perda_pct"] >= 1:
            raise LinhaErro("a perda tem de ser menor que 100%", "perda_pct")
        extras |= {c: x for c, x in (("unidade", unidade_linha), ("perda_pct", v.get("perda_pct"))) if x is not None}
    if v.get("ordem") is not None:
        extras["ordem"] = v["ordem"]
    if not existentes:
        _exigir_ao_criar(aba, v, "quantidade")
        ctx.session.add(modelo(
            produto_id=produto.id, atividade_id=atividade.id, atualizado_em=_mes_do_registro(v, True),
            **{campo_fk: alvo.id, campo_qtd: v["quantidade"]}, **extras,
        ))
        ctx.session.flush()
        return "criado"
    return "atualizado" if _atualizar(ctx, existentes[0], {campo_qtd: v["quantidade"], **extras}, _mes_do_registro(v, False)) else "igual"


def _aplicar_cenario(ctx: _Ctx, aba: Aba, linha: _Linha) -> str:
    v = linha.valores
    existente = ctx.mapa(CenarioPrecificacao, "codigo").get(v["codigo"])
    if existente is not None and existente.a_alocar:
        raise LinhaErro("o Centro 'Produção a alocar' é do sistema: não se altera", "codigo")
    return _upsert(
        ctx, aba, linha, CenarioPrecificacao, "codigo", v["codigo"],
        {"nome": v["nome"], "descricao": v["descricao"], "percentual_desconto": v["percentual_desconto"]}, ("nome",),
    )


def _aplicar_volume(ctx: _Ctx, aba: Aba, linha: _Linha, vistos: set) -> str:
    v = linha.valores
    ctx.exigir_planejamento_em_uso(v["planejamento"], "planejamento")
    plano = ctx.achar(Planejamento, "codigo", v["planejamento"], "planejamento", "Planejamento")
    cenario = ctx.achar(CenarioPrecificacao, "codigo", v["cenario"], "cenario", "Cenário")
    produto = ctx.achar(Produto, "codigo", v["produto"], "produto", "Produto")
    chave = (plano.id, cenario.id, produto.id)
    if chave in vistos:
        raise LinhaErro("linha repetida na planilha (mesmo Planejamento, Cenário e Produto)")
    vistos.add(chave)
    existente = ctx.session.query(VolumeProjetado).filter_by(planejamento_id=plano.id, cenario_id=cenario.id, produto_id=produto.id).one_or_none()
    if existente is None:
        ctx.session.add(VolumeProjetado(planejamento_id=plano.id, cenario_id=cenario.id, produto_id=produto.id, quantidade=v["quantidade"]))
        ctx.session.flush()
        return "criado"
    return "atualizado" if _atualizar(ctx, existente, {"quantidade": v["quantidade"]}, None) else "igual"


def _aplicar_despesa(ctx: _Ctx, aba: Aba, linha: _Linha) -> str:
    v = linha.valores
    existente = ctx.mapa(Despesa, "codigo").get(v["codigo"])
    centro_resultado = None
    if v.get("cenario") is not None:
        centro_resultado = ctx.achar(CenarioPrecificacao, "codigo", v["cenario"], "cenario", "Centro de Resultados")
        if centro_resultado.a_alocar:
            raise LinhaErro("o Centro 'Produção a alocar' não recebe despesas adicionais: ele segue a precificação primária", "cenario")
        if existente is not None and existente.cenario_id != centro_resultado.id:
            raise LinhaErro("a despesa não muda de Centro de Resultados (ou de Planejamento para Centro)", "cenario")
    tipo = v.get("tipo") or (existente.tipo if existente is not None else None)
    if tipo == "variavel" and (v.get("valor") is not None or v.get("direcionador") is not None):
        raise LinhaErro("despesa variável usa só o % da receita (sem valor nem Direcionador)", "valor" if v.get("valor") is not None else "direcionador")
    if tipo == "fixa" and v.get("percentual_receita") is not None:
        raise LinhaErro("despesa fixa usa o valor mensal (sem % da receita)", "percentual_receita")
    plano_id = _ref(ctx, v, "plano_de_contas", PlanoDeContas, "codigo", "Plano de Contas")
    centro_id = _ref(ctx, v, "centro_de_custo", CentroDeCusto, "codigo", "Centro de Custo")
    direcionador_id = _ref(ctx, v, "direcionador", Direcionador, "codigo", "Direcionador")
    if existente is None:
        _exigir_ao_criar(aba, v, "titulo", "tipo", "plano_de_contas", "centro_de_custo")
        despesa = Despesa(
            codigo=v["codigo"], titulo=v["titulo"], tipo=tipo, plano_de_contas_id=plano_id, centro_de_custo_id=centro_id, direcionador_id=direcionador_id,
            valor=v.get("valor") or Decimal(0), percentual_receita=v.get("percentual_receita"), descricao=v.get("descricao"),
            cenario_id=centro_resultado.id if centro_resultado is not None else None,
        )
        try:
            despesas_dominio.validar_despesa(despesa)
        except ValueError as exc:
            raise LinhaErro(str(exc)) from exc
        ctx.criado(Despesa, "codigo", despesa)
        _creditos_do_item(ctx, v, "despesa", despesa.id)
        return "criado"
    antes = (existente.titulo, existente.tipo, existente.plano_de_contas_id, existente.centro_de_custo_id, existente.direcionador_id, existente.valor,
             existente.percentual_receita, existente.descricao)
    for campo, valor in (("titulo", v.get("titulo")), ("tipo", tipo), ("plano_de_contas_id", plano_id), ("centro_de_custo_id", centro_id),
                         ("direcionador_id", direcionador_id), ("valor", v.get("valor")), ("percentual_receita", v.get("percentual_receita")),
                         ("descricao", v.get("descricao"))):
        if valor is not None:
            setattr(existente, campo, valor)
    try:
        despesas_dominio.validar_despesa(existente)
    except ValueError as exc:
        ctx.session.expire(existente)
        raise LinhaErro(str(exc)) from exc
    depois = (existente.titulo, existente.tipo, existente.plano_de_contas_id, existente.centro_de_custo_id, existente.direcionador_id, existente.valor,
              existente.percentual_receita, existente.descricao)
    ctx.session.flush()
    mudou = antes != depois
    if _creditos_do_item(ctx, v, "despesa", existente.id):
        mudou = True
    return "atualizado" if mudou else "igual"


def _aplicar_aliquota(ctx: _Ctx, aba: Aba, linha: _Linha, vistos: set) -> str:
    v = linha.valores
    classe, mes = v["imposto"], v["vigencia"]
    if (classe, mes) in vistos:
        raise LinhaErro("linha repetida na planilha (mesmo imposto e vigência)")
    vistos.add((classe, mes))
    existente = ctx.session.query(AliquotaImposto).filter_by(classe=classe, vigencia=mes).one_or_none()
    nota = v["nota"] if v["nota"] is not None else (existente.nota if existente is not None else None)
    if existente is not None and existente.aliquota_pct == v["aliquota"] and existente.nota == nota:
        return "igual"
    try:
        impostos.definir_aliquota(ctx.session, classe, mes, v["aliquota"], nota)
    except ValueError as exc:
        raise LinhaErro(str(exc), "aliquota") from exc
    return "atualizado" if existente is not None else "criado"


def _aplicar_icms_destino(ctx: _Ctx, aba: Aba, linha: _Linha, vistos: set) -> str:
    v = linha.valores
    uf = v["uf"].upper()
    if uf in vistos:
        raise LinhaErro("linha repetida na planilha (mesma UF)", "uf")
    vistos.add(uf)
    impostos.garantir_destinos(ctx.session)
    destino = ctx.session.query(IcmsDestino).filter_by(uf=uf).one_or_none()
    if destino is None:
        raise LinhaErro(f"UF {v['uf']!r} não encontrada", "uf")
    mudou = (v["aliquota"] is not None and destino.aliquota_pct != v["aliquota"]) or (v["reducao_base"] is not None and destino.reducao_base_pct != v["reducao_base"])
    if mudou:
        impostos.atualizar_destino(ctx.session, uf, aliquota=v["aliquota"], reducao_base=v["reducao_base"])
    return "atualizado" if mudou else "igual"


def _aplicar_icms_compra(ctx: _Ctx, aba: Aba, linha: _Linha, vistos: set) -> str:
    v = linha.valores
    if v["origem"] in vistos:
        raise LinhaErro("linha repetida na planilha (mesma origem)", "origem")
    vistos.add(v["origem"])
    impostos.garantir_compras(ctx.session)
    compra = ctx.session.query(IcmsCompra).filter_by(faixa=v["origem"]).one()
    if compra.aliquota_pct == v["aliquota"]:
        return "igual"
    impostos.atualizar_compra(ctx.session, v["origem"], v["aliquota"])
    return "atualizado"


def _aplicar_mix(ctx: _Ctx, aba: Aba, linha: _Linha, vistos: set) -> str:
    v = linha.valores
    centro = ctx.achar(CenarioPrecificacao, "codigo", v["cenario"], "cenario", "Centro de Resultados")
    if centro.a_alocar:
        raise LinhaErro("o Centro 'Produção a alocar' não tem destino de vendas: ele segue a precificação primária", "cenario")
    uf = v["uf"].upper()
    if (centro.id, uf) in vistos:
        raise LinhaErro("linha repetida na planilha (mesmo Centro e UF)", "uf")
    vistos.add((centro.id, uf))
    impostos.garantir_destinos(ctx.session)
    if ctx.session.query(IcmsDestino).filter_by(uf=uf).one_or_none() is None:
        raise LinhaErro(f"UF {v['uf']!r} não encontrada", "uf")
    ctx.mix.setdefault(centro.id, {"linha": linha.numero, "mix": {}})["mix"][uf] = v["percentual"]
    atual = impostos.obter_mix(ctx.session, centro.id).get(uf)
    if atual is None:
        return "criado"
    return "igual" if atual == v["percentual"] else "atualizado"


def _fechar_mix(ctx: _Ctx, aba: Aba, res: "ResultadoAba") -> None:
    """Ao fim da aba, grava o mix de cada Centro que apareceu (as linhas dele substituem o anterior; a soma tem de fechar 100%)."""
    for cenario_id, dados in ctx.mix.items():
        try:
            impostos.definir_mix(ctx.session, cenario_id, dados["mix"])
        except ValueError as exc:
            res.erros.append(ErroCelula(aba.nome, dados["linha"], aba.coluna("percentual").rotulo, str(exc)))


def _aplicar_desenvolve(ctx: _Ctx, aba: Aba, linha: _Linha, vistos: set) -> str:
    v = linha.valores
    if vistos:
        raise LinhaErro("a aba Desenvolve-BA aceita uma linha só")
    vistos.add(1)
    campos: dict[str, Any] = {c: v[c] for c in ("pct_dilatado", "pct_pago_a_vista", "pct_fundo_pobreza") if v.get(c) is not None}
    if v.get("ativo") is not None:
        campos["ativo"] = v["ativo"] == "sim"
    if v.get("plano_de_contas") is not None:
        campos["plano_de_contas_codigo"] = ctx.achar(PlanoDeContas, "codigo", v["plano_de_contas"], "plano_de_contas", "Plano de Contas").codigo
    if v.get("observacao") is not None:
        campos["observacao"] = v["observacao"]
    if not campos:
        return "igual"
    existia = ctx.session.query(IcmsDesenvolve).first() is not None
    atual = impostos.obter_desenvolve(ctx.session)
    foto = lambda d: (d.ativo, d.pct_dilatado, d.pct_pago_a_vista, d.pct_fundo_pobreza, d.plano_de_contas_id, d.observacao)  # noqa: E731
    antes = foto(atual)
    try:
        impostos.definir_desenvolve(ctx.session, **campos)
    except (ValueError, arvore.ErroDeArvore) as exc:
        raise LinhaErro(str(exc), "plano_de_contas" if "plano_de_contas_codigo" in campos else None) from exc
    if not existia:
        return "criado"
    return "atualizado" if foto(atual) != antes else "igual"


def _aplicar_planejamento(ctx: _Ctx, aba: Aba, linha: _Linha) -> str:
    """Postgres: Planejamento novo nasce como cópia (tela Planejamentos), nunca da planilha; aqui só se renomeia o selecionado."""
    v = linha.valores
    if ctx.planejamento_em_uso:
        ctx.exigir_planejamento_em_uso(v["codigo"], "codigo")
    return _simples(Planejamento, "codigo", {"nome": "nome"}, ("nome",))(ctx, aba, linha)


def _aplicar_premissa(ctx: _Ctx, aba: Aba, linha: _Linha) -> str:
    v = linha.valores
    if v.get("planejamento"):
        ctx.exigir_planejamento_em_uso(v["planejamento"], "planejamento")
    campos = {
        "tipo": v["tipo"], "texto": v["texto"],
        "planejamento_id": _ref(ctx, v, "planejamento", Planejamento, "codigo", "Planejamento"),
    }
    return _upsert(ctx, aba, linha, Premissa, "codigo", v["codigo"], campos, ("planejamento", "tipo", "texto"))


_aplicar_plano = _no_em_arvore(PlanoDeContas, "Plano de Contas", ("nome",), lambda ctx, v: {"tipo": v.get("tipo")})


def _dicionario(modelo: type, pai: tuple[str, type, str] | None, extras: tuple[str, ...] = ()):
    """Aba do dicionário de processos (Área → Macroprocesso → Processo → Atividade → Direcionador): código + nome (+ pai e extras)."""

    def aplicar(ctx: _Ctx, aba: Aba, linha: _Linha) -> str:
        v = linha.valores
        campos: dict[str, Any] = {"nome": v["nome"], **{c: v.get(c) for c in extras}}
        obrigatorios = ["nome", *extras]
        if pai is not None:
            coluna, modelo_pai, fk = pai
            campos[fk] = _ref(ctx, v, coluna, modelo_pai, "codigo", aba.coluna(coluna).rotulo)
            obrigatorios.append(coluna)
        return _upsert(ctx, aba, linha, modelo, "codigo", v["codigo"], campos, tuple(obrigatorios))

    return aplicar


# (chave da aba, aplicador, usa o conjunto `vistos` para detectar repetição por chave composta)
_PIPELINE: tuple[tuple[str, Callable, bool], ...] = (
    ("planos", _aplicar_plano, False),
    ("areas", _dicionario(Area, None), False),
    ("macroprocessos", _dicionario(Macroprocesso, ("area", Area, "area_id")), False),
    ("processos", _dicionario(Processo, ("macroprocesso", Macroprocesso, "macroprocesso_id")), False),
    ("atividades", _dicionario(Atividade, ("processo", Processo, "processo_id")), False),
    ("direcionadores", _dicionario(Direcionador, ("atividade", Atividade, "atividade_id"), ("base",)), False),
    ("centros", _aplicar_centro, False),
    ("ambientes", _aplicar_ambiente, False),
    ("maquinas", _aplicar_maquina, False),
    ("outros", _aplicar_outro, False),
    ("funcionarios", _aplicar_funcionario, False),
    ("gastos", _aplicar_gasto, True),
    ("insumos", _aplicar_insumo, False),
    ("produtos", _aplicar_produto, False),
    ("roteiros", _aplicar_roteiro, True),
    ("cenarios", _aplicar_cenario, False),
    ("planejamentos", _aplicar_planejamento, False),
    ("volumes", _aplicar_volume, True),
    ("despesas", _aplicar_despesa, False),
    ("aliquotas", _aplicar_aliquota, True),
    ("icms_destinos", _aplicar_icms_destino, True),
    ("icms_compras", _aplicar_icms_compra, True),
    ("mix", _aplicar_mix, True),
    ("desenvolve", _aplicar_desenvolve, True),
    ("premissas", _aplicar_premissa, False),
)
_CHAVE_PRIMARIA = {
    "planos": "codigo", "areas": "codigo", "macroprocessos": "codigo", "processos": "codigo", "atividades": "codigo",
    "direcionadores": "codigo", "centros": "codigo", "ambientes": "codigo", "maquinas": "codigo", "outros": "codigo", "funcionarios": "codigo_fopag",
    "insumos": "codigo", "produtos": "codigo", "cenarios": "codigo", "planejamentos": "codigo", "despesas": "codigo",
    "premissas": "codigo",
}


def _aplicar_aba(ctx: _Ctx, aba: Aba, res: ResultadoAba, linhas: list[_Linha], aplicador: Callable, composta: bool) -> None:
    vistos: set = set()
    codigos: dict[str, int] = {}
    pendentes_subordinado: list[tuple[_Linha, str]] = []
    estados: list[tuple[_Linha, str]] = []
    chave_primaria = _CHAVE_PRIMARIA.get(aba.chave)
    for linha in linhas:
        try:
            if chave_primaria:
                codigo = linha.valores[chave_primaria]
                if codigo in codigos:
                    raise LinhaErro(f"código {codigo!r} repetido na planilha (linha {codigos[codigo]})", chave_primaria)
                codigos[codigo] = linha.numero
            estado = aplicador(ctx, aba, linha, vistos) if composta else aplicador(ctx, aba, linha)
        except LinhaErro as exc:
            coluna = aba.coluna(exc.coluna).rotulo if exc.coluna and any(c.chave == exc.coluna for c in aba.colunas) else None
            res.erros.append(ErroCelula(aba.nome, linha.numero, coluna, str(exc)))
            continue
        except DBAPIError as exc:  # estado da transação perdido: não dá para seguir com segurança
            ctx.session.rollback()
            raise ErroImportacao(f"Erro de banco de dados em {aba.nome}, linha {linha.numero}: {exc.orig}") from exc
        estados.append((linha, estado))
        if aba.chave == "funcionarios" and linha.valores.get("subordinado_a") is not None:
            pendentes_subordinado.append((linha, estado))

    if pendentes_subordinado:  # segunda passada: o gestor pode estar mais abaixo na mesma aba
        for linha, estado in pendentes_subordinado:
            try:
                funcionario = ctx.mapa(Funcionario, "codigo_fopag")[linha.valores["codigo_fopag"]]
                gestor = ctx.achar(Funcionario, "codigo_fopag", linha.valores["subordinado_a"], "subordinado_a", "Gestor")
                if gestor.id == funcionario.id:
                    raise LinhaErro("um Funcionário não pode ser subordinado a si mesmo", "subordinado_a")
                if funcionario.subordinado_a_id != gestor.id:
                    funcionario.subordinado_a_id = gestor.id
                    if estado == "igual":
                        estados[[i for i, (l, _) in enumerate(estados) if l is linha][0]] = (linha, "atualizado")
            except LinhaErro as exc:
                res.erros.append(ErroCelula(aba.nome, linha.numero, aba.coluna(exc.coluna).rotulo if exc.coluna else None, str(exc)))
        ctx.session.flush()

    for _, estado in estados:
        if estado == "criado":
            res.criados += 1
        elif estado == "atualizado":
            res.atualizados += 1
        else:
            res.iguais += 1


def importar(session: Session, dados: bytes, *, gravar: bool) -> Resultado:
    """Valida e aplica a planilha. `gravar=False` é a prévia (tudo é desfeito no fim). Qualquer erro desfaz tudo."""
    resultado = Resultado()
    try:
        por_aba = _ler_planilha(dados, resultado)
    except ErroImportacao as exc:
        resultado.erros_gerais.append(str(exc))
        return resultado

    res_por_aba = {a.nome: a for a in resultado.abas}
    ctx = _Ctx(session)
    try:
        for chave, aplicador, composta in _PIPELINE:
            if chave not in por_aba:
                continue
            aba = next(a for a in ABAS if a.chave == chave)
            _aplicar_aba(ctx, aba, res_por_aba[aba.nome], por_aba[chave], aplicador, composta)
            if chave == "mix":
                _fechar_mix(ctx, aba, res_por_aba[aba.nome])
        resultado.abas.sort(key=lambda a: [x.nome for x in ABAS].index(a.nome))
        if gravar and resultado.ok:
            session.commit()
            resultado.gravado = True
        else:
            session.rollback()
    except ErroImportacao as exc:
        session.rollback()
        resultado.erros_gerais.append(str(exc))
    except Exception:
        session.rollback()
        raise
    return resultado