Guía de Integración Rápida con Python
El SDK de Python está implementado como una utilidad interactiva por consola (CLI) ideal para integrarse en scripts de backend, CRONs de orquestación en Linux, o como base para adaptarlo a tu propio proxy en Django/Flask/FastAPI.
🛠️ Entorno: Instalación y Ejecución
Para correr este motor, es imperativo disponer de un entorno de ejecución Python accesible (versión 3.7 o superior recomendada).
- Instalar Python: Si no lo tienes, descárgalo e instálalo desde python.org asegurándote de marcar "Add Python to PATH" durante la instalación (en Windows) o usando tu gestor de paquetes favorito (
apt install python3en Linux). - Ubícate en la carpeta donde has descomprimido estos ficheros de ejemplo.
- Instala las dependencias HTTP (solo usamos
requests):bashpip install -r requirements.txt - Arranca el motor interactivo:bash(Nota: en Linux o macOS podrías necesitar correr
python verifactu_cli.pypython3 verifactu_cli.py)
WARNING
Aviso sobre el bloque metadata (JSON de Facturas) En algunos JSON de prueba locales verás un bloque raíz llamado "metadata": { "enviar_aeat": false, "simulacion": true, ... }. La API funciona perfectamente CON y SIN ese bloque. Se introdujo exclusivamente por compatibilidad con futuras versiones, pero se aconseja encarecidamente que prescindas totalmente de él para los procesos y desarrollos actuales, enviando únicamente "cabecera" y "detalle".
Vídeo Demostrativo
TIP
Contexto de la demostración: Presta atención a la pantalla dividida. En la mitad de la izquierda se muestra el panel de administración central del Micro Server. En la mitad derecha podrás ver una ventana de terminal ejecutando el cliente interactivo. Verás en tiempo real cómo el cliente solicita ingestas y acuses de recibo (ACK) en masa, mientras el Micro Server reacciona procesándolo paralelamente validando con la AEAT.
🏗️ 1. Configuración del Contexto (Config.json)
Hemos delegado los ajustes estáticos a Config.json para no empotrar tu Token y URL dentro del script.
json
{
"api": {
"base_url": "http://localhost:8000",
"timeout_sec": 60,
"token": "TXT-VFACTU-TU-TOKEN",
"ssl_verify": false
},
"demo": {
"nif_emisor": "B12345678",
"serie": "TEST",
"num_inicio": "1"
}
}🚀 2. Ingesta a la API de Alta Velocidad
En Python todo es síncrono por defecto, pero la librería requests te permite despachar peticiones a la pasarela de forma fiable y capturar el JSON en una sola línea de código sin preocuparte de cabeceras HTTP.
python
# Así de simple es envolver el objeto de tu factura
factura = {
"cabecera": {
"emisor": "B12345678",
# ... demás campos ...
}
}
# Instancias el SDK con tu Config.json
engine = VFEngine("http://localhost:8000", "MI-TOKEN", "B12345678", False, 60)
# El envío devuelve un diccionario Python puro automáticamente, parseado por el motor
res = engine.ingesta_json(factura)
if res.get("ok"):
print("¡Factura subida y procesando en la cola de la AEAT!")📦 Código Fuente del SDK completo
Aquí exponemos de forma cristalina la tripas de la integración en Python. Puedes copiar y pegar todo este ecosistema en tu propia carpeta /scripts/ de tu backend o ERP y orquestarlo en 5 minutos.
Config.json
Configuración centralizada de credenciales y URL.
Ver/Ocultar Código Fuente (Config.json)
python
{
"api": {
"base_url": "http://localhost:8000",
"timeout_sec": 60,
"token": "vf_sys_fgh_BEwEMUFqXWEfPOBKcY9INWdmaRFoAhLV",
"ssl_verify": false
},
"demo": {
"nif_emisor": "B12345678",
"serie": "",
"num_inicio": ""
}
}verifactu_cli.py
Clase nativa con requests para gestionar las comunicaciones B2B.
Ver/Ocultar Código Fuente (verifactu_cli.py)
python
import sys
import json
import time
from datetime import datetime
try:
import requests
except ImportError:
print("❌ Error: falta la librería 'requests'.")
print("Por favor, instala la dependencia ejecutando:")
print(" pip install requests")
sys.exit(1)
# =========================================================
# CLIENTE CONSOLA - VERIFACTU (PYTHON B2B PORT DE .NET)
# =========================================================
class VFEngine:
def __init__(self, base_url, token, nif_emisor, ssl_verify, timeout):
self.base_url = base_url.rstrip("/")
if not self.base_url.startswith("http"):
self.base_url = "http://" + self.base_url
self.token = token
self.nif_emisor = nif_emisor
self.ssl_verify = ssl_verify
self.timeout = timeout
self.headers = {
"Accept": "application/json",
"Content-Type": "application/json"
}
if self.token:
self.headers["X-API-Key"] = self.token
self.set_emisor_header(self.nif_emisor)
def set_emisor_header(self, nif):
if nif:
self.nif_emisor = nif
self.headers["X-Verifactu-Emisor"] = nif
def parse_response(self, response):
try:
return response.json()
except:
return {
"ok": False,
"status": "error",
"mensaje": f"No se pudo parsear el JSON de respuesta. Error {response.status_code}.",
"raw": response.text[:200]
}
def ingesta_json(self, json_data):
cab = json_data.get("cabecera", {})
nif = cab.get("emisor") or cab.get("nif_emisor")
if nif:
self.set_emisor_header(nif)
url = f"{self.base_url}/v1/ingesta"
try:
resp = requests.post(url, json=json_data, headers=self.headers, verify=self.ssl_verify, timeout=self.timeout)
return self.parse_response(resp)
except Exception as e:
return {"ok": False, "status": "error", "mensaje": str(e)}
def check_status(self, emisor, serie, num):
self.set_emisor_header(emisor)
url = f"{self.base_url}/v1/check_status"
params = {"emisor": emisor, "serie": serie, "num": num}
try:
resp = requests.get(url, params=params, headers=self.headers, verify=self.ssl_verify, timeout=self.timeout)
return self.parse_response(resp)
except Exception as e:
return {"ok": False, "status": "error", "mensaje": str(e)}
def get_pendientes(self, emisor, limit=50):
self.set_emisor_header(emisor)
url = f"{self.base_url}/verifactu/pendientes"
params = {"nif_emisor": emisor, "n_ultimos": limit}
try:
resp = requests.get(url, params=params, headers=self.headers, verify=self.ssl_verify, timeout=self.timeout)
return self.parse_response(resp)
except Exception as e:
return {"ok": False, "status": "error", "mensaje": str(e)}
def ack_indice(self, emisor, indice_log):
self.set_emisor_header(emisor)
url = f"{self.base_url}/verifactu/ack"
data = {"nif_emisor": emisor, "indice_log": int(indice_log)}
try:
resp = requests.post(url, json=data, headers=self.headers, verify=self.ssl_verify, timeout=self.timeout)
return self.parse_response(resp)
except Exception as e:
return {"ok": False, "status": "error", "mensaje": str(e)}
# =========================================================
class Program:
def __init__(self):
self.engine = None
self.nif = ""
self.serie = ""
self.num = ""
def info(self, msg):
print(f"[*] {msg}")
def success(self, msg):
print(f"\033[92m{msg}\033[0m")
def error(self, msg):
print(f"\033[91m{msg}\033[0m")
def main(self):
print("========================================")
print(" VERIFACTU MICRO SERVER - PYTHON SDK")
print("========================================")
config_path = "Config.json"
try:
with open(config_path, "r", encoding="utf-8") as f:
config = json.load(f)
except FileNotFoundError:
self.error(f"Fichero {config_path} no encontrado.")
return
except Exception as e:
self.error(f"Error cargando configuración: {e}")
return
api = config.get("api", {})
demo = config.get("demo", {})
base_url = api.get("base_url", "http://localhost:8000")
token = api.get("token", "")
timeout = api.get("timeout_sec", 60)
ssl_verify = api.get("ssl_verify", False)
self.nif = demo.get("nif_emisor", "")
self.serie = demo.get("serie", "")
self.num = demo.get("num_inicio", "")
self.engine = VFEngine(base_url, token, self.nif, ssl_verify, timeout)
self.success(f"[*] API Conectada URL: {base_url}")
self.sync_with_json_template()
while True:
print("\n================= MENÚ =================")
print(f" Emisor: {self.nif} | Serie: {self.serie} | Prox. Núm: {self.num}")
print("----------------------------------------")
print(" 1. 🚀 Envío de una operacion -> Espera de respuesta")
print(" 2. 📦 Ingestar Lote de Facturas (Masivo)")
print(" 3. 🔍 Consultar Estado de una Factura")
print(" 4. 📬 Ver Facturas Pendientes (AEAT)")
print(" 5. ✔️ Confirmar (ACK) MASIVO (Todas)")
print(" 6. ✔️ Confirmar (ACK) Individual (Por ID)")
print(" 7. 🚪 Salir")
opc = input("Seleccione opción: ").strip()
try:
if opc == '1':
self.ingesta_individual()
elif opc == '2':
self.ingesta_lote()
elif opc == '3':
self.check_status()
elif opc == '4':
self.get_pendientes()
elif opc == '5':
self.ack_masivo()
elif opc == '6':
self.ack_manual()
elif opc == '7' or opc.lower() in ('q', 'exit', 'quit'):
break
else:
print("Opción no válida.")
except Exception as e:
self.error(f"Excepción global capturada: {e}")
# --- Funciones Base ---
def load_template(self):
tpl_path = "factura_ejemplo.json"
try:
with open(tpl_path, "r", encoding="utf-8") as f:
return json.load(f)
except Exception as e:
self.error(f"Error leyendo {tpl_path}: {e}")
return None
def save_template(self, factura):
tpl_path = "factura_ejemplo.json"
try:
with open(tpl_path, "w", encoding="utf-8") as f:
json.dump(factura, f, indent=2, ensure_ascii=False)
self.sync_with_json_template()
except Exception as e:
self.error(f"Error guardando {tpl_path}: {e}")
def sync_with_json_template(self):
factura = self.load_template()
if factura:
cab = factura.get("cabecera", {})
cab_emisor = cab.get("emisor") or cab.get("nif_emisor")
if cab_emisor:
self.nif = str(cab_emisor).strip()
cab_serie = cab.get("serie")
if cab_serie is not None:
self.serie = str(cab_serie).strip()
cab_num = cab.get("numfactura")
if cab_num is not None:
self.num = str(cab_num).strip()
def increment_num(self, current_num):
current_num = str(current_num).strip()
try:
# Simple int increment
num = int(current_num)
return str(num + 1)
except ValueError:
# Fallback for complex strings "F2024-100" -> "F2024-100-1"
return current_num + "-1"
def inject_metadata_and_date(self, factura):
if "cabecera" not in factura:
factura["cabecera"] = {}
factura["cabecera"]["emisor"] = self.nif
now = datetime.now()
factura["cabecera"]["fecha"] = now.strftime("%d/%m/%Y")
factura["cabecera"]["hora"] = now.strftime("%H:%M:%S")
if "metadata" not in factura:
factura["metadata"] = {}
factura["metadata"]["enviar_aeat"] = True
# --- Casos de Uso del Menú ---
def ingesta_individual(self):
self.info(f"Preparando factura {self.serie}-{self.num}...")
factura = self.load_template()
if not factura: return
self.inject_metadata_and_date(factura)
self.info("Enviando...")
res = self.engine.ingesta_json(factura)
print(json.dumps(res, indent=2, ensure_ascii=False))
consultar_serie = self.serie
consultar_num = self.num
factura["cabecera"]["numfactura"] = self.increment_num(self.num)
self.save_template(factura)
self.success(f"Factura enviada. Json guardado preparado para número {self.num}.")
self.info("Esperando respuesta de la AEAT (Polling)...")
for _ in range(15):
time.sleep(1)
st = self.engine.check_status(self.nif, consultar_serie, consultar_num)
status_txt = str(st.get("status", "")).lower()
if status_txt and status_txt not in ["pending", "unknown", "error", "processing", "en cola"]:
print("\n✅ \033[92mRespuesta de la AEAT recibida!\033[0m")
print(f" Estado: {st.get('status', '???')}")
det = st.get("detalles_aeat", {})
csv_val = st.get("csv") or det.get("cab_csv_verifactu") or "N/A"
print(f" CSV: {csv_val}")
huella = det.get("huella", "N/A")
print(f" Huella: {huella}")
qr = det.get("cab_url_qr") or det.get("url_qr_verifactu") or "N/A"
print(f" URL QR: {qr}\n")
break
else:
self.error("⏳ Timeout esperando la respuesta de la AEAT. Puede seguir en cola.")
def ingesta_lote(self):
try:
cantidad = int(input("¿Cuántas facturas deseas enviar en lote? (ej: 5): ").strip())
if cantidad <= 0: raise ValueError()
except ValueError:
self.error("Cantidad inválida.")
return
factura_base = self.load_template()
if not factura_base: return
self.info(f"--- INICIO INGESTA LOTE ({cantidad} FACTURAS) ---")
current_num = self.num
for i in range(cantidad):
# Clonamos via json dumps/loads
factura = json.loads(json.dumps(factura_base))
factura["cabecera"]["numfactura"] = current_num
self.inject_metadata_and_date(factura)
self.info(f"-> Enviando {self.serie}-{current_num} (Factura {i+1}/{cantidad})...")
res = self.engine.ingesta_json(factura)
status = str(res.get("status", res.get("ok", ""))).lower()
if status in ["ok", "true"]:
self.success(f" OK: {current_num}")
else:
self.error(f" Error en {current_num}: {json.dumps(res, ensure_ascii=False)}")
current_num = self.increment_num(current_num)
time.sleep(0.2) # Cortesía al servidor
factura_base["cabecera"]["numfactura"] = current_num
self.save_template(factura_base)
self.success(f"--- FIN DE LOTE ---. Preparado el json para la siguiente: {current_num}")
def check_status(self):
s = input(f"Serie (actual: {self.serie}): ").strip()
if not s: s = self.serie
n = input("Núm (ej: 100): ").strip()
if not n:
self.error("Debe introducir un número")
return
self.info(f"Consultando {self.nif} - {s}-{n}...")
res = self.engine.check_status(self.nif, s, n)
print(json.dumps(res, indent=2, ensure_ascii=False))
def get_pendientes(self):
self.info("Consultando Bandeja de Pendientes AEAT (Últimas 50)...")
res = self.engine.get_pendientes(self.nif, 50)
# Si viene error en json
if isinstance(res, dict) and (str(res.get("ok")).lower() == "false" or "error" in res):
print(json.dumps(res, indent=2, ensure_ascii=False))
return
if isinstance(res, list):
self.draw_pendientes_table(res)
elif "items" in res and isinstance(res["items"], list):
self.draw_pendientes_table(res["items"])
else:
print(json.dumps(res, indent=2, ensure_ascii=False))
def draw_pendientes_table(self, array):
if len(array) == 0:
self.success("\n 🎉 BANDEJA LIMPIA: 0 facturas pendientes.")
return
print("\n--- LISTADO DE PENDIENTES ---")
header = f"{'ID Log':<8} | {'Factura':<15} | {'Importe':<10} | {'Fecha':<10}"
print(header)
print("-" * len(header))
for item in array:
id_log = str(item.get("indice_log", "???"))
id_fact = str(item.get("num_serie_factura", "???"))
total = str(item.get("total", "0"))
date = str(item.get("fecha_expedicion_factura", "???"))
print(f"{id_log:<8} | {id_fact:<15} | {total:<10} | {date:<10}")
print("\nTotal Pendientes:", len(array))
return array
def ack_masivo(self):
self.info("Consiguiendo IDs pendientes para vaciar bandeja...")
res = self.engine.get_pendientes(self.nif, 50)
items_to_ack = []
if isinstance(res, list):
items_to_ack = res
elif "items" in res and isinstance(res["items"], list):
items_to_ack = res["items"]
if not items_to_ack:
self.success("No hay facturas pendientes para hacer ACK.")
return
self.info(f"Iniciando confirmación masiva de {len(items_to_ack)} elementos...")
success_count = 0
for item in items_to_ack:
idx = item.get("indice_log")
if idx is None: continue
fact = item.get("num_serie_factura", str(idx))
print(f" -> Confirmando {fact} (ID: {idx})... ", end="", flush=True)
ack_res = self.engine.ack_indice(self.nif, idx)
ok_status = str(ack_res.get("ok", ack_res.get("status", ""))).lower()
if ok_status in ["true", "ok"]:
print("✅ OK")
success_count += 1
else:
err_msg = json.dumps(ack_res, ensure_ascii=False)
print(f"❌ FALLO: {err_msg}")
self.success(f"Proceso Finalizado. Confirmadas {success_count} de {len(items_to_ack)}.")
def ack_manual(self):
s = input("Introduce el ID (indice_log) interno a confirmar: ").strip()
try:
idx = int(s)
self.info(f"Enviando ACK aislado al ID {idx}...")
res = self.engine.ack_indice(self.nif, idx)
print(json.dumps(res, indent=2, ensure_ascii=False))
except ValueError:
self.error("ID inválido. Debe ser un número entero.")
if __name__ == "__main__":
app = Program()
app.main()factura_ejemplo.json
Plantilla de factura (JSON).
Ver/Ocultar Código Fuente (factura_ejemplo.json)
python
{
"metadata": {
"enviar_aeat": true,
"generar_xml": false,
"insertar_en_db": true,
"simulacion": true
},
"cabecera": {
"emisor": "",
"numfactura": "3662",
"serie": "TK",
"rectificativa": "N",
"tipofacturarectificativa": "",
"fecharectificada": "",
"facturarectificada": "",
"rectificativasustitucion": "",
"facturaf3": "N",
"numseriesustituidaf3": "",
"fechafactsustituidaf3": "",
"descripcionoperacion": "Factura sobre tickets liquidados",
"eliminacion": "",
"tipodoc": "02",
"pais": "ES",
"nif": "",
"nombre": "",
"fecha": "22/02/2026",
"totaliva": "63.04",
"totalrecargo": "0.00",
"base": "442.09",
"base1": "270.83",
"piva1": "10.00",
"iva1": "27.08",
"precargo1": "",
"recargo1": "",
"base2": "171.26",
"piva2": "21.00",
"iva2": "35.96",
"precargo2": "",
"recargo2": "",
"base3": "",
"piva3": "",
"iva3": "",
"precargo3": "",
"recargo3": "",
"base4": "",
"piva4": "",
"iva4": "",
"precargo4": "",
"recargo4": "",
"hora": "17:50:14"
},
"detalle": {
"base": [
"270.83",
"171.26",
"",
""
],
"piva": [
"10.00",
"21.00",
"",
""
],
"iva": [
"27.08",
"35.96",
"",
""
],
"precargo": [
"",
"",
"",
""
],
"recargo": [
"",
"",
"",
""
]
}
}requirements.txt
Dependencias del proyecto.
Ver/Ocultar Código Fuente (requirements.txt)
python
requests