diff --git a/app/db/clientes.py b/app/db/clientes.py
index 2fe7af8..139e221 100644
--- a/app/db/clientes.py
+++ b/app/db/clientes.py
@@ -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:
diff --git a/app/routers/clientes.py b/app/routers/clientes.py
index aae8757..d445706 100644
--- a/app/routers/clientes.py
+++ b/app/routers/clientes.py
@@ -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})
diff --git a/app/templates/cliente_form.html b/app/templates/cliente_form.html
index e291d0f..35d3bdd 100644
--- a/app/templates/cliente_form.html
+++ b/app/templates/cliente_form.html
@@ -8,6 +8,11 @@
O código {{ dados.CODIGO }} já está em uso por outro cliente — escolha outro código antes de salvar.
{% endif %}
+ {% if erro_cnpj_duplicado %}
+
+ O CNPJ/CPF {{ dados.CNPJ }} já está cadastrado para outro cliente.
+
+ {% endif %}
{% if aviso_dv_invalido %}
O dígito verificador do documento ({{ dados.CNPJ }}) não confere — confirme se digitou certo antes de salvar.
diff --git a/tests/test_db_clientes.py b/tests/test_db_clientes.py
index 26ce736..b042b79 100644
--- a/tests/test_db_clientes.py
+++ b/tests/test_db_clientes.py
@@ -74,6 +74,26 @@ def test_codigo_em_uso(cliente_id):
assert db_clientes.codigo_em_uso("000000") is False
+def test_cnpj_em_uso(cliente_id):
+ assert db_clientes.cnpj_em_uso("11222333000181") is True
+ assert db_clientes.cnpj_em_uso("11222333000181", ignorar_id=cliente_id) is False
+ assert db_clientes.cnpj_em_uso("99988877000166") is False
+
+
+def test_cnpj_em_uso_ignora_pontuacao_de_registros_legados(cliente_id):
+ # CLIENTE.CNPJ tem formatação inconsistente em registros legados (ver
+ # licenciamento-n8n/docs/ANALISE_CLIENTE_IPREAL.md) — cnpj_em_uso
+ # precisa achar o duplicado mesmo se o legado ainda tem máscara.
+ conn = get_connection()
+ try:
+ with conn.cursor() as cur:
+ cur.execute("UPDATE CLIENTE SET CNPJ = %s WHERE ID = %s", ("11.222.333/0001-81", cliente_id))
+ finally:
+ conn.close()
+
+ assert db_clientes.cnpj_em_uso("11222333000181") is True
+
+
def test_atualizar_cliente(cliente_id):
dados = dict(CLIENTE_TESTE)
dados["FANTASIA"] = "TESTE PAINEL FANTASIA EDITADA"
diff --git a/tests/test_router_clientes.py b/tests/test_router_clientes.py
index 68069ae..9b008d2 100644
--- a/tests/test_router_clientes.py
+++ b/tests/test_router_clientes.py
@@ -253,6 +253,7 @@ def test_criar_cliente_codigo_duplicado_bloqueia_sem_bypass(monkeypatch):
def test_criar_cliente_campo_obrigatorio_faltando_bloqueia(monkeypatch):
monkeypatch.setattr(router_clientes.db_clientes, "codigo_em_uso", lambda codigo, ignorar_id=None: False)
+ monkeypatch.setattr(router_clientes.db_clientes, "cnpj_em_uso", lambda documento, ignorar_id=None: False)
criado = {}
monkeypatch.setattr(
router_clientes.db_clientes, "criar_cliente", lambda dados: criado.setdefault("ok", True) and 42
@@ -276,6 +277,7 @@ def test_criar_cliente_campos_administrativos_em_branco_usa_defaults(monkeypatch
import datetime
monkeypatch.setattr(router_clientes.db_clientes, "codigo_em_uso", lambda codigo, ignorar_id=None: False)
+ monkeypatch.setattr(router_clientes.db_clientes, "cnpj_em_uso", lambda documento, ignorar_id=None: False)
capturado = {}
monkeypatch.setattr(
router_clientes.db_clientes, "criar_cliente", lambda dados: capturado.setdefault("dados", dados) and 42
@@ -299,6 +301,7 @@ def test_criar_cliente_campos_administrativos_em_branco_usa_defaults(monkeypatch
def test_criar_cliente_normaliza_cnpj_e_telefones(monkeypatch):
monkeypatch.setattr(router_clientes.db_clientes, "codigo_em_uso", lambda codigo, ignorar_id=None: False)
+ monkeypatch.setattr(router_clientes.db_clientes, "cnpj_em_uso", lambda documento, ignorar_id=None: False)
capturado = {}
monkeypatch.setattr(
router_clientes.db_clientes, "criar_cliente", lambda dados: capturado.setdefault("dados", dados) and 42
@@ -323,6 +326,7 @@ def test_criar_cliente_normaliza_cnpj_e_telefones(monkeypatch):
def test_editar_cliente_normaliza_cnpj_e_telefones(monkeypatch):
monkeypatch.setattr(router_clientes.db_clientes, "obter_cliente", lambda cliente_id: {"ID": cliente_id})
monkeypatch.setattr(router_clientes.db_clientes, "codigo_em_uso", lambda codigo, ignorar_id=None: False)
+ monkeypatch.setattr(router_clientes.db_clientes, "cnpj_em_uso", lambda documento, ignorar_id=None: False)
capturado = {}
monkeypatch.setattr(
router_clientes.db_clientes,
@@ -363,6 +367,37 @@ def test_editar_cliente_codigo_duplicado_bloqueia_sem_bypass(monkeypatch):
assert "ok" not in atualizado
+def test_criar_cliente_cnpj_duplicado_bloqueia(monkeypatch):
+ monkeypatch.setattr(router_clientes.db_clientes, "codigo_em_uso", lambda codigo, ignorar_id=None: False)
+ monkeypatch.setattr(router_clientes.db_clientes, "cnpj_em_uso", lambda documento, ignorar_id=None: True)
+ criado = {}
+ monkeypatch.setattr(
+ router_clientes.db_clientes, "criar_cliente", lambda dados: criado.setdefault("ok", True) and 42
+ )
+
+ resposta = client.post("/clientes/novo", data=_form_valido())
+ assert resposta.status_code == 200
+ assert "já está cadastrado" in resposta.text
+ assert "ok" not in criado
+
+
+def test_editar_cliente_cnpj_duplicado_bloqueia(monkeypatch):
+ monkeypatch.setattr(router_clientes.db_clientes, "obter_cliente", lambda cliente_id: {"ID": cliente_id})
+ monkeypatch.setattr(router_clientes.db_clientes, "codigo_em_uso", lambda codigo, ignorar_id=None: False)
+ monkeypatch.setattr(router_clientes.db_clientes, "cnpj_em_uso", lambda documento, ignorar_id=None: True)
+ atualizado = {}
+ monkeypatch.setattr(
+ router_clientes.db_clientes,
+ "atualizar_cliente",
+ lambda cliente_id, dados: atualizado.setdefault("ok", True),
+ )
+
+ resposta = client.post("/clientes/1/editar", data=_form_valido())
+ assert resposta.status_code == 200
+ assert "já está cadastrado" in resposta.text
+ assert "ok" not in atualizado
+
+
def test_editar_cliente_inexistente_nao_finge_sucesso(monkeypatch):
# Sem a checagem de existência, isso caía direto no UPDATE (0 linhas
# afetadas, sem erro do MySQL) e ainda registrava "sucesso" no log.
@@ -437,6 +472,7 @@ def test_criar_cliente_com_data_e_decimais_vazios_cria_com_defaults():
def test_criar_cliente_com_dv_invalido_exige_confirmacao(monkeypatch):
monkeypatch.setattr(router_clientes.db_clientes, "codigo_em_uso", lambda codigo, ignorar_id=None: False)
+ monkeypatch.setattr(router_clientes.db_clientes, "cnpj_em_uso", lambda documento, ignorar_id=None: False)
criado = {}
monkeypatch.setattr(
router_clientes.db_clientes, "criar_cliente", lambda dados: criado.setdefault("ok", True) and 42
@@ -464,6 +500,7 @@ def test_criar_cliente_com_cnpj_dv_invalido_bloqueia_sempre(monkeypatch):
"cadastrar mesmo assim" — nem com confirmar_dv_invalido=1.
"""
monkeypatch.setattr(router_clientes.db_clientes, "codigo_em_uso", lambda codigo, ignorar_id=None: False)
+ monkeypatch.setattr(router_clientes.db_clientes, "cnpj_em_uso", lambda documento, ignorar_id=None: False)
criado = {}
monkeypatch.setattr(
router_clientes.db_clientes, "criar_cliente", lambda dados: criado.setdefault("ok", True) and 42
@@ -487,6 +524,7 @@ def test_criar_cliente_com_cnpj_dv_invalido_bloqueia_sempre(monkeypatch):
def test_editar_cliente_com_cnpj_dv_invalido_bloqueia(monkeypatch):
monkeypatch.setattr(router_clientes.db_clientes, "obter_cliente", lambda cliente_id: {"ID": cliente_id})
monkeypatch.setattr(router_clientes.db_clientes, "codigo_em_uso", lambda codigo, ignorar_id=None: False)
+ monkeypatch.setattr(router_clientes.db_clientes, "cnpj_em_uso", lambda documento, ignorar_id=None: False)
atualizado = {}
monkeypatch.setattr(
router_clientes.db_clientes,
@@ -506,6 +544,7 @@ def test_editar_cliente_com_cnpj_dv_invalido_bloqueia(monkeypatch):
def test_criar_cliente_com_cpf_dv_invalido_continua_permitindo_confirmar(monkeypatch):
"""Regressão: CPF com DV inválido não muda de comportamento."""
monkeypatch.setattr(router_clientes.db_clientes, "codigo_em_uso", lambda codigo, ignorar_id=None: False)
+ monkeypatch.setattr(router_clientes.db_clientes, "cnpj_em_uso", lambda documento, ignorar_id=None: False)
criado = {}
monkeypatch.setattr(
router_clientes.db_clientes, "criar_cliente", lambda dados: criado.setdefault("ok", True) and 42
@@ -527,6 +566,7 @@ def test_criar_cliente_com_cpf_dv_invalido_continua_permitindo_confirmar(monkeyp
def test_criar_cliente_grava_telefone_com_ddi_55(monkeypatch):
monkeypatch.setattr(router_clientes.db_clientes, "codigo_em_uso", lambda codigo, ignorar_id=None: False)
+ monkeypatch.setattr(router_clientes.db_clientes, "cnpj_em_uso", lambda documento, ignorar_id=None: False)
capturado = {}
monkeypatch.setattr(
router_clientes.db_clientes, "criar_cliente", lambda dados: capturado.setdefault("dados", dados) and 42
@@ -543,6 +583,7 @@ def test_criar_cliente_grava_telefone_com_ddi_55(monkeypatch):
def test_criar_cliente_telefone_ja_com_55_nao_duplica(monkeypatch):
monkeypatch.setattr(router_clientes.db_clientes, "codigo_em_uso", lambda codigo, ignorar_id=None: False)
+ monkeypatch.setattr(router_clientes.db_clientes, "cnpj_em_uso", lambda documento, ignorar_id=None: False)
capturado = {}
monkeypatch.setattr(
router_clientes.db_clientes, "criar_cliente", lambda dados: capturado.setdefault("dados", dados) and 42
@@ -558,6 +599,7 @@ def test_criar_cliente_telefone_ja_com_55_nao_duplica(monkeypatch):
def test_criar_cliente_grava_uf_maiuscula_e_cep_sem_traco(monkeypatch):
monkeypatch.setattr(router_clientes.db_clientes, "codigo_em_uso", lambda codigo, ignorar_id=None: False)
+ monkeypatch.setattr(router_clientes.db_clientes, "cnpj_em_uso", lambda documento, ignorar_id=None: False)
capturado = {}
monkeypatch.setattr(
router_clientes.db_clientes, "criar_cliente", lambda dados: capturado.setdefault("dados", dados) and 42