GTI_GERENCIADOR_FISCAL/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).
"""
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",
]

_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,
) -> list[dict]:
    coluna = ORDENACOES_VALIDAS.get(ordenar_por, ORDENACOES_VALIDAS[ORDENACAO_PADRAO])
    direcao_sql = DIRECOES_VALIDAS.get(direcao, DIRECOES_VALIDAS[DIRECAO_PADRAO])
    conn = get_connection()
    try:
        with conn.cursor() as cur:
            if busca:
                # 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.
                sql = (
                    f"SELECT {_COLUNAS_SELECT} FROM CLIENTE "
                    f"WHERE {coluna} LIKE %s ORDER BY {coluna} {direcao_sql} LIMIT %s OFFSET %s"
                )
                cur.execute(sql, (f"%{busca}%", limite, offset))
            else:
                cur.execute(
                    f"SELECT {_COLUNAS_SELECT} FROM CLIENTE ORDER BY {coluna} {direcao_sql} LIMIT %s OFFSET %s",
                    (limite, offset),
                )
            return 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) -> int:
    coluna = ORDENACOES_VALIDAS.get(ordenar_por, ORDENACOES_VALIDAS[ORDENACAO_PADRAO])
    conn = get_connection()
    try:
        with conn.cursor() as cur:
            if busca:
                cur.execute(f"SELECT COUNT(*) AS total FROM CLIENTE WHERE {coluna} LIKE %s", (f"%{busca}%",))
            else:
                cur.execute("SELECT COUNT(*) AS total FROM CLIENTE")
            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 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) -> None:
    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
    finally:
        conn.close()