Como preencher o cadastro pelo CNPJ: endereço automático no formulário

import { createServer } from 'node:http';
import { readFile } from 'node:fs/promises';

const API = (process.env.APIPJ_BASE_URL ?? 'https://apipj.com.br').replace(/\/+$/, '');
const CHAVE = process.env.APIPJ_KEY; // só no servidor: nunca no HTML nem no JavaScript do navegador
const PORTA = Number(process.env.PORT ?? 3000);
const DIA = 24 * 60 * 60 * 1000;
const PESOS = [6, 5, 4, 3, 2, 9, 8, 7, 6, 5, 4, 3, 2];

/** Formato e dígitos verificadores, numérico ou alfanumérico (valor = código ASCII - 48). */
function cnpjValido(cnpj) {
  if (!/^[A-Z0-9]{12}[0-9]{2}$/.test(cnpj) || /^0+$/.test(cnpj)) return false;
  const v = [...cnpj].map((ch) => ch.charCodeAt(0) - 48);
  const dv = (n) => {
    let soma = 0;
    for (let i = 0; i < n; i++) soma += v[i] * PESOS[i + 13 - n];
    const resto = soma % 11;
    return resto < 2 ? 0 : 11 - resto;
  };
  return v[12] === dv(12) && v[13] === dv(13);
}

/** Só o que o formulário usa; o resto da resposta (sócios, CNAEs...) não sai do servidor. */
function paraFormulario(e) {
  const end = e.endereco;
  const tel = e.telefones[0];
  return {
    cnpj: e.cnpj,
    razao_social: e.razao_social,
    nome_fantasia: e.nome_fantasia ?? '',
    logradouro: [end.tipo_logradouro, end.logradouro].filter(Boolean).join(' '),
    numero: end.numero ?? '',
    complemento: end.complemento ?? '',
    bairro: end.bairro ?? '',
    cep: (end.cep ?? '').replace(/^(\d{5})(\d{3})$/, '$1-$2'),
    cidade: end.municipio?.nome ?? '',
    uf: end.uf ?? '',
    codigo_ibge: end.municipio?.codigo_ibge ?? '',
    telefone: tel ? `(${tel.ddd}) ${tel.numero.slice(0, -4)}-${tel.numero.slice(-4)}` : '',
    email: (e.email ?? '').toLowerCase(),
    situacao: e.situacao_cadastral.descricao,
  };
}

const cache = new Map(); // cnpj -> { expira, resposta }

async function consultar(cnpj) {
  const guardado = cache.get(cnpj);
  if (guardado && guardado.expira > Date.now()) return guardado.resposta;
  const r = await fetch(`${API}/v1/cnpj/${cnpj}?formato=completo`, {
    headers: { Authorization: `Bearer ${CHAVE}` },
    signal: AbortSignal.timeout(8000),
  });
  let resposta;
  if (r.status === 200) resposta = { status: 200, corpo: paraFormulario(await r.json()) };
  else if (r.status === 404) resposta = { status: 404, corpo: { erro: 'CNPJ não encontrado na base da Receita Federal' } };
  else if (r.status === 429) {
    return { status: 429, corpo: { erro: 'Limite de consultas atingido; tente de novo em instantes' }, retryAfter: r.headers.get('retry-after') };
  } else if (r.status === 400) return { status: 400, corpo: { erro: 'CNPJ inválido' } };
  else throw new Error(`apipj respondeu HTTP ${r.status}`); // 401 (chave), 5xx...
  // Encontrado: 30 dias (a base é mensal). Não encontrado: 1 dia, para não segurar uma empresa recém-aberta.
  if (cache.size >= 10_000) cache.delete(cache.keys().next().value);
  cache.set(cnpj, { expira: Date.now() + (resposta.status === 200 ? 30 * DIA : DIA), resposta });
  return resposta;
}

function json(res, status, corpo, extras = {}) {
  res.writeHead(status, { 'Content-Type': 'application/json; charset=utf-8', ...extras });
  res.end(JSON.stringify(corpo));
}

const ESTATICOS = {
  '/': ['cadastro.html', 'text/html; charset=utf-8'],
  '/cadastro.js': ['cadastro.js', 'text/javascript; charset=utf-8'],
};

createServer(async (req, res) => {
  const { pathname } = new URL(req.url, 'http://localhost');
  const rota = pathname.match(/^\/api\/cnpj\/(.*)$/);
  if (req.method === 'GET' && rota) {
    const cnpj = rota[1].replace(/[.-]/g, '').toUpperCase();
    if (!cnpjValido(cnpj)) return json(res, 400, { erro: 'CNPJ inválido' }); // nem chega à API
    try {
      const r = await consultar(cnpj);
      return json(res, r.status, r.corpo, r.retryAfter ? { 'Retry-After': r.retryAfter } : {});
    } catch (e) {
      console.error('apipj:', e.message);
      return json(res, 502, { erro: 'Consulta de CNPJ indisponível' });
    }
  }
  const estatico = ESTATICOS[pathname];
  if (req.method === 'GET' && estatico) {
    res.writeHead(200, { 'Content-Type': estatico[1] });
    return res.end(await readFile(new URL(`./public/${estatico[0]}`, import.meta.url)));
  }
  json(res, 404, { erro: 'Rota não encontrada' });
}).listen(PORTA, () => console.log(`Formulário em http://localhost:${PORTA}`));

import json
import os
import re
import threading
import time
import urllib.error
import urllib.request
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
from pathlib import Path

API = os.environ.get("APIPJ_BASE_URL", "https://apipj.com.br").rstrip("/")
CHAVE = os.environ["APIPJ_KEY"]  # só no servidor: nunca no HTML nem no JavaScript do navegador
PORTA = int(os.environ.get("PORT", "8000"))
PUBLICO = Path(__file__).parent / "public"
DIA = 24 * 60 * 60
PESOS = [6, 5, 4, 3, 2, 9, 8, 7, 6, 5, 4, 3, 2]
FORMATO = re.compile(r"[A-Z0-9]{12}[0-9]{2}")


def cnpj_valido(cnpj):
    """Formato e dígitos verificadores, numérico ou alfanumérico (valor = código ASCII - 48)."""
    if not FORMATO.fullmatch(cnpj) or cnpj == "0" * 14:
        return False
    v = [ord(ch) - 48 for ch in cnpj]

    def dv(n):
        resto = sum(v[i] * PESOS[i + 13 - n] for i in range(n)) % 11
        return 0 if resto < 2 else 11 - resto

    return v[12] == dv(12) and v[13] == dv(13)


def para_formulario(e):
    """Só o que o formulário usa; o resto da resposta (sócios, CNAEs...) não sai do servidor."""
    end = e["endereco"]
    municipio = end.get("municipio") or {}
    tel = e["telefones"][0] if e["telefones"] else None
    cep = end.get("cep") or ""
    return {
        "cnpj": e["cnpj"],
        "razao_social": e["razao_social"],
        "nome_fantasia": e.get("nome_fantasia") or "",
        "logradouro": " ".join(p for p in (end.get("tipo_logradouro"), end.get("logradouro")) if p),
        "numero": end.get("numero") or "",
        "complemento": end.get("complemento") or "",
        "bairro": end.get("bairro") or "",
        "cep": f"{cep[:5]}-{cep[5:]}" if len(cep) == 8 else cep,
        "cidade": municipio.get("nome") or "",
        "uf": end.get("uf") or "",
        "codigo_ibge": municipio.get("codigo_ibge") or "",
        "telefone": f"({tel['ddd']}) {tel['numero'][:-4]}-{tel['numero'][-4:]}" if tel else "",
        "email": (e.get("email") or "").lower(),
        "situacao": e["situacao_cadastral"]["descricao"],
    }


_cache = {}  # cnpj -> (expira, resposta)
_trava = threading.Lock()


def consultar(cnpj):
    with _trava:
        guardado = _cache.get(cnpj)
    if guardado and guardado[0] > time.time():
        return guardado[1]
    pedido = urllib.request.Request(
        f"{API}/v1/cnpj/{cnpj}?formato=completo", headers={"Authorization": f"Bearer {CHAVE}"}
    )
    try:
        with urllib.request.urlopen(pedido, timeout=8) as r:
            resposta = (200, para_formulario(json.load(r)), None)
    except urllib.error.HTTPError as e:
        if e.code == 404:
            resposta = (404, {"erro": "CNPJ não encontrado na base da Receita Federal"}, None)
        elif e.code == 429:
            return (429, {"erro": "Limite de consultas atingido; tente de novo em instantes"}, e.headers.get("Retry-After"))
        elif e.code == 400:
            return (400, {"erro": "CNPJ inválido"}, None)
        else:
            raise  # 401 (chave), 5xx...
    # Encontrado: 30 dias (a base é mensal). Não encontrado: 1 dia, para não segurar uma empresa recém-aberta.
    with _trava:
        if len(_cache) >= 10_000:
            _cache.pop(next(iter(_cache)))
        _cache[cnpj] = (time.time() + (30 * DIA if resposta[0] == 200 else DIA), resposta)
    return resposta


ESTATICOS = {
    "/": ("cadastro.html", "text/html; charset=utf-8"),
    "/cadastro.js": ("cadastro.js", "text/javascript; charset=utf-8"),
}


class Rotas(BaseHTTPRequestHandler):
    def do_GET(self):
        caminho = self.path.split("?")[0]
        if caminho.startswith("/api/cnpj/"):
            cnpj = re.sub(r"[.-]", "", caminho[len("/api/cnpj/"):]).upper()
            if not cnpj_valido(cnpj):
                return self.enviar(400, {"erro": "CNPJ inválido"})  # nem chega à API
            try:
                status, corpo, retry_after = consultar(cnpj)
            except Exception as e:
                self.log_error("apipj: %s", e)
                return self.enviar(502, {"erro": "Consulta de CNPJ indisponível"})
            return self.enviar(status, corpo, retry_after)
        if caminho in ESTATICOS:
            arquivo, tipo = ESTATICOS[caminho]
            return self.enviar(200, (PUBLICO / arquivo).read_bytes(), tipo=tipo)
        self.enviar(404, {"erro": "Rota não encontrada"})

    def enviar(self, status, corpo, retry_after=None, tipo="application/json; charset=utf-8"):
        dados = corpo if isinstance(corpo, bytes) else json.dumps(corpo, ensure_ascii=False).encode()
        self.send_response(status)
        self.send_header("Content-Type", tipo)
        self.send_header("Content-Length", str(len(dados)))
        if retry_after:
            self.send_header("Retry-After", retry_after)
        self.end_headers()
        self.wfile.write(dados)


if __name__ == "__main__":
    print(f"Formulário em http://localhost:{PORTA}")
    ThreadingHTTPServer(("", PORTA), Rotas).serve_forever()