diff --git a/app/cnpj_lookup.py b/app/cnpj_lookup.py new file mode 100644 index 0000000..b28920e --- /dev/null +++ b/app/cnpj_lookup.py @@ -0,0 +1,57 @@ +""" +Consulta de CNPJ na BrasilAPI (https://brasilapi.com.br/api/cnpj/v1) para +preencher automaticamente o formulário de cliente. Fonte pública/gratuita, +sem autenticação, mantida pela comunidade a partir de dados abertos da +Receita Federal — não é a API oficial da Receita (que não tem consulta +pública direta por CNPJ). + +Só faz sentido para CNPJ (empresa) — CPF é dado pessoal e não tem consulta +pública, então o chamador nunca deve passar um CPF aqui (ver +app/routers/clientes.py, que checa o tipo via normalizar_documento antes). + +Degrada de forma segura: qualquer falha (CNPJ não encontrado, rede fora do +ar, formato que a BrasilAPI ainda não reconhece — o novo CNPJ alfanumérico, +vigente desde 31/07/2026, pode não estar coberto por todo provedor ainda) +retorna None. O formulário simplesmente não preenche nada sozinho nesse caso. +""" +import httpx + +_BASE_URL = "https://brasilapi.com.br/api/cnpj/v1" + + +def consultar_cnpj(cnpj_normalizado: str) -> dict | None: + """ + cnpj_normalizado: 14 caracteres (12 alfanuméricos + 2 dígitos), já sem + máscara — ver app/normalize.py. Retorna um dict só com os campos que + existem em CLIENTE, prontos para preencher o formulário, ou None se não + encontrou/a consulta falhou. + """ + if len(cnpj_normalizado) != 14: + return None + + try: + resposta = httpx.get(f"{_BASE_URL}/{cnpj_normalizado}", timeout=8.0) + if resposta.status_code != 200: + return None + dados = resposta.json() + except (httpx.HTTPError, ValueError): + return None + + if not isinstance(dados, dict): + return None + + logradouro = dados.get("logradouro") or "" + numero = dados.get("numero") or "" + endereco = f"{logradouro}, {numero}" if numero else logradouro + + return { + "RAZAO": dados.get("razao_social") or "", + "FANTASIA": dados.get("nome_fantasia") or dados.get("razao_social") or "", + "ENDERECO": endereco[:45], # CLIENTE.ENDERECO é varchar(45) + "CIDADE": dados.get("municipio") or "", + "BAIRRO": dados.get("bairro") or "", + "UF": dados.get("uf") or "", + "CEP": (dados.get("cep") or "")[:10], + "TELEFONE": dados.get("ddd_telefone_1") or "", + "EMAIL": dados.get("email") or "", + } diff --git a/app/routers/clientes.py b/app/routers/clientes.py index 2e39b90..fe37e01 100644 --- a/app/routers/clientes.py +++ b/app/routers/clientes.py @@ -10,9 +10,10 @@ from urllib.parse import urlencode import pymysql from fastapi import APIRouter, Depends, Form, Request -from fastapi.responses import HTMLResponse, RedirectResponse +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 @@ -173,6 +174,24 @@ async def criar(request: Request, usuario: str = Depends(exigir_login)): 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, diff --git a/app/templates/cliente_form.html b/app/templates/cliente_form.html index 5c6701b..014f932 100644 --- a/app/templates/cliente_form.html +++ b/app/templates/cliente_form.html @@ -34,9 +34,63 @@ {% else %} + {% if campo == "CNPJ" %} + + {% endif %} {% endif %} {% endfor %}
* campos obrigatórios. Os demais podem ficar em branco — implantação/mensalidade/juros ficam 0,00 e as datas de contrato ficam com a data de hoje até serem ajustadas.
+ + {% endblock %} diff --git a/tests/test_cnpj_lookup.py b/tests/test_cnpj_lookup.py new file mode 100644 index 0000000..a366329 --- /dev/null +++ b/tests/test_cnpj_lookup.py @@ -0,0 +1,75 @@ +import httpx + +from app import cnpj_lookup + + +def test_documento_com_tamanho_errado_nao_consulta(monkeypatch): + chamou = {} + monkeypatch.setattr(cnpj_lookup.httpx, "get", lambda *a, **k: chamou.setdefault("ok", True)) + assert cnpj_lookup.consultar_cnpj("123") is None + assert "ok" not in chamou + + +def test_consulta_com_sucesso_mapeia_campos(monkeypatch): + class _RespostaFalsa: + status_code = 200 + + def json(self): + return { + "razao_social": "EMPRESA TESTE LTDA", + "nome_fantasia": "EMPRESA TESTE", + "logradouro": "RUA TESTE", + "numero": "123", + "bairro": "CENTRO", + "municipio": "FEIRA DE SANTANA", + "uf": "BA", + "cep": "44000000", + "ddd_telefone_1": "7533334444", + "email": "contato@empresateste.com", + } + + monkeypatch.setattr(cnpj_lookup.httpx, "get", lambda url, timeout=None: _RespostaFalsa()) + + resultado = cnpj_lookup.consultar_cnpj("11222333000181") + assert resultado == { + "RAZAO": "EMPRESA TESTE LTDA", + "FANTASIA": "EMPRESA TESTE", + "ENDERECO": "RUA TESTE, 123", + "CIDADE": "FEIRA DE SANTANA", + "BAIRRO": "CENTRO", + "UF": "BA", + "CEP": "44000000", + "TELEFONE": "7533334444", + "EMAIL": "contato@empresateste.com", + } + + +def test_consulta_nao_encontrada_retorna_none(monkeypatch): + class _RespostaFalsa: + status_code = 404 + + def json(self): + return {} + + monkeypatch.setattr(cnpj_lookup.httpx, "get", lambda url, timeout=None: _RespostaFalsa()) + assert cnpj_lookup.consultar_cnpj("11222333000181") is None + + +def test_falha_de_rede_retorna_none(monkeypatch): + def _get_com_erro(url, timeout=None): + raise httpx.ConnectError("falhou") + + monkeypatch.setattr(cnpj_lookup.httpx, "get", _get_com_erro) + assert cnpj_lookup.consultar_cnpj("11222333000181") is None + + +def test_sem_nome_fantasia_usa_razao_social(monkeypatch): + class _RespostaFalsa: + status_code = 200 + + def json(self): + return {"razao_social": "EMPRESA TESTE LTDA", "nome_fantasia": None} + + monkeypatch.setattr(cnpj_lookup.httpx, "get", lambda url, timeout=None: _RespostaFalsa()) + resultado = cnpj_lookup.consultar_cnpj("11222333000181") + assert resultado["FANTASIA"] == "EMPRESA TESTE LTDA" diff --git a/tests/test_router_clientes.py b/tests/test_router_clientes.py index 3681d3f..40dad1e 100644 --- a/tests/test_router_clientes.py +++ b/tests/test_router_clientes.py @@ -140,6 +140,48 @@ def test_listar_clientes_direcao_invalida_cai_no_padrao(monkeypatch): assert capturado["direcao"] == "desc" +def test_consulta_cnpj_encontrado(monkeypatch): + monkeypatch.setattr( + router_clientes.cnpj_lookup, + "consultar_cnpj", + lambda cnpj: {"RAZAO": "EMPRESA TESTE LTDA", "FANTASIA": "EMPRESA TESTE"}, + ) + resposta = client.get("/clientes/consulta-cnpj/11222333000181") + assert resposta.status_code == 200 + corpo = resposta.json() + assert corpo["encontrado"] is True + assert corpo["dados"]["RAZAO"] == "EMPRESA TESTE LTDA" + + +def test_consulta_cnpj_nao_encontrado(monkeypatch): + monkeypatch.setattr(router_clientes.cnpj_lookup, "consultar_cnpj", lambda cnpj: None) + resposta = client.get("/clientes/consulta-cnpj/11222333000181") + assert resposta.status_code == 404 + assert resposta.json()["encontrado"] is False + + +def test_consulta_cnpj_rejeita_cpf(monkeypatch): + chamou = {} + monkeypatch.setattr( + router_clientes.cnpj_lookup, "consultar_cnpj", lambda cnpj: chamou.setdefault("ok", True) + ) + # CPF válido (mesma fixture usada em test_normalize.py) — nunca deve + # chamar a consulta de CNPJ, é dado pessoal. + resposta = client.get("/clientes/consulta-cnpj/11144477735") + assert resposta.status_code == 404 + assert "ok" not in chamou + + +def test_consulta_cnpj_rejeita_formato_invalido(monkeypatch): + chamou = {} + monkeypatch.setattr( + router_clientes.cnpj_lookup, "consultar_cnpj", lambda cnpj: chamou.setdefault("ok", True) + ) + resposta = client.get("/clientes/consulta-cnpj/123") + assert resposta.status_code == 404 + assert "ok" not in chamou + + def test_form_novo_preenche_codigo_sugerido(monkeypatch): monkeypatch.setattr(router_clientes.db_clientes, "proximo_codigo_sugerido", lambda: "000243") resposta = client.get("/clientes/novo")