# app/routers/clientes.py """ Rotas de gestão de CLIENTE: listar/buscar/ordenar (paginado), criar, editar, ver detalhe (contatos + status de Clientes_Chaves), excluir. """ import datetime import math from typing import Optional from urllib.parse import urlencode import pymysql from fastapi import APIRouter, Depends, Form, Request from fastapi.responses import HTMLResponse, JSONResponse, RedirectResponse from fastapi.templating import Jinja2Templates from .. import cnpj_lookup from ..auth import exigir_login from ..csrf import exigir_csrf, token_do_contexto from ..db import chaves as db_chaves from ..db import clientes as db_clientes from ..db import contatos as db_contatos from ..db import log as db_log from ..normalize import ( normalizar_documento, normalizar_telefone_com_ddi, normalizar_uf, normalizar_cep, validar_dv_cpf, validar_dv_cnpj, ) router = APIRouter(prefix="/clientes") templates = Jinja2Templates(directory="app/templates") OPCOES_POR_PAGINA = [5, 10, 15, 20, 30, 40, 50, 100] POR_PAGINA_PADRAO = 15 POR_PAGINA_COOKIE = "clientes_por_pagina" # Únicos campos que o formulário exige preenchidos — os demais (contrato, # valores, datas) podem ficar em branco e recebem um default sensato # (ver _aplicar_defaults_opcionais) em vez de bloquear o cadastro. CAMPOS_OBRIGATORIOS = ["CODIGO", "CNPJ", "RAZAO", "FANTASIA"] _CAMPOS_DECIMAIS_OPCIONAIS = ["IMPLANTACAO", "MENSALIDADE", "JUROS_MORA", "JUROS_MULTA"] _CAMPOS_DATA_OPCIONAIS = ["DATA_CONTRATO", "VENCIMENTO_CONTRATO"] FILTROS_STATUS_VALIDOS = {"ativos": 1, "inativos": 0, "todos": None} STATUS_PADRAO = "ativos" def _redirect_com_mensagem(url: str, mensagem: str, tipo: str = "success") -> RedirectResponse: separador = "&" if "?" in url else "?" return RedirectResponse(url + separador + urlencode({"mensagem": mensagem, "mensagem_tipo": tipo}), status_code=303) def _ler_form_cliente(form: dict) -> dict: return {campo: (form.get(campo) or "").strip() for campo in db_clientes.CAMPOS_CLIENTE} def _normalizar_documento_e_telefones(dados: dict) -> None: """ Grava CNPJ/CPF sempre sem máscara, TELEFONE/CELULAR sempre com DDI 55 embutido, UF em maiúsculas (2 letras) e CEP só com dígitos — evita que CLIENTE fique com formatos inconsistentes convivendo. Só afeta escritas novas feitas por este painel; registros legados só passam a ficar normalizados quando forem reeditados e salvos aqui. """ dados["CNPJ"] = normalizar_documento(dados["CNPJ"])[0] dados["TELEFONE"] = normalizar_telefone_com_ddi(dados["TELEFONE"]) dados["CELULAR"] = normalizar_telefone_com_ddi(dados["CELULAR"]) dados["UF"] = normalizar_uf(dados["UF"]) dados["CEP"] = normalizar_cep(dados["CEP"]) def _aplicar_defaults_opcionais(dados: dict) -> None: """ IMPLANTACAO/MENSALIDADE/JUROS_MORA/JUROS_MULTA/DATA_CONTRATO/ VENCIMENTO_CONTRATO não são mais exigidos no formulário — se vierem em branco, gravamos um valor neutro (0,00 / data de hoje) em vez de bloquear o cadastro. Quem cadastra pode ajustar depois. """ for campo in _CAMPOS_DECIMAIS_OPCIONAIS: if not dados.get(campo): dados[campo] = "0.00" hoje = datetime.date.today().isoformat() for campo in _CAMPOS_DATA_OPCIONAIS: if not dados.get(campo): dados[campo] = hoje def _campos_obrigatorios_faltando(dados: dict) -> list[str]: return [campo for campo in CAMPOS_OBRIGATORIOS if not dados.get(campo)] def _resolver_por_pagina(por_pagina_query: int | None, cookie_valor: str | None) -> int: """ Decide o tamanho de página efetivo: query string (se for um dos valores de OPCOES_POR_PAGINA) -> cookie clientes_por_pagina (se for um valor válido) -> padrão (15). Nunca deixa passar um valor fora de OPCOES_POR_PAGINA, mesmo que venha adulterado na URL ou no cookie. """ if por_pagina_query in OPCOES_POR_PAGINA: return por_pagina_query try: cookie_int = int(cookie_valor) if cookie_valor is not None else None except ValueError: cookie_int = None if cookie_int in OPCOES_POR_PAGINA: return cookie_int return POR_PAGINA_PADRAO def _janela_paginas(pagina: int, total_paginas: int) -> list[int | None]: """ Lista de números de página a mostrar na paginação, com None representando uma reticência ("…"). Até 7 páginas, mostra todas. Acima disso, mostra a primeira, a última, a página atual e uma vizinha de cada lado, preenchendo os buracos com None. Ex.: página 5 de 20 -> [1, None, 4, 5, 6, None, 20]. """ if total_paginas <= 7: return list(range(1, total_paginas + 1)) nucleo = {1, total_paginas, pagina} for vizinha in (pagina - 1, pagina + 1): if 1 <= vizinha <= total_paginas: nucleo.add(vizinha) ordenado = sorted(nucleo) resultado: list[int | None] = [] anterior = None for pag in ordenado: if anterior is not None and pag - anterior > 1: resultado.append(None) resultado.append(pag) anterior = pag return resultado def _validar_documento(dados: dict, confirmar_dv: bool) -> dict: """ Confere o dígito verificador do documento em CLIENTE.CNPJ (o mesmo campo aceita CPF ou CNPJ, ver app/normalize.py). CNPJ com DV inválido bloqueia sempre — não existe "cadastrar mesmo assim" para CNPJ. CPF com DV inválido mantém o fluxo já existente: aviso + confirmação. Retorna um dict para mesclar no contexto do template: {} se o documento está ok (ou é indeterminado), ou uma das chaves de erro que cliente_form.html já sabe renderizar. """ documento_normalizado, formato_valido, tipo = normalizar_documento(dados["CNPJ"]) if not formato_valido: return {} if tipo == "CNPJ": if not validar_dv_cnpj(documento_normalizado): return {"erro_cnpj_invalido": True} return {} if not validar_dv_cpf(documento_normalizado) and not confirmar_dv: return {"aviso_dv_invalido": True} return {} @router.get("", response_class=HTMLResponse) def listar( request: Request, busca: Optional[str] = None, ordenar_por: str = db_clientes.ORDENACAO_PADRAO, direcao: str = db_clientes.DIRECAO_PADRAO, pagina: int = 1, por_pagina: Optional[int] = None, status: str = STATUS_PADRAO, mensagem: Optional[str] = None, mensagem_tipo: Optional[str] = None, usuario: str = Depends(exigir_login), ): if ordenar_por not in db_clientes.ORDENACOES_VALIDAS: ordenar_por = db_clientes.ORDENACAO_PADRAO if direcao not in db_clientes.DIRECOES_VALIDAS: direcao = db_clientes.DIRECAO_PADRAO if status not in FILTROS_STATUS_VALIDOS: status = STATUS_PADRAO filtro_ativo = FILTROS_STATUS_VALIDOS[status] por_pagina_efetivo = _resolver_por_pagina(por_pagina, request.cookies.get(POR_PAGINA_COOKIE)) pagina = max(pagina, 1) total = db_clientes.contar_clientes(busca=busca or None, ordenar_por=ordenar_por, filtro_ativo=filtro_ativo) total_paginas = max(math.ceil(total / por_pagina_efetivo), 1) pagina = min(pagina, total_paginas) offset = (pagina - 1) * por_pagina_efetivo clientes = db_clientes.listar_clientes( busca=busca or None, limite=por_pagina_efetivo, offset=offset, ordenar_por=ordenar_por, direcao=direcao, filtro_ativo=filtro_ativo, ) primeiro_item = offset + 1 if total > 0 else 0 ultimo_item = min(offset + por_pagina_efetivo, total) resposta = templates.TemplateResponse( request, "clientes_lista.html", { "clientes": clientes, "busca": busca, "ordenar_por": ordenar_por, "direcao": direcao, "mensagem": mensagem, "mensagem_tipo": mensagem_tipo, "status": status, "pagina": pagina, "total_paginas": total_paginas, "total": total, "por_pagina": por_pagina_efetivo, "opcoes_por_pagina": OPCOES_POR_PAGINA, "primeiro_item": primeiro_item, "ultimo_item": ultimo_item, "janela_paginas": _janela_paginas(pagina, total_paginas), "csrf_token": token_do_contexto(request), }, ) if por_pagina in OPCOES_POR_PAGINA: # Só regrava o cookie quando a query string trouxer um valor válido # (não um valor herdado do cookie que já estava lá) — evita ficar # reescrevendo o mesmo Set-Cookie em toda requisição à toa. resposta.set_cookie( POR_PAGINA_COOKIE, str(por_pagina), samesite="lax", secure=request.url.scheme == "https", httponly=True, ) return resposta @router.get("/novo", response_class=HTMLResponse) def form_novo(request: Request, usuario: str = Depends(exigir_login)): dados = {c: "" for c in db_clientes.CAMPOS_CLIENTE} dados["CODIGO"] = db_clientes.proximo_codigo_sugerido() dados["ATIVO"] = "1" return templates.TemplateResponse( request, "cliente_form.html", { "cliente": None, "dados": dados, "campos": db_clientes.CAMPOS_CLIENTE, "campos_obrigatorios": CAMPOS_OBRIGATORIOS, "csrf_token": token_do_contexto(request), }, ) @router.post("/novo", dependencies=[Depends(exigir_csrf)]) async def criar(request: Request, usuario: str = Depends(exigir_login)): form = dict((await request.form())) dados = _ler_form_cliente(form) _normalizar_documento_e_telefones(dados) confirmar_dv = form.get("confirmar_dv_invalido") == "1" contexto_base = { "cliente": None, "dados": dados, "campos": db_clientes.CAMPOS_CLIENTE, "campos_obrigatorios": CAMPOS_OBRIGATORIOS, "csrf_token": token_do_contexto(request), } faltando = _campos_obrigatorios_faltando(dados) if faltando: return templates.TemplateResponse( request, "cliente_form.html", {**contexto_base, "erro_campos_obrigatorios": faltando} ) if db_clientes.codigo_em_uso(dados["CODIGO"]): db_log.registrar(usuario, "criar", "cliente", dados["CODIGO"], False, "código já em uso") return templates.TemplateResponse( request, "cliente_form.html", {**contexto_base, "erro_codigo_duplicado": True} ) if dados["CNPJ"] and db_clientes.cnpj_em_uso(dados["CNPJ"]): db_log.registrar(usuario, "criar", "cliente", dados["CODIGO"], False, "CNPJ/CPF já cadastrado") return templates.TemplateResponse( request, "cliente_form.html", {**contexto_base, "erro_cnpj_duplicado": True} ) extra_erro = _validar_documento(dados, confirmar_dv) if extra_erro: return templates.TemplateResponse(request, "cliente_form.html", {**contexto_base, **extra_erro}) _aplicar_defaults_opcionais(dados) try: novo_id = db_clientes.criar_cliente(dados) except (pymysql.err.DataError, pymysql.err.IntegrityError) as exc: db_log.registrar(usuario, "criar", "cliente", dados["CODIGO"], False, str(exc)) return templates.TemplateResponse( request, "cliente_form.html", {**contexto_base, "dados": dados, "erro_gravacao": str(exc)} ) db_log.registrar(usuario, "criar", "cliente", dados["CODIGO"], True) return _redirect_com_mensagem(f"/clientes/{novo_id}", "Cliente criado.") @router.get("/consulta-cnpj/{documento}") def consulta_cnpj(documento: str, usuario: str = Depends(exigir_login)): """ Usado via JS pelo formulário de cliente (ver cliente_form.html) para preencher automaticamente Razão/Fantasia/Endereço/etc. ao digitar um CNPJ. Só consulta para CNPJ (nunca CPF — dado pessoal, sem consulta pública). Ver app/cnpj_lookup.py para a fonte (BrasilAPI). """ documento_normalizado, formato_valido, tipo = normalizar_documento(documento) if not formato_valido or tipo != "CNPJ": return JSONResponse({"encontrado": False}, status_code=404) dados = cnpj_lookup.consultar_cnpj(documento_normalizado) if dados is None: return JSONResponse({"encontrado": False}, status_code=404) return {"encontrado": True, "dados": dados} @router.get("/{cliente_id}", response_class=HTMLResponse) def detalhe( request: Request, cliente_id: int, mensagem: Optional[str] = None, mensagem_tipo: Optional[str] = None, usuario: str = Depends(exigir_login), ): cliente = db_clientes.obter_cliente(cliente_id) if cliente is None: return _redirect_com_mensagem("/clientes", "Cliente não encontrado.", "danger") contatos = db_contatos.listar_contatos_por_cliente(cliente_id) status = db_chaves.status_chaves(cliente["CODIGO"]) return templates.TemplateResponse( request, "cliente_detalhe.html", { "cliente": cliente, "contatos": contatos, "status_chaves": status, "mensagem": mensagem, "mensagem_tipo": mensagem_tipo, "csrf_token": token_do_contexto(request), }, ) @router.get("/{cliente_id}/editar", response_class=HTMLResponse) def form_editar(request: Request, cliente_id: int, usuario: str = Depends(exigir_login)): cliente = db_clientes.obter_cliente(cliente_id) if cliente is None: return _redirect_com_mensagem("/clientes", "Cliente não encontrado.", "danger") return templates.TemplateResponse( request, "cliente_form.html", { "cliente": cliente, "dados": cliente, "campos": db_clientes.CAMPOS_CLIENTE, "campos_obrigatorios": CAMPOS_OBRIGATORIOS, "csrf_token": token_do_contexto(request), }, ) @router.post("/{cliente_id}/editar", dependencies=[Depends(exigir_csrf)]) async def editar(request: Request, cliente_id: int, usuario: str = Depends(exigir_login)): if db_clientes.obter_cliente(cliente_id) is None: # Sem essa checagem, um cliente_id inexistente (ex.: URL editada à # mão, ou um duplo submit depois de já ter sido excluído em outra # aba) caía direto no UPDATE — que não afeta nenhuma linha, não # levanta erro nenhum do MySQL, e ainda assim registrava "editar # ... sucesso" no log antes de redirecionar pra uma página de # detalhe que também vai dizer "cliente não encontrado". return _redirect_com_mensagem("/clientes", "Cliente não encontrado.", "danger") form = dict((await request.form())) dados = _ler_form_cliente(form) _normalizar_documento_e_telefones(dados) confirmar_dv = form.get("confirmar_dv_invalido") == "1" contexto_base = { "cliente": {"ID": cliente_id}, "dados": dados, "campos": db_clientes.CAMPOS_CLIENTE, "campos_obrigatorios": CAMPOS_OBRIGATORIOS, "csrf_token": token_do_contexto(request), } faltando = _campos_obrigatorios_faltando(dados) if faltando: return templates.TemplateResponse( request, "cliente_form.html", {**contexto_base, "erro_campos_obrigatorios": faltando} ) if db_clientes.codigo_em_uso(dados["CODIGO"], ignorar_id=cliente_id): db_log.registrar(usuario, "editar", "cliente", str(cliente_id), False, "código já em uso") return templates.TemplateResponse( request, "cliente_form.html", {**contexto_base, "erro_codigo_duplicado": True} ) if dados["CNPJ"] and db_clientes.cnpj_em_uso(dados["CNPJ"], ignorar_id=cliente_id): db_log.registrar(usuario, "editar", "cliente", str(cliente_id), False, "CNPJ/CPF já cadastrado") return templates.TemplateResponse( request, "cliente_form.html", {**contexto_base, "erro_cnpj_duplicado": True} ) extra_erro = _validar_documento(dados, confirmar_dv) if extra_erro: return templates.TemplateResponse(request, "cliente_form.html", {**contexto_base, **extra_erro}) _aplicar_defaults_opcionais(dados) try: db_clientes.atualizar_cliente(cliente_id, dados) except (pymysql.err.DataError, pymysql.err.IntegrityError) as exc: db_log.registrar(usuario, "editar", "cliente", str(cliente_id), False, str(exc)) return templates.TemplateResponse( request, "cliente_form.html", {**contexto_base, "dados": dados, "erro_gravacao": str(exc)} ) db_log.registrar(usuario, "editar", "cliente", str(cliente_id), True) return _redirect_com_mensagem(f"/clientes/{cliente_id}", "Cliente atualizado.") @router.post("/{cliente_id}/excluir", dependencies=[Depends(exigir_csrf)]) def excluir(cliente_id: int, usuario: str = Depends(exigir_login)): try: excluido = db_clientes.excluir_cliente(cliente_id) except db_clientes.ClienteComDependenciasError as exc: db_log.registrar(usuario, "excluir", "cliente", str(cliente_id), False, str(exc)) return _redirect_com_mensagem(f"/clientes/{cliente_id}", str(exc), "danger") if not excluido: db_log.registrar(usuario, "excluir", "cliente", str(cliente_id), False, "cliente não encontrado") return _redirect_com_mensagem("/clientes", "Cliente não encontrado.", "danger") db_log.registrar(usuario, "excluir", "cliente", str(cliente_id), True) return _redirect_com_mensagem("/clientes", "Cliente excluído.")