# 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, RedirectResponse from fastapi.templating import Jinja2Templates 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, validar_dv_cpf, validar_dv_cnpj router = APIRouter(prefix="/clientes") templates = Jinja2Templates(directory="app/templates") TAMANHO_PAGINA = 100 # Ú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"] 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 e telefones sempre sem máscara (só dígitos/letras), igual ao que CLIENTE_CONTATO.TELEFONE já faz — evita que CLIENTE fique com dois formatos diferentes convivendo. Só afeta escritas novas feitas por este painel; registros legados só passam a ficar sem máscara quando forem reeditados e salvos aqui. """ dados["CNPJ"] = normalizar_documento(dados["CNPJ"])[0] dados["TELEFONE"] = normalizar_telefone(dados["TELEFONE"]) dados["CELULAR"] = normalizar_telefone(dados["CELULAR"]) 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)] @router.get("", response_class=HTMLResponse) def listar( request: Request, busca: Optional[str] = None, ordenar_por: str = db_clientes.ORDENACAO_PADRAO, pagina: int = 1, 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 pagina = max(pagina, 1) total = db_clientes.contar_clientes(busca=busca or None) total_paginas = max(math.ceil(total / TAMANHO_PAGINA), 1) pagina = min(pagina, total_paginas) offset = (pagina - 1) * TAMANHO_PAGINA clientes = db_clientes.listar_clientes( busca=busca or None, limite=TAMANHO_PAGINA, offset=offset, ordenar_por=ordenar_por ) return templates.TemplateResponse( request, "clientes_lista.html", { "clientes": clientes, "busca": busca, "ordenar_por": ordenar_por, "mensagem": mensagem, "mensagem_tipo": mensagem_tipo, "pagina": pagina, "total_paginas": total_paginas, "total": total, }, ) @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() 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} ) documento_normalizado, formato_valido, tipo = normalizar_documento(dados["CNPJ"]) dv_invalido = False if formato_valido: dv_invalido = not ( validar_dv_cpf(documento_normalizado) if tipo == "CPF" else validar_dv_cnpj(documento_normalizado) ) if dv_invalido and not confirmar_dv: return templates.TemplateResponse( request, "cliente_form.html", {**contexto_base, "aviso_dv_invalido": True} ) _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("/{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)): form = dict((await request.form())) dados = _ler_form_cliente(form) _normalizar_documento_e_telefones(dados) 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} ) _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: 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") db_log.registrar(usuario, "excluir", "cliente", str(cliente_id), True) return _redirect_com_mensagem("/clientes", "Cliente excluído.")