adif-api-reverse-engineering/adif_client.py

#!/usr/bin/env python3
"""
Cliente completo de la API de ADIF

Implementa todos los endpoints funcionales con métodos simples de usar.
Incluye manejo de errores y validación de datos.
"""

import requests
import uuid
from datetime import datetime
from typing import List, Dict, Optional, Any
from adif_auth import AdifAuthenticator


class AdifClient:
    """Cliente para interactuar con la API de ADIF"""

    def __init__(self, access_key: str, secret_key: str):
        """
        Inicializa el cliente

        Args:
            access_key: Clave de acceso
            secret_key: Clave secreta
        """
        self.auth = AdifAuthenticator(access_key=access_key, secret_key=secret_key)
        self.session = requests.Session()

    def _make_request(
        self,
        url: str,
        payload: Dict[str, Any],
        use_stations_key: bool = False
    ) -> Dict[str, Any]:
        """
        Realiza una petición a la API

        Args:
            url: URL del endpoint
            payload: Datos a enviar
            use_stations_key: Si True, usa USER_KEY_STATIONS en lugar de USER_KEY_CIRCULATION

        Returns:
            Respuesta JSON

        Raises:
            Exception: Si hay un error en la petición
        """
        user_id = str(uuid.uuid4())
        headers = self.auth.get_auth_headers("POST", url, payload, user_id=user_id)

        if use_stations_key:
            headers["User-key"] = self.auth.USER_KEY_STATIONS
        else:
            headers["User-key"] = self.auth.USER_KEY_CIRCULATION

        response = self.session.post(url, json=payload, headers=headers, timeout=30)

        if response.status_code == 200:
            return response.json()
        elif response.status_code == 204:
            return {"message": "No content available", "commercialPaths": []}
        elif response.status_code == 401:
            raise PermissionError(
                f"Unauthorized - Las claves no tienen permisos para este endpoint"
            )
        elif response.status_code == 400:
            raise ValueError(
                f"Bad Request - Payload incorrecto: {response.text}"
            )
        else:
            raise Exception(
                f"Error {response.status_code}: {response.text}"
            )

    def get_departures(
        self,
        station_code: str,
        traffic_type: str = "ALL",
        page_number: int = 0,
        commercial_service: str = "BOTH",
        commercial_stop_type: str = "BOTH"
    ) -> List[Dict[str, Any]]:
        """
        Obtiene las salidas de una estación

        Args:
            station_code: Código de la estación (ej: "10200")
            traffic_type: Tipo de tráfico (ALL, CERCANIAS, AVLDMD, TRAVELERS, GOODS)
            page_number: Número de página (por defecto 0)
            commercial_service: BOTH, YES, NOT
            commercial_stop_type: BOTH, YES, NOT

        Returns:
            Lista de trenes

        Example:
            >>> client = AdifClient(ACCESS_KEY, SECRET_KEY)
            >>> trains = client.get_departures("10200", "AVLDMD")
            >>> for train in trains:
            ...     print(f"{train['commercialNumber']} - Destino: {train['destination']}")
        """
        url = "https://circulacion.api.adif.es/portroyalmanager/secure/circulationpaths/departures/traffictype/"
        payload = {
            "commercialService": commercial_service,
            "commercialStopType": commercial_stop_type,
            "page": {"pageNumber": page_number},
            "stationCode": station_code,
            "trafficType": traffic_type
        }

        data = self._make_request(url, payload)
        return data.get("commercialPaths", [])

    def get_arrivals(
        self,
        station_code: str,
        traffic_type: str = "ALL",
        page_number: int = 0,
        commercial_service: str = "BOTH",
        commercial_stop_type: str = "BOTH"
    ) -> List[Dict[str, Any]]:
        """
        Obtiene las llegadas a una estación

        Args:
            station_code: Código de la estación (ej: "10200")
            traffic_type: Tipo de tráfico (ALL, CERCANIAS, AVLDMD, TRAVELERS, GOODS)
            page_number: Número de página (por defecto 0)
            commercial_service: BOTH, YES, NOT
            commercial_stop_type: BOTH, YES, NOT

        Returns:
            Lista de trenes

        Example:
            >>> client = AdifClient(ACCESS_KEY, SECRET_KEY)
            >>> trains = client.get_arrivals("71801", "ALL")
        """
        url = "https://circulacion.api.adif.es/portroyalmanager/secure/circulationpaths/arrivals/traffictype/"
        payload = {
            "commercialService": commercial_service,
            "commercialStopType": commercial_stop_type,
            "page": {"pageNumber": page_number},
            "stationCode": station_code,
            "trafficType": traffic_type
        }

        data = self._make_request(url, payload)
        return data.get("commercialPaths", [])

    def get_train_route(
        self,
        commercial_number: str,
        launching_date: int,
        origin_station_code: str,
        destination_station_code: str,
        all_control_points: bool = True
    ) -> List[Dict[str, Any]]:
        """
        Obtiene la ruta completa de un tren (todas las paradas)

        Args:
            commercial_number: Número comercial del tren (ej: "03194")
            launching_date: Fecha de salida en milisegundos desde epoch
            origin_station_code: Código de estación de origen
            destination_station_code: Código de estación de destino
            all_control_points: Si True, incluye todos los puntos de control

        Returns:
            Lista de paradas del tren

        Example:
            >>> # Primero obtener un tren real
            >>> trains = client.get_departures("10200", "AVLDMD")
            >>> train = trains[0]
            >>> info = train['commercialPathInfo']
            >>> key = info['commercialPathKey']
            >>>
            >>> # Obtener su ruta completa
            >>> route = client.get_train_route(
            ...     commercial_number=key['commercialCirculationKey']['commercialNumber'],
            ...     launching_date=key['commercialCirculationKey']['launchingDate'],
            ...     origin_station_code=key['originStationCode'],
            ...     destination_station_code=key['destinationStationCode']
            ... )
            >>> for stop in route:
            ...     print(f"Parada: {stop['stationCode']}")
        """
        url = "https://circulacion.api.adif.es/portroyalmanager/secure/circulationpathdetails/onepaths/"
        payload = {
            "allControlPoints": all_control_points,
            "commercialNumber": commercial_number,
            "destinationStationCode": destination_station_code,
            "launchingDate": launching_date,
            "originStationCode": origin_station_code
        }

        data = self._make_request(url, payload)
        commercial_paths = data.get("commercialPaths", [])

        if commercial_paths:
            return commercial_paths[0].get("passthroughSteps", [])
        return []

    def get_station_observations(
        self,
        station_codes: List[str]
    ) -> List[Dict[str, Any]]:
        """
        Obtiene observaciones de estaciones

        Args:
            station_codes: Lista de códigos de estación

        Returns:
            Lista de observaciones

        Example:
            >>> client = AdifClient(ACCESS_KEY, SECRET_KEY)
            >>> obs = client.get_station_observations(["10200", "71801"])
        """
        url = "https://estaciones.api.adif.es/portroyalmanager/secure/stationsobservations/"
        payload = {"stationCodes": station_codes}

        data = self._make_request(url, payload, use_stations_key=True)
        return data.get("stationObservations", [])

    def get_all_departures_with_routes(
        self,
        station_code: str,
        traffic_type: str = "ALL",
        max_trains: int = 5
    ) -> List[Dict[str, Any]]:
        """
        Obtiene salidas de una estación Y sus rutas completas

        Args:
            station_code: Código de estación
            traffic_type: Tipo de tráfico
            max_trains: Número máximo de trenes a procesar

        Returns:
            Lista de trenes con sus rutas

        Example:
            >>> client = AdifClient(ACCESS_KEY, SECRET_KEY)
            >>> trains_with_routes = client.get_all_departures_with_routes("10200", "AVLDMD", max_trains=3)
            >>> for train in trains_with_routes:
            ...     print(f"Tren {train['commercial_number']}")
            ...     for stop in train['route']:
            ...         print(f"  - {stop['stationCode']}")
        """
        departures = self.get_departures(station_code, traffic_type)
        result = []

        for i, train in enumerate(departures[:max_trains]):
            info = train['commercialPathInfo']
            key = info['commercialPathKey']
            commercial_key = key['commercialCirculationKey']

            try:
                route = self.get_train_route(
                    commercial_number=commercial_key['commercialNumber'],
                    launching_date=commercial_key['launchingDate'],
                    origin_station_code=key['originStationCode'],
                    destination_station_code=key['destinationStationCode']
                )

                result.append({
                    "commercial_number": commercial_key['commercialNumber'],
                    "traffic_type": info['trafficType'],
                    "origin_station": key['originStationCode'],
                    "destination_station": key['destinationStationCode'],
                    "launching_date": commercial_key['launchingDate'],
                    "train_info": train,
                    "route": route
                })
            except Exception as e:
                print(f"⚠️ Error obteniendo ruta del tren {commercial_key['commercialNumber']}: {e}")
                continue

        return result


def demo():
    """Demostración del cliente"""
    print("="*70)
    print("DEMO DEL CLIENTE DE ADIF")
    print("="*70)

    ACCESS_KEY = "and20210615"
    SECRET_KEY = "Jthjtr946RTt"

    client = AdifClient(ACCESS_KEY, SECRET_KEY)

    # 1. Salidas de Madrid Atocha
    print("\n1️⃣  SALIDAS DE MADRID ATOCHA (Alta Velocidad)")
    print("-" * 70)
    try:
        departures = client.get_departures("10200", "AVLDMD")
        print(f"✅ Encontrados {len(departures)} trenes")

        for i, train in enumerate(departures[:3]):
            info = train['commercialPathInfo']
            key = info['commercialPathKey']
            passthrough = train.get('passthroughStep', {})
            dep_sides = passthrough.get('departurePassthroughStepSides', {})

            planned_time = dep_sides.get('plannedTime', 0)
            if planned_time:
                time_str = datetime.fromtimestamp(planned_time/1000).strftime('%H:%M')
            else:
                time_str = "N/A"

            print(f"\n   {i+1}. Tren {key['commercialCirculationKey']['commercialNumber']}")
            print(f"      Destino: {key['destinationStationCode']}")
            print(f"      Hora salida: {time_str}")
            print(f"      Estado: {dep_sides.get('circulationState', 'N/A')}")

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

    # 2. Ruta completa de un tren
    print("\n\n2️⃣  RUTA COMPLETA DE UN TREN")
    print("-" * 70)
    try:
        departures = client.get_departures("10200", "ALL")
        if departures:
            train = departures[0]
            info = train['commercialPathInfo']
            key = info['commercialPathKey']
            commercial_key = key['commercialCirculationKey']

            print(f"Consultando ruta del tren {commercial_key['commercialNumber']}...")

            route = client.get_train_route(
                commercial_number=commercial_key['commercialNumber'],
                launching_date=commercial_key['launchingDate'],
                origin_station_code=key['originStationCode'],
                destination_station_code=key['destinationStationCode']
            )

            print(f"✅ Ruta con {len(route)} paradas:\n")
            for i, stop in enumerate(route[:10]):  # Primeras 10 paradas
                stop_type = stop.get('stopType', 'N/A')
                station_code = stop.get('stationCode', 'N/A')

                # Info de salida
                dep_sides = stop.get('departurePassthroughStepSides', {})
                arr_sides = stop.get('arrivalPassthroughStepSides', {})

                if dep_sides:
                    time_ms = dep_sides.get('plannedTime', 0)
                    if time_ms:
                        time_str = datetime.fromtimestamp(time_ms/1000).strftime('%H:%M')
                        print(f"   {i+1}. {station_code} - Salida: {time_str} ({stop_type})")
                elif arr_sides:
                    time_ms = arr_sides.get('plannedTime', 0)
                    if time_ms:
                        time_str = datetime.fromtimestamp(time_ms/1000).strftime('%H:%M')
                        print(f"   {i+1}. {station_code} - Llegada: {time_str} ({stop_type})")
                else:
                    print(f"   {i+1}. {station_code} ({stop_type})")

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

    # 3. Observaciones de estaciones
    print("\n\n3️⃣  OBSERVACIONES DE ESTACIONES")
    print("-" * 70)
    try:
        observations = client.get_station_observations(["10200", "71801"])
        print(f"✅ Observaciones de {len(observations)} estaciones")

        for obs in observations:
            station_code = obs.get('stationCode', 'N/A')
            observation_text = obs.get('observation', 'Sin observaciones')
            print(f"\n   Estación {station_code}:")
            print(f"   {observation_text}")

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

    print("\n" + "="*70)
    print("DEMO COMPLETADA")
    print("="*70)


if __name__ == "__main__":
    demo()