GTI_GERENCIADOR_FISCAL/gerenciador/app/db/clientes.py

"""
CRUD completo de CLIENTE. Única tabela cadastral (ID autoincrement é a PK
técnica; CODIGO é o identificador de negócio usado pelo resto do sistema —
ver licenciamento-n8n/docs/MAPEAMENTO_BANCO.md).

CLIENTE.ATIVO (TINYINT(1) NOT NULL DEFAULT 1) controla se o cliente
aparece no filtro padrão da listagem ("Ativos"). Se CLIENTE já existir de
uma instalação anterior sem essa coluna, rode:

    ALTER TABLE CLIENTE
      ADD COLUMN ATIVO TINYINT(1) NOT NULL DEFAULT 1
      AFTER DIA_VENCIMENTO;

DEFAULT 1 é intencional: todo cliente cadastrado antes da migração nasce
como Ativo — decisão do projeto (quem administra desativa manualmente os
que sabe que não são mais, ver
docs/superpowers/specs/2026-08-20-status-ativo-clientes-design.md).
Cadastros/edições novos sempre informam ATIVO explicitamente (vindo do
formulário), nunca dependem desse default.
"""
import pymysql

from .connection import get_connection

CAMPOS_CLIENTE = [
    "CODIGO", "CNPJ", "RAZAO", "FANTASIA", "IE", "ENDERECO", "CIDADE",
    "BAIRRO", "UF", "CONTATO", "TELEFONE", "CELULAR", "CEP", "IMPLANTACAO",
    "MENSALIDADE", "JUROS_MORA", "JUROS_MULTA", "DATA_CONTRATO", "EMAIL",
    "OBS", "VENCIMENTO_CONTRATO", "DIA_VENCIMENTO", "ATIVO",
]

_COLUNAS_SELECT = ", ".join(["ID", *CAMPOS_CLIENTE])

# Whitelist de colunas aceitas para ordenar a listagem — nunca interpolar
# o valor de "ordenar_por" vindo da URL direto no SQL sem passar por aqui.
ORDENACOES_VALIDAS = {
    "codigo": "CODIGO",
    "razao": "RAZAO",
    "fantasia": "FANTASIA",
    "cnpj": "CNPJ",
}
ORDENACAO_PADRAO = "razao"

# Idem para a direção — nunca interpolar "asc"/"desc" vindo da URL sem
# passar por aqui.
DIRECOES_VALIDAS = {"asc": "ASC", "desc": "DESC"}
DIRECAO_PADRAO = "asc"


class ClienteComDependenciasError(Exception):
    """Levantada ao tentar excluir um CLIENTE que ainda tem CLIENTE_CONTATO."""


def listar_clientes(
    busca: str | None = None,
    limite: int = 100,
    offset: int = 0,
    ordenar_por: str = ORDENACAO_PADRAO,
    direcao: str = DIRECAO_PADRAO,
    filtro_ativo: int | None = 1,
) -> list[dict]:
    coluna = ORDENACOES_VALIDAS.get(ordenar_por, ORDENACOES_VALIDAS[ORDENACAO_PADRAO])
    direcao_sql = DIRECOES_VALIDAS.get(direcao, DIRECOES_VALIDAS[DIRECAO_PADRAO])
    # A busca é restrita à mesma coluna escolhida em "ordenar_por" — selecionar
    # "Código" busca só em CODIGO, não em todos os campos de uma vez.
    # filtro_ativo: 1 = só ativos, 0 = só inativos, None = sem filtro (todos).
    condicoes = []
    params: list = []
    if busca:
        condicoes.append(f"{coluna} LIKE %s")
        params.append(f"%{busca}%")
    if filtro_ativo is not None:
        condicoes.append("ATIVO = %s")
        params.append(filtro_ativo)
    where_sql = f"WHERE {' AND '.join(condicoes)} " if condicoes else ""

    conn = get_connection()
    try:
        with conn.cursor() as cur:
            sql = (
                f"SELECT {_COLUNAS_SELECT} FROM CLIENTE {where_sql}"
                f"ORDER BY {coluna} {direcao_sql} LIMIT %s OFFSET %s"
            )
            cur.execute(sql, [*params, limite, offset])
            return cur.fetchall()
    finally:
        conn.close()


def listar_clientes_para_fiscal() -> list[dict]:
    """Para o endpoint interno GET /internal/fiscal/clients (app/internal_fiscal.py):
    o Fiscal espelha TODOS os clientes (ativos e inativos — inativo continua
    aparecendo em histórico do lado de lá), nunca só uma página. Payload
    mínimo: ID é a PK técnica estável (CODIGO é editável, não serve de
    identidade — mesma razão de FUNCIONARIO_PAINEL.ID no SSO). CNPJ (só dígitos) entra só para a busca/identificação no seletor do
    Fiscal; sem endereço/contato."""
    conn = get_connection()
    try:
        with conn.cursor() as cur:
            cur.execute(
                "SELECT ID, CODIGO, CNPJ, COALESCE(NULLIF(FANTASIA, ''), RAZAO) AS NOME, ATIVO "
                "FROM CLIENTE ORDER BY ID"
            )
            return [
                {"external_client_id": row["ID"], "code": row["CODIGO"], "display_name": row["NOME"], "cnpj": "".join(ch for ch in (row["CNPJ"] or "") if ch.isdigit()) or None, "active": bool(row["ATIVO"])}
                for row in cur.fetchall()
            ]
    finally:
        conn.close()


def proximo_codigo_sugerido() -> str:
    """
    Sugere o próximo CODIGO sequencial (maior CODIGO numérico + 1, com zeros
    à esquerda até 6 dígitos). É só uma sugestão pré-preenchida no formulário
    de novo cliente — o campo continua editável, e codigo_em_uso() ainda
    confere unicidade de verdade antes de gravar.
    """
    conn = get_connection()
    try:
        with conn.cursor() as cur:
            # CODIGO >= 900000 é a faixa reservada para dados de teste deste
            # painel (ver tests/test_db_clientes.py) — excluída aqui para a
            # sugestão não disparar lá em cima se sobrar alguma linha de teste.
            cur.execute(
                "SELECT MAX(CAST(CODIGO AS UNSIGNED)) AS maior FROM CLIENTE "
                "WHERE CODIGO REGEXP '^[0-9]+$' AND CAST(CODIGO AS UNSIGNED) < 900000"
            )
            maior = cur.fetchone()["maior"]
    finally:
        conn.close()
    proximo = (maior or 0) + 1
    return str(proximo).zfill(6)


def contar_clientes(busca: str | None = None, ordenar_por: str = ORDENACAO_PADRAO, filtro_ativo: int | None = 1) -> int:
    coluna = ORDENACOES_VALIDAS.get(ordenar_por, ORDENACOES_VALIDAS[ORDENACAO_PADRAO])
    condicoes = []
    params: list = []
    if busca:
        condicoes.append(f"{coluna} LIKE %s")
        params.append(f"%{busca}%")
    if filtro_ativo is not None:
        condicoes.append("ATIVO = %s")
        params.append(filtro_ativo)
    where_sql = f"WHERE {' AND '.join(condicoes)}" if condicoes else ""

    conn = get_connection()
    try:
        with conn.cursor() as cur:
            cur.execute(f"SELECT COUNT(*) AS total FROM CLIENTE {where_sql}", params)
            return cur.fetchone()["total"]
    finally:
        conn.close()


def obter_cliente(cliente_id: int) -> dict | None:
    conn = get_connection()
    try:
        with conn.cursor() as cur:
            cur.execute(f"SELECT {_COLUNAS_SELECT} FROM CLIENTE WHERE ID = %s", (cliente_id,))
            return cur.fetchone()
    finally:
        conn.close()


def cnpj_em_uso(documento_normalizado: str, ignorar_id: int | None = None) -> bool:
    """
    True se já existe outro CLIENTE com esse mesmo CNPJ/CPF (o mesmo campo
    CLIENTE.CNPJ aceita os dois — ver app/normalize.py). Compara ignorando
    pontuação em vez de igualdade exata porque CLIENTE.CNPJ tem
    formatação inconsistente em registros legados (alguns com máscara,
    outros sem — ver licenciamento-n8n/docs/ANALISE_CLIENTE_IPREAL.md);
    sem normalizar dos dois lados, um cadastro novo (sempre gravado sem
    máscara) não bateria com um legado que ainda tem pontos/traços,
    deixando passar o duplicado. `documento_normalizado` já chega sem
    máscara (normalizar_documento); a normalização abaixo só precisa
    tratar o lado do banco.
    """
    conn = get_connection()
    try:
        with conn.cursor() as cur:
            sql = (
                "SELECT ID FROM CLIENTE WHERE "
                "UPPER(REPLACE(REPLACE(REPLACE(REPLACE(CNPJ, '.', ''), '/', ''), '-', ''), ' ', '')) = %s"
            )
            params = [documento_normalizado]
            if ignorar_id is not None:
                sql += " AND ID != %s"
                params.append(ignorar_id)
            cur.execute(sql, params)
            return cur.fetchone() is not None
    finally:
        conn.close()


def codigo_em_uso(codigo: str, ignorar_id: int | None = None) -> bool:
    conn = get_connection()
    try:
        with conn.cursor() as cur:
            sql = "SELECT ID FROM CLIENTE WHERE CODIGO = %s"
            params = [codigo]
            if ignorar_id is not None:
                sql += " AND ID != %s"
                params.append(ignorar_id)
            cur.execute(sql, params)
            return cur.fetchone() is not None
    finally:
        conn.close()


def criar_cliente(dados: dict) -> int:
    colunas = ", ".join(CAMPOS_CLIENTE)
    marcadores = ", ".join(["%s"] * len(CAMPOS_CLIENTE))
    valores = [dados[campo] for campo in CAMPOS_CLIENTE]
    conn = get_connection()
    try:
        with conn.cursor() as cur:
            cur.execute(f"INSERT INTO CLIENTE ({colunas}) VALUES ({marcadores})", valores)
            return cur.lastrowid
    finally:
        conn.close()


def atualizar_cliente(cliente_id: int, dados: dict) -> None:
    atribuicoes = ", ".join(f"{campo} = %s" for campo in CAMPOS_CLIENTE)
    valores = [dados[campo] for campo in CAMPOS_CLIENTE]
    valores.append(cliente_id)
    conn = get_connection()
    try:
        with conn.cursor() as cur:
            cur.execute(f"UPDATE CLIENTE SET {atribuicoes} WHERE ID = %s", valores)
    finally:
        conn.close()


def excluir_cliente(cliente_id: int) -> bool:
    """True se a linha realmente existia (cur.rowcount > 0) — um ID
    inexistente não gera erro nenhum do MySQL, então sem checar rowcount
    o chamador não tem como distinguir "excluiu" de "não tinha nada pra
    excluir"."""
    conn = get_connection()
    try:
        with conn.cursor() as cur:
            try:
                cur.execute("DELETE FROM CLIENTE WHERE ID = %s", (cliente_id,))
            except pymysql.err.IntegrityError as exc:
                raise ClienteComDependenciasError(
                    "Este cliente tem contatos cadastrados (CLIENTE_CONTATO) — remova-os antes de excluir."
                ) from exc
            return cur.rowcount > 0
    finally:
        conn.close()


def alternar_ativo(cliente_id: int) -> bool:
    """True se a linha existia (cur.rowcount > 0) — mesma lógica de
    excluir_cliente/redefinir_senha pra distinguir "alternou" de "não
    tinha nada pra alternar". UPDATE ... SET ATIVO = 1 - ATIVO inverte o
    valor atomicamente, sem precisar de um SELECT antes."""
    conn = get_connection()
    try:
        with conn.cursor() as cur:
            cur.execute("UPDATE CLIENTE SET ATIVO = 1 - ATIVO WHERE ID = %s", (cliente_id,))
            return cur.rowcount > 0
    finally:
        conn.close()