import os import json import time import argparse import base64 import zipfile import io import requests import urllib3 from pathlib import Path from datetime import datetime urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning) class TokenExpiredError(Exception): pass TOKEN_FILE = Path(__file__).parent / "token.json" DOWNLOAD_DIR = Path(__file__).parent / "descargas" DOWNLOAD_DIR.mkdir(exist_ok=True) API_HOST = "https://api-cpe.sunat.gob.pe" BASE_PATH = "/v1/contribuyente/consultacpe/comprobantes" TIPO_COMPROBANTE = { "01": "Factura", "02": "Recibo por Honorarios", "03": "Boleta de Venta", "04": "Liquidación de Compra", "07": "Nota de Crédito", "08": "Nota de Débito", } TIPO_DESCARGA = {"01": "PDF", "02": "XML", "03": "CDR"} def load_token(): if not TOKEN_FILE.exists(): print(f"❌ No se encontró {TOKEN_FILE}") print(" Ejecuta primero: python get_token.py") return None data = json.loads(TOKEN_FILE.read_text()) token = data.get("token") expires = data.get("expires_at", 0) now = time.time() if expires and now > expires: print(f"⚠️ Token expiró. Ejecuta de nuevo: python get_token.py") return None remaining = expires - now print(f"🔑 Token válido por {int(remaining // 60)} min {int(remaining % 60)} seg") return token def cpe_id(ruc, tipo, serie, numero): return f"{ruc}-{tipo}-{serie}-{numero}-2" def get_headers(token): return { "Authorization": f"Bearer {token}", "Accept": "application/json", "User-Agent": "Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/120.0.0.0 Safari/537.36", "Origin": "https://e-factura.sunat.gob.pe", "Referer": "https://e-factura.sunat.gob.pe/", } def consultar_cpe(token, ruc_emisor, tipo, serie, numero, descargar=True, verbose=True): cid = cpe_id(ruc_emisor, tipo, serie, numero) url = f"{API_HOST}{BASE_PATH}/{cid}" if verbose: print(f"\n🌐 Consultando: {ruc_emisor}-{tipo}-{serie}-{numero}") print(f" {url}") session = requests.Session() session.verify = False session.headers.update(get_headers(token)) r = None for intento in range(3): try: r = session.get(url, timeout=120) if verbose: print(f"📡 {r.status_code}") break except Exception as e: if verbose: print(f" ⚠️ Intento {intento+1} falló: {e}") if intento < 2: time.sleep(3) else: if verbose: print(f" ❌ Error después de 3 intentos") return None, 0 if r is None: return None, 0, 0 if r.status_code == 200: data = r.json() ts = datetime.now().strftime("%Y%m%d_%H%M%S") safe = cid.replace("/", "_") json_file = DOWNLOAD_DIR / f"{safe}_{ts}.json" json_file.write_text(json.dumps(data, indent=2, ensure_ascii=False)) if verbose: print(f"💾 JSON: {json_file}") if descargar: for tipo_dl in ["02", "01", "03"]: descargar_archivo(token, cid, tipo_dl, ts, verbose=verbose) return data, 200 elif r.status_code == 401: if verbose: print(" ❌ Token inválido. Ejecuta: python get_token.py") raise TokenExpiredError("Token expirado o inválido (401)") elif r.status_code == 404: if verbose: print(" ❌ No encontrado (404)") else: if verbose: print(f" ❌ Error {r.status_code}: {r.text[:300]}") return None, r.status_code def descargar_archivo(token, cid, tipo, ts, verbose=True): label = TIPO_DESCARGA.get(tipo, tipo) url = f"{API_HOST}{BASE_PATH}/{cid}/{tipo}" if verbose: print(f"\n📥 Descargando {label}...") print(f" {url}") session = requests.Session() session.verify = False session.headers.update(get_headers(token)) r = None for intento in range(3): try: r = session.get(url, timeout=120) if verbose: print(f"📡 {r.status_code}") break except Exception as e: if verbose: print(f" ⚠️ Intento {intento+1} falló: {e}") if intento < 2: time.sleep(3) else: if verbose: print(f" ❌ Error después de 3 intentos") return False if r is None: return False if r.status_code == 200: data = r.json() val_archivo = data.get("valArchivo", "") if not val_archivo: if verbose: print(f" ⚠️ Respuesta sin valArchivo") return False raw = base64.b64decode(val_archivo) file_base = cid[:-2] if cid.endswith("-2") else cid zip_name = f"{file_base}-{label}.zip" zip_path = DOWNLOAD_DIR / zip_name zip_path.write_bytes(raw) if verbose: print(f" 💾 ZIP: {zip_path} ({len(raw)} bytes)") try: with zipfile.ZipFile(io.BytesIO(raw)) as zf: for member in zf.namelist(): content = zf.read(member) member_lower = member.lower() if label == "PDF" and member_lower.endswith(".pdf"): out_name = f"{file_base}.pdf" elif label == "XML": if not member_lower.endswith(".xml"): continue if member.startswith("R-") or member.startswith("r-"): out_name = f"CDR-{file_base}.xml" else: out_name = f"{file_base}.xml" elif label == "CDR" and member_lower.endswith(".xml"): out_name = f"CDR-{file_base}.xml" else: continue out_path = DOWNLOAD_DIR / out_name out_path.write_bytes(content) if verbose: print(f" ✅ {out_name} ({len(content)} bytes)") except Exception as e: if verbose: print(f" ⚠️ Error extrayendo ZIP: {e}") return True elif r.status_code == 401: if verbose: print(" ❌ Token inválido (401)") raise TokenExpiredError("Token expirado o inválido (401)") elif r.status_code == 400: if verbose: print(f" ⚠️ No disponible (400)") else: if verbose: print(f" ❌ Error {r.status_code}") return False def main(): print("=" * 60) print("🔍 CONSULTA CPE - SUNAT API") print("=" * 60) parser = argparse.ArgumentParser(description="Consultar CPE en SUNAT") parser.add_argument("--ruc", required=True, help="RUC del emisor") parser.add_argument("--tipo", required=True, help="Tipo (01=Factura, 03=Boleta, 07=NC, 08=ND)") parser.add_argument("--serie", required=True, help="Serie (ej: F007)") parser.add_argument("--numero", required=True, help="Número del comprobante") parser.add_argument("--no-download", action="store_true", help="No descargar archivos") args = parser.parse_args() token = load_token() if not token: return consultar_cpe(token, args.ruc, args.tipo, args.serie, args.numero, descargar=not args.no_download) if __name__ == "__main__": main()