Solución de Problemas

Esta sección cubre los problemas más comunes que puedes encontrar al usar el SDK de Datadis y cómo resolverlos.

Problemas de Autenticación

Error: «Error de autenticación»

Síntomas: - AuthenticationError al inicializar el cliente - Mensaje: «Error de autenticación: 401» o similar

Causas posibles:

1. Credenciales incorrectas

   # Incorrecto
   client = SimpleDatadisClientV1("12345678A", "password_incorrecto")
   

   Solución: Verifica que tu NIF y contraseña sean correctos en datadis.es

2. Formato de NIF incorrecto

   # Incorrecto
   client = SimpleDatadisClientV1("12345678", "password")  # Sin letra
   
   # Correcto
   client = SimpleDatadisClientV1("12345678A", "password")
   

3. Cuenta bloqueada o inactiva

   Solución: Inicia sesión directamente en la web de Datadis para verificar el estado de tu cuenta.

Error: «Timeout en autenticación»

Síntomas: - El proceso de autenticación se cuelga o falla por timeout

Soluciones:

1. Aumentar timeout de autenticación

   client = SimpleDatadisClientV1(
       username="tu_nif",
       password="tu_password",
       timeout=180  # Aumentar a 3 minutos
   )
   

2. Problemas de conectividad

   # Verificar conectividad
   ping datadis.es
   curl -I https://datadis.es
   

Problemas de Conectividad

Error: «Timeout después de X intentos»

Síntomas: - Timeouts frecuentes en operaciones de la API - Respuestas lentas o incompletas

Soluciones:

1. Configurar timeouts más largos

   client = SimpleDatadisClientV1(
       username="tu_nif",
       password="tu_password",
       timeout=300,  # 5 minutos
       retries=5     # Más reintentos
   )
   

2. Implementar reintentos personalizados

   import time
   from datadis_python.exceptions import DatadisError
   
   def obtener_con_reintentos(func, max_intentos=5):
       for intento in range(max_intentos):
           try:
               return func()
           except DatadisError as e:
               if "timeout" in str(e).lower():
                   espera = min(60, (2 ** intento) * 5)
                   print(f"Timeout. Reintentando en {espera}s...")
                   time.sleep(espera)
               else:
                   raise
       raise DatadisError("Demasiados timeouts")
   

Error: «Connection refused» o «Network unreachable»

Diagnóstico:

import requests

def verificar_conectividad():
    """Verifica la conectividad con Datadis"""
    urls = [
        "https://datadis.es",
        "https://api.datadis.es",
    ]

    for url in urls:
        try:
            response = requests.get(url, timeout=10)
            print(f"{url}: {response.status_code}")
        except Exception as e:
            print(f"{url}: {e}")

Soluciones: - Verificar conexión a internet - Comprobar firewall/proxy corporativo - Usar VPN si estás fuera de España

Problemas con Datos

Error: «No se encontraron datos»

Síntomas: - Las listas devueltas están vacías - No se obtienen datos de consumo o suministros

Diagnósticos:

def diagnosticar_datos(client):
    """Diagnóstica problemas con datos"""

    print("Diagnosticando disponibilidad de datos...")

    # 1. Verificar distribuidores
    distribuidores = client.get_distributors()
    print(f"Distribuidores: {len(distribuidores)}")

    # 2. Verificar suministros
    suministros = client.get_supplies()
    print(f"Suministros: {len(suministros)}")

    if not suministros:
        print("No hay suministros disponibles. Verifica tu cuenta en datadis.es")
        return

    # 3. Probar diferentes rangos de fechas
    from datetime import datetime, timedelta

    suministro = suministros[0]
    # Obtener código de distribuidor correcto
    distribuidor = "2"  # Por defecto
    if distribuidores and distribuidores[0].distributor_codes:
        distribuidor = distribuidores[0].distributor_codes[0]

    rangos = [
        (30, "último mes"),
        (90, "últimos 3 meses"),
        (365, "último año")
    ]

    for dias, descripcion in rangos:
        fin = datetime.now()
        inicio = fin - timedelta(days=dias)

        try:
            # CORRECCIÓN: Usar formato mensual YYYY/MM requerido por la API
            consumo = client.get_consumption(
                cups=suministro.cups,
                distributor_code=distribuidor,
                date_from=inicio.strftime("%Y/%m"),
                date_to=fin.strftime("%Y/%m")
            )
            print(f"{descripcion}: {len(consumo)} registros")

            if consumo:
                break

        except Exception as e:
            print(f"Error en {descripcion}: {e}")

Error: «Error validando datos»

Síntomas: - Warnings sobre errores de validación - Algunos registros no se procesan

Soluciones:

1. Inspeccionar datos problemáticos

   def inspeccionar_respuesta_api(endpoint_func, *args, **kwargs):
       """Inspecciona la respuesta cruda de la API"""
   
       # Hacer request manual para ver datos crudos
       import json
       from datadis_python.client.v1.simple_client import SimpleDatadisClientV1
   
       with SimpleDatadisClientV1("tu_nif", "tu_password") as client:
           # Acceder al método interno para ver respuesta cruda
           response = client._make_authenticated_request(
               endpoint="/get-consumption",  # Ejemplo
               params={
                   "cups": "tu_cups",
                   "distributorCode": "2",
                   "startDate": "2024/01/01",
                   "endDate": "2024/01/31"
               }
           )
   
           # Guardar respuesta para análisis
           with open("respuesta_cruda.json", "w") as f:
               json.dump(response, f, indent=2)
   
           print("Respuesta guardada en respuesta_cruda.json")
   

2. Manejo robusto de validación

   from pydantic import ValidationError
   from datadis_python.models.consumption import ConsumptionData
   
   def procesar_datos_con_tolerancia(datos_crudos):
       """Procesa datos con tolerancia a errores"""
   
       datos_validos = []
       errores = []
   
       for i, item in enumerate(datos_crudos):
           try:
               dato_validado = ConsumptionData(**item)
               datos_validos.append(dato_validado)
           except ValidationError as e:
               errores.append({
                   "indice": i,
                   "datos": item,
                   "error": str(e)
               })
   
       print(f"Procesados: {len(datos_validos)}")
       print(f"Errores: {len(errores)}")
   
       return datos_validos, errores
   

Problemas de Rendimiento

Consultas Muy Lentas

Síntomas: - Las consultas tardan minutos en completarse - Timeouts frecuentes

Optimizaciones:

1. Reducir el rango de fechas

   # Muy amplio (puede ser lento)
   consumo = client.get_consumption(
       cups=cups,
       distributor_code=distributor_code,
       date_from="2020/01",  # 4 años de datos - formato mensual
       date_to="2024/01"
   )
   
   # Rangos más pequeños
   from datetime import datetime, timedelta
   
   def obtener_consumo_por_meses(client, cups, distributor_code, fecha_inicio, fecha_fin):
       """Obtiene datos mes a mes para evitar timeouts"""
   
       todos_los_datos = []
       # CORRECCIÓN: Trabajar con formato mensual desde el inicio
       fecha_actual = datetime.strptime(fecha_inicio, "%Y/%m")
       fecha_limite = datetime.strptime(fecha_fin, "%Y/%m")
   
       while fecha_actual <= fecha_limite:
           # Calcular fin de mes
           if fecha_actual.month == 12:
               fin_mes = fecha_actual.replace(year=fecha_actual.year + 1, month=1)
           else:
               fin_mes = fecha_actual.replace(month=fecha_actual.month + 1)
   
           fin_mes = min(fin_mes, fecha_limite)
   
           print(f"Obteniendo datos: {fecha_actual.strftime('%Y/%m')} - {fin_mes.strftime('%Y/%m')}")
   
           try:
               datos_mes = client.get_consumption(
                   cups=cups,
                   distributor_code=distributor_code,
                   date_from=fecha_actual.strftime("%Y/%m"),
                   date_to=fin_mes.strftime("%Y/%m")
               )
               todos_los_datos.extend(datos_mes)
   
           except Exception as e:
               print(f"Error en mes {fecha_actual.strftime('%Y/%m')}: {e}")
   
           # Siguiente mes
           if fecha_actual.month == 12:
               fecha_actual = fecha_actual.replace(year=fecha_actual.year + 1, month=1)
           else:
               fecha_actual = fecha_actual.replace(month=fecha_actual.month + 1)
   
       return todos_los_datos
   

2. Procesamiento en paralelo (con cuidado)

   import asyncio
   import time
   from concurrent.futures import ThreadPoolExecutor
   
   def obtener_datos_paralelo(suministros, username, password):
       """Obtiene datos de múltiples suministros en paralelo"""
   
       def procesar_suministro(suministro):
           with SimpleDatadisClientV1(username, password) as client:
            return client.get_consumption(
                cups=suministro.cups,
                distributor_code="2",
                date_from="2024/01",  # Formato mensual
                date_to="2024/01"
            )          # Limitar concurrencia para no sobrecargar la API
       with ThreadPoolExecutor(max_workers=2) as executor:
           resultados = list(executor.map(procesar_suministro, suministros))
   
       return resultados
   

Problemas Específicos de la API

Error: «CUPS no válido»

Síntomas: - Error 400 con mensaje sobre CUPS inválido

Verificaciones:

def validar_cups(cups):
    """Valida formato de CUPS"""
    import re

    # CUPS debe tener 22 caracteres: ES + 20 dígitos/letras
    patron = r'^ES\d{16}[A-Z]{2}\d[A-Z]$'

    if not re.match(patron, cups):
        print(f"CUPS inválido: {cups}")
        print("Formato esperado: ES + 16 dígitos + 2 letras + 1 dígito + 1 letra")
        return False

    print(f"CUPS válido: {cups}")
    return True

Error: «Distributor code no válido»

Solución:

def obtener_codigo_distribuidor_valido(client, cups):
    """Obtiene el código de distribuidor correcto"""

    # Primero obtener lista de distribuidores
    distribuidores = client.get_distributors()
    print("Distribuidores disponibles:")
    if distribuidores and distribuidores[0].distributor_codes:
        for code in distribuidores[0].distributor_codes:
            print(f"  - {code}")

    # Si conoces la provincia, puedes intentar mapear
    mapeo_provincias = {
        "Madrid": "2",
        "Barcelona": "1",
        "Valencia": "3",
        # Añadir más según necesidad
    }

    return distribuidores[0].distributor_codes[0] if distribuidores and distribuidores[0].distributor_codes else "2"

Herramientas de Diagnóstico

Script de Diagnóstico Completo

#!/usr/bin/env python3
"""
Script de diagnóstico para problemas con el SDK de Datadis
"""

import sys
import traceback
from datetime import datetime, timedelta
from datadis_python.client.v1.simple_client import SimpleDatadisClientV1
from datadis_python.exceptions import DatadisError

def ejecutar_diagnostico(username, password):
    """Ejecuta diagnóstico completo"""

    print("DIAGNÓSTICO DEL SDK DE DATADIS")
    print("=" * 50)

    try:
        with SimpleDatadisClientV1(username, password, timeout=60) as client:

            # 1. Test de autenticación
            print("\n1. Test de autenticación...")
            if client.token:
                print("Autenticación exitosa")
            else:
                print("Error de autenticación")
                return

            # 2. Test de distribuidores
            print("\n2. Test de distribuidores...")
            try:
                distribuidores = client.get_distributors()
                print(f"Distribuidores obtenidos: {len(distribuidores)}")
                if distribuidores and distribuidores[0].distributor_codes:
                    for code in distribuidores[0].distributor_codes:
                        print(f"   - {code}")
            except Exception as e:
                print(f"Error obteniendo distribuidores: {e}")

            # 3. Test de suministros
            print("\n3. Test de suministros...")
            try:
                suministros = client.get_supplies()
                print(f"Suministros obtenidos: {len(suministros)}")
                for i, sup in enumerate(suministros[:3]):  # Solo primeros 3
                    print(f"   {i+1}. {sup.cups} - {sup.address}")
            except Exception as e:
                print(f"Error obteniendo suministros: {e}")
                return

            if not suministros:
                print("No hay suministros disponibles")
                return

            # 4. Test de consumo
            print("\n4. Test de consumo...")
            suministro = suministros[0]
            # Obtener código de distribuidor correcto
            codigo_dist = "2"  # Por defecto
            if distribuidores and distribuidores[0].distributor_codes:
                codigo_dist = distribuidores[0].distributor_codes[0]

            # Probar diferentes rangos
            rangos_test = [
                (1, "este mes"),
                (2, "últimos 2 meses"),
                (3, "últimos 3 meses")
            ]

            for meses, descripcion in rangos_test:
                try:
                    fin = datetime.now()
                    inicio = fin - timedelta(days=30*meses)

                    # CORRECCIÓN: Usar formato mensual YYYY/MM
                    consumo = client.get_consumption(
                        cups=suministro.cups,
                        distributor_code=codigo_dist,
                        date_from=inicio.strftime("%Y/%m"),
                        date_to=fin.strftime("%Y/%m")
                    )

                    if consumo:
                        total = sum(c.consumption_kwh for c in consumo)
                        print(f"   {descripcion}: {len(consumo)} registros, {total:.2f} kWh total")
                        break
                    else:
                        print(f"   {descripcion}: sin datos")

                except Exception as e:
                    print(f"   {descripcion}: {e}")

            print("\nDiagnóstico completado")

    except DatadisError as e:
        print(f"\nError del SDK: {e}")
        print("\nDetalles técnicos:")
        traceback.print_exc()

    except Exception as e:
        print(f"\nError inesperado: {e}")
        print("\nDetalles técnicos:")
        traceback.print_exc()

if __name__ == "__main__":
    if len(sys.argv) != 3:
        print("Uso: python diagnostico.py <NIF> <contraseña>")
        sys.exit(1)

    username, password = sys.argv[1], sys.argv[2]
    ejecutar_diagnostico(username, password)

Contacto y Soporte

Si después de seguir esta guía sigues teniendo problemas:

1. Revisa los logs: Activa logging detallado para obtener más información

2. Documenta el error: Incluye el mensaje de error completo y contexto

3. Verifica versiones: Asegúrate de usar la versión más reciente del SDK

4. Crea un issue: Reporta el problema en el repositorio de GitHub

# Activar logging detallado
import logging
logging.basicConfig(level=logging.DEBUG)

# Mostrar información del sistema
import sys
import datadis_python

print(f"Python: {sys.version}")
print(f"SDK Datadis: {datadis_python.__version__}")

Información útil para reportar bugs: - Versión de Python - Versión del SDK - Sistema operativo - Mensaje de error completo - Código que causa el problema - Datos de entrada (sin credenciales)