PAIRUS Product Data SDK para Python 🐍

SDK oficial da PAIRUS Soluções Tecnológicas em Python moderno (3.10+) com tipagem estrita (Pydantic V2), suporte nativo a clientes síncronos e assíncronos (asyncio / httpx), retentativas automáticas com Exponential Backoff e tratamento amigável de erros no padrão fiscal SEFAZ (cStat e xMotivo).

---

⚡ Quickstart em 3 Linhas

from pairus_product_data import PairusProductData

client = PairusProductData(api_key="sua_chave_pairus")
produto = client.products.get("7891000100103")
print(f"{produto.xProd} | NCM: {produto.ncm} | CEST: {produto.cest}")

---

📦 Instalação

pip install pairus-product-data

---

🛠️ Inicialização e Configuração

Você pode inicializar o cliente informando a chave explicitamente ou definindo a variável de ambiente PAIRUS_API_KEY:

import os
from pairus_product_data import PairusProductData, AsyncPairusProductData

# Síncrono
client = PairusProductData(api_key="pk_live_...")

# Assíncrono com Context Manager
async with AsyncPairusProductData(api_key="pk_live_...") as async_client:
    produto = await async_client.products.get("7891000100103")

---

📚 Módulos e Casos de Uso Práticos

1. 🔍 Catálogo GTIN & Produtos

A. Consulta Básica (V1)

produto = client.products.get("7891000100103")
print(produto.xProd, produto.marca, produto.ncm, produto.cest)

B. Consulta Enriquecida por IA com SEO e Ficha Técnica (V2)

enriched = client.products.get_enriched("7891000100103")
print(enriched.descricao_completa)
print(enriched.ficha_tecnica) # Dimensões, peso, ingredientes, etc.
print(enriched.palavras_chave) # Otimizadas para e-commerce e busca

C. Busca Semântica por Intenção (Linguagem Natural)

Encontre produtos sem saber o código de barras ou o nome exato:

resultado = client.products.search("refrigerante zero açúcar lata")
for item in resultado.produtos:
    print(f"[{item.score:.2f}] {item.gtin} - {item.xProd} ({item.marca})")

D. Leitura de Código de Barras por Foto / Imagem (OCR)

# A partir de arquivo no disco
produto = client.products.scan("foto_rotulo.jpg")

# Ou diretamente a partir de bytes em memória
with open("rotulo.jpg", "rb") as f:
    produto = client.products.scan(f.read())

print(f"Detectado GTIN: {produto.gtin} - {produto.xProd}")

---

2. ⚖️ Predição Fiscal & Reforma Tributária (IBS / CBS / IS)

Obtenha o enquadramento tributário exato com regras anti-rejeição da SEFAZ e os novos tributos da Reforma Tributária (LC 214/2025).

A. Predição de Item Único

predicao = client.fiscal.predict(
    xProd="Refrigerante Coca-Cola 350ml",
    regime_tributario="simples_nacional", # 'simples_nacional', 'lucro_presumido' ou 'lucro_real'
    uf_origem="SP",
    uf_destino="RJ",
    finalidade="revenda", # 'revenda', 'consumo_final' ou 'industrializacao'
    destinatario_contribuinte=True
)

trib = predicao.dados_tributarios
print(f"CFOP: {trib.cfop} | CST/CSOSN: {trib.icms_cst}")
print(f"PIS: CST {trib.pis_cst} ({trib.pis_aliquota}%) | COFINS: CST {trib.cofins_cst}")
print(f"IBS Efetivo: {trib.ibscbs.ibs_aliquota_efetiva}% | CBS Efetiva: {trib.ibscbs.cbs_aliquota_efetiva}%")

B. Saneamento e Auditoria Fiscal em Lote

Valide centenas de produtos de uma vez, detectando NCMs extintos e atribuindo o CEST oficial:

resultado = client.fiscal.sanitize([
    {"ncm": "22021000", "descricao": "Refrigerante Cola"},
    {"ncm": "22030000", "descricao": "Cerveja Pilsen"},
])

for item in resultado["itens"]:
    print(f"NCM: {item['ncm']} | Válido: {item['ncm_valido']} | CEST Sugerido: {item['cest_sugerido']}")

---

3. 🧾 Emissão de NF-e e NFC-e com Autocura SEFAZ

O SDK suporta diferentes formatos de envio, do mais rápido e dinâmico ao mais estrito e tipado:

Forma 1: Modo Rápido (Via Dicionário - Zero Boilerplate)

nota = client.emissao.emitir_nfe(
    serie=1,
    natureza_operacao="Venda de Mercadorias",
    destinatario={
        "documento": "12345678000195",
        "razao_social": "Cliente Exemplo LTDA",
        "email": "financeiro@cliente.com.br",
        "logradouro": "Av. Paulista",
        "numero": "1000",
        "bairro": "Bela Vista",
        "codigo_municipio": "3550308",
        "nome_municipio": "São Paulo",
        "uf": "SP",
        "cep": "01310100",
    },
    itens=[
        {
            "descricao": "Mouse Sem Fio",
            "ncm": "84716053",
            "quantidade": 2,
            "valor_unitario": 50.0,
            "cfop": "5102",
        }
    ],
)

if nota.sucesso:
    print(f"✅ NF-e Autorizada! Chave: {nota.chave_acesso}")
    print(f"📄 Protocolo: {nota.protocolo_autorizacao}")
    print(f"🖨️ DANFE URL: {nota.danfe_url}")
else:
    print(f"❌ Rejeição SEFAZ [{nota.cStat}]: {nota.xMotivo}")

Forma 2: Modo Tipado / Enterprise (Com validação Pydantic)

from pairus_product_data.models.emissao import EmissaoNFeInput, DestinatarioInput, ItemEmissaoInput

payload = EmissaoNFeInput(
    serie=1,
    natureza_operacao="Venda de Mercadorias",
    destinatario=DestinatarioInput(
        documento="12345678000195",
        razao_social="Cliente Exemplo LTDA",
        uf="SP",
    ),
    itens=[
        ItemEmissaoInput(
            descricao="Teclado Mecânico",
            valor_unitario=250.0,
            quantidade=1.0,
            cfop="5102",
        )
    ],
)

nota = client.emissao.emitir_nfe(payload)

Forma 3: Simulação Fiscal Gratuita (Custo ZERO de créditos)

Valide seus cálculos e espelho de DANFE antes de transmitir:

simulacao = client.emissao.simular(
    natureza_operacao="Venda",
    destinatario={"documento": "12345678000195", "razao_social": "Cliente Teste", "uf": "SP"},
    itens=[{"descricao": "Item Teste", "valor_unitario": 100.0}],
)
print("Simulação aprovada:", simulacao.sucesso)

---

4. 🏛️ Emissão de NFS-e (Nota Fiscal de Serviços Eletrônica)

Compatível com o Padrão Nacional (SND) e provedores municipais (ABRASF v2).

nfse = client.nfse.emitir(
    prestador={
        "cnpj": "12345678000195",
        "inscricao_municipal": "123456",
        "razao_social": "Minha Empresa de Software LTDA",
    },
    tomador={
        "cpf_cnpj": "98765432000198",
        "razao_social": "Tomador de Serviços S.A.",
        "email": "contas@tomador.com.br",
    },
    servico={
        "item_lista_servico": "1.07", # Suporte técnico e consultoria em TI (LC 116/2003)
        "discriminacao": "Desenvolvimento de software e consultoria técnica especializada",
        "municipio_prestacao_ibge": "3550308", # São Paulo/SP
        "valor_servicos": 5000.0,
        "aliquota_iss": 2.0,
        "iss_retido": False,
        "retencoes_federais": {
            "pis_retido": True, "aliquota_pis": 0.65, "valor_pis": 32.50,
            "cofins_retido": True, "aliquota_cofins": 3.0, "valor_cofins": 150.0,
            "csll_retida": True, "aliquota_csll": 1.0, "valor_csll": 50.0,
            "irrf_retido": True, "aliquota_irrf": 1.5, "valor_irrf": 75.0,
        }
    }
)

if nfse.sucesso:
    print(f"✅ NFS-e Emitida! Número: {nfse.numero_nfse}")
    print(f"🔑 Chave Nacional: {nfse.chave_acesso_nacional}")
    print(f"💰 Valor Líquido: R$ {nfse.valor_liquido:.2f}")
    print(f"🖨️ Link DANFSE: {nfse.link_visualizacao}")

---

5. 🔄 Ciclo de Vida e Eventos Fiscais

Cancelamento de NF-e

evento = client.emissao.cancelar(
    chave_acesso="35260912345678000195550010000000451234567890",
    justificativa="Cancelamento por desacordo comercial formal entre as partes",
)
print(f"Cancelamento homologado: {evento.sucesso}")

Carta de Correção Eletrônica (CC-e)

cce = client.emissao.carta_correcao(
    chave_acesso="35260912345678000195550010000000451234567890",
    correcao="Correção do endereço de entrega: Rua das Flores, 123 - Bairro Jardim",
)
print(f"CC-e protocolada: {cce.sucesso}")

Inutilização de Faixa de Numeração Quebrada

inut = client.emissao.inutilizar(
    cnpj_emitente="12345678000195",
    serie=1,
    numero_inicial=100,
    numero_final=105,
    justificativa="Quebra de numeração por travamento de sistema emissor legado",
)
print(f"Inutilização homologada: {inut.sucesso}")

6. 📥 DF-e Inbound & Captura Ativa SEFAZ (Gestão de Compras)

Monitore e capture automaticamente as notas fiscais que fornecedores emitem contra o seu CNPJ via WebService NFeDistribuicaoDFe da SEFAZ Nacional. O SDK gerencia o controle incremental de NSU contra a Rejeição 656 (Consumo Indevido), descompacta pacotes docZip e oferece Manifestação do Destinatário completa.

A. Disparar Busca Ativa na SEFAZ Nacional

sync = client.dfe.sincronizar(cnpj="12345678000195", ambiente="producao")

print(f"Status SEFAZ [{sync.cstat}]: {sync.xmotivo}")
print(f"Novos documentos capturados: {sync.novos_documentos}")
print(f"Progresso do NSU: {sync.ult_nsu} -> {sync.max_nsu}")

for doc in sync.documentos:
    print(f"NF-e {doc.numero}/{doc.serie} - R$ {doc.valor_total:.2f} de {doc.nome_emitente}")

B. Listar Documentos Fiscais de Compras

compras = client.dfe.listar_documentos(cnpj="12345678000195", limite=50)

for doc in compras.documentos:
    print(f"Chave: {doc.chave_acesso} | Status: {doc.manifestacao_status} | Tem XML Completo: {doc.tem_xml_completo}")

C. Manifestação do Destinatário perante a SEFAZ

Permite registrar oficialmente a posição da empresa sobre a nota emitida:

from pairus_product_data import TipoManifestacaoDFe

# 1. Ciência da Emissão (Libera o download do XML completo na SEFAZ)
manif = client.dfe.manifestar(
    chave_acesso="35260912345678000195550010000000451234567890",
    cnpj="12345678000195",
    tipo_evento=TipoManifestacaoDFe.CIENCIA_DA_EMISSAO, # '210210'
)
print(f"Ciência homologada com protocolo: {manif.protocolo}")

# 2. Confirmação da Operação (Atesta recebimento da mercadoria)
client.dfe.manifestar(
    chave_acesso="35260912345678000195550010000000451234567890",
    cnpj="12345678000195",
    tipo_evento=TipoManifestacaoDFe.CONFIRMACAO_DA_OPERACAO, # '210200'
)

# 3. Operação Não Realizada (Exige justificativa mínima de 15 caracteres)
client.dfe.manifestar(
    chave_acesso="35260912345678000195550010000000451234567890",
    cnpj="12345678000195",
    tipo_evento=TipoManifestacaoDFe.OPERACAO_NAO_REALIZADA, # '210240'
    justificativa="Mercadoria avariada durante o transporte e devolvida integralmente",
)

D. Download de XML Autorizado e DANFE em PDF

chave = "35260912345678000195550010000000451234567890"

# Baixa o XML completo autorizado (procNFe)
xml_string = client.dfe.baixar_xml(chave)
with open(f"{chave}.xml", "w", encoding="utf-8") as f:
    f.write(xml_string)

# Baixa o DANFE em PDF
danfe_pdf_bytes = client.dfe.baixar_danfe(chave)
with open(f"DANFE_{chave}.pdf", "wb") as f:
    f.write(danfe_pdf_bytes)

---

7. 🔐 Webhooks Seguros (HMAC-SHA256)

Valide a assinatura do cabeçalho X-Pairus-Signature contra ataques de repetição:

from pairus_product_data import Webhooks

payload_bruto = request.body # Bytes brutos recebidos no webhook
assinatura = request.headers.get("X-Pairus-Signature")
webhook_secret = "whsec_..."

try:
    evento = Webhooks.construct_event(payload_bruto, assinatura, webhook_secret)
    print(f"Evento legítimo recebido: {evento.event} na data {evento.timestamp}")
except Exception as e:
    print(f"Assinatura inválida! Rejeitar requisição: {e}")

---

📋 Tabela de Parâmetros (Obrigatórios vs. Opcionais)

📌 Emissão de NF-e / NFC-e (client.emissao.emitir_nfe)

Parâmetro	Tipo	Obrigatoriedade	Padrão	Descrição & Regras Fiscais
destinatario.documento	str	Obrigatório	-	CPF (11 dígitos) ou CNPJ (14 dígitos) limpos.
destinatario.razao_social	str	Obrigatório	-	Razão Social ou Nome completo do comprador.
destinatario.uf	str	Obrigatório	-	Sigla da UF do comprador (2 letras).
itens	list	Obrigatório	-	Lista contendo ao menos 1 item comercial.
itens[].descricao	str	Obrigatório	-	Descrição do produto na NF-e.
itens[].valor_unitario	float	Obrigatório	-	Preço unitário maior que zero.
itens[].quantidade	float	Opcional	1.0	Quantidade comercializada.
itens[].ncm	str	Opcional	Auto	Código NCM de 8 dígitos (enquadrado por IA se omitido).
itens[].cfop	str	Opcional	Auto	CFOP da operação (determinado automaticamente se omitido).
natureza_operacao	str	Opcional	Venda	Texto da natureza da operação.
serie	int	Opcional	1	Série da nota fiscal (1 a 999).

📌 Emissão de NFS-e (client.nfse.emitir)

Parâmetro	Tipo	Obrigatoriedade	Padrão	Descrição & Regras Fiscais
prestador.cnpj	str	Obrigatório	-	CNPJ da empresa prestadora (14 dígitos).
prestador.inscricao_municipal	str	Obrigatório	-	Inscrição Municipal na prefeitura.
prestador.razao_social	str	Obrigatório	-	Razão Social da prestadora.
tomador.cpf_cnpj	str	Obrigatório	-	Documento do tomador (CPF 11 ou CNPJ 14).
tomador.razao_social	str	Obrigatório	-	Nome ou Razão Social do tomador.
servico.item_lista_servico	str	Obrigatório	-	Subitem da LC 116/2003 (ex: "1.07", "17.01").
servico.discriminacao	str	Obrigatório	-	Descrição do serviço (mínimo 5 caracteres).
servico.municipio_prestacao_ibge	str	Obrigatório	-	Código IBGE do local do serviço (7 dígitos).
servico.valor_servicos	float	Obrigatório	-	Valor bruto do serviço prestado (R$).
servico.aliquota_iss	float	Opcional	2.0	Alíquota de ISS entre 0.0% e 5.0%.
servico.iss_retido	bool	Opcional	False	True se o ISS é retido na fonte pelo tomador.
modo	str	Opcional	direto	'direto' (pass-through) ou 'assistido' (IA).
ambiente	str	Opcional	producao	'producao' ou 'homologacao'.

📌 Manifestação do Destinatário (client.dfe.manifestar)

Parâmetro	Tipo	Obrigatoriedade	Padrão	Descrição & Regras Fiscais
chave_acesso	str	Obrigatório	-	Chave de acesso de 44 dígitos da NF-e emitida pelo fornecedor.
cnpj	str	Obrigatório	-	CNPJ da sua empresa (destinatária da nota).
tipo_evento	str / Enum	Obrigatório	-	'210210' (Ciência), '210200' (Confirmação), '210220' (Desconhecimento) ou '210240' (Não Realizada).
justificativa	str	Condicional	None	Obrigatória exclusivamente para '210240' (mínimo de 15 caracteres).
ambiente	str	Opcional	producao	'producao' ou 'homologacao'.

---

🛡️ Tratamento de Erros e Exceções

O SDK sanitiza todas as falhas de rede e da SEFAZ em classes de erro claras:

from pairus_product_data import (
    PairusAPIError,
    AuthenticationError,
    RateLimitError,
    NetworkError,
)

try:
    client.emissao.emitir_nfe(...)
except AuthenticationError as e:
    print(f"Chave de API inválida: {e.xMotivo}")
except RateLimitError as e:
    print(f"Rate limit atingido. Aguarde {e.retry_after}s")
except PairusAPIError as e:
    print(f"Erro fiscal [{e.cStat}]: {e.xMotivo}")
except NetworkError as e:
    print(f"Falha de conectividade após retentativas: {e}")

---

📄 Licença

Distribuído sob a licença MIT. Consulte o arquivo LICENSE para obter mais informações.