403 lines
16 KiB
Python
403 lines
16 KiB
Python
# 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")
|
|
|
|
TAMANHO_PAGINA = 100
|
|
|
|
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"]
|
|
|
|
|
|
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,
|
|
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
|
|
|
|
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)
|
|
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
|
|
)
|
|
|
|
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,
|
|
"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),
|
|
},
|
|
)
|
|
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",
|
|
)
|
|
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()
|
|
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.")
|