proyecto en producción

sunat_manager.py

@@ -0,0 +1,80 @@
+import requests
+import json
+import urllib3
+import time
+
+# Desactivar advertencias de SSL no verificado
+urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning)
+
+# ============================================================================
+def token(datos):
+
+    CLIENT_ID=datos["Client_ID"]
+    CLIENT_SECRET=datos["Client_Secret"]
+    
+    url = f"https://api-seguridad.sunat.gob.pe/v1/clientesextranet/{CLIENT_ID}/oauth2/token/"
+
+    headers = {
+        'Content-Type': 'application/x-www-form-urlencoded',
+        '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',
+        'Accept': 'application/json, text/plain, */*',
+        'Origin': 'https://api-seguridad.sunat.gob.pe'
+    }
+    
+    payload = 'grant_type=client_credentials' + '&'
+    payload += 'scope=https://api.sunat.gob.pe/v1/contribuyente/contribuyentes' + '&'
+    payload += f'client_id={CLIENT_ID}' + '&'
+    payload += f'client_secret={CLIENT_SECRET}'
+
+    response = requests.request("POST", url, headers=headers, data=payload, verify=False, timeout=30)
+
+    print(f"   📤 Token request enviado")
+    print(f"   📥 Token response: {response.status_code}")
+
+    if response.status_code == 200:
+        result = response.json()
+        
+        if "access_token" in result:
+            token = result["access_token"]
+            return token
+        else:
+            return ""
+    else:
+        return ""
+
+# ============================================================================
+def validar_comprobante(token,RUC,payload:dict):
+    url = f"https://api.sunat.gob.pe/v1/contribuyente/contribuyentes/{RUC}/validarcomprobante"
+
+    # payload = {
+    # "numRuc": "20538298485",
+    # "codComp": "08",
+    # "numeroSerie": "E001",
+    # "numero": "3773",
+    # "fechaEmision": "15/11/2023",
+    # "monto": "1328.52"
+    # }
+    
+    headers = {
+        'Content-Type': 'application/json',
+        'Authorization': f'Bearer {token}',
+        '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',
+        'Accept': 'application/json, text/plain, */*'
+    }
+
+    max_retries = 3
+    for attempt in range(max_retries):
+        try:
+            response = requests.request("POST", url, headers=headers, json=payload, verify=False, timeout=30)
+            print(f"   📥 Status code: {response.status_code}")
+            return response.json()
+        except (requests.exceptions.ConnectionError, requests.exceptions.Timeout) as e:
+            print(f"   ⚠️ Error de conexión (intento {attempt + 1}/{max_retries}): {e}")
+            if attempt < max_retries - 1:
+                time_to_wait = 2 ** (attempt + 1)
+                print(f"   🔄 Reintentando en {time_to_wait} segundos...")
+                time.sleep(time_to_wait)
+            else:
+                raise e
+
+# ============================================================================