feat: bloqueia CNPJ/CPF duplicado ao cadastrar ou editar cliente

Mesmo padrão já usado pro CODIGO (db_clientes.codigo_em_uso): nova
db_clientes.cnpj_em_uso(documento, ignorar_id=None) checada em
criar()/editar() antes de gravar. A comparação ignora pontuação dos dois
lados (não só normaliza o valor recebido) porque CLIENTE.CNPJ tem
formatação inconsistente em registros legados (alguns com máscara,
outros sem) — sem isso, um cadastro novo com o mesmo documento de um
legado ainda mascarado passaria batido.

app/templates/cliente_form.html: nova mensagem de erro
'erro_cnpj_duplicado', mesmo estilo visual do erro de código duplicado.

6 testes novos (unitário do db + integração do router pros dois
fluxos), suíte completa (196 testes) passando.

Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
This commit is contained in:
Gustavo
2026-08-19 09:13:43 -03:00
co-authored by Claude Sonnet 5
parent eed473c5f3
commit f940ff5768
5 changed files with 109 additions and 0 deletions
+30
View File
@@ -115,6 +115,36 @@ def obter_cliente(cliente_id: int) -> dict | None:
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:
+12
View File
@@ -183,6 +183,12 @@ async def criar(request: Request, usuario: str = Depends(exigir_login)):
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})
@@ -290,6 +296,12 @@ async def editar(request: Request, cliente_id: int, usuario: str = Depends(exigi
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})
+5
View File
@@ -8,6 +8,11 @@
O código {{ dados.CODIGO }} já está em uso por outro cliente — escolha outro código antes de salvar.
</div>
{% endif %}
{% if erro_cnpj_duplicado %}
<div class="mensagem danger">
O CNPJ/CPF {{ dados.CNPJ }} já está cadastrado para outro cliente.
</div>
{% endif %}
{% if aviso_dv_invalido %}
<div class="mensagem warning">
O dígito verificador do documento ({{ dados.CNPJ }}) não confere — confirme se digitou certo antes de salvar.