246 lines
8.1 KiB
Python
246 lines
8.1 KiB
Python
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()
|