#!/usr/bin/env python3 """ Cliente Python para la API de Adif (Elcano) Obtenido mediante ingeniería reversa de la aplicación móvil """ import requests import json from typing import Optional, Dict, Any, List from datetime import datetime from enum import Enum class TrafficType(Enum): """Tipos de tráfico ferroviario""" CERCANIAS = "CERCANIAS" MEDIA_DISTANCIA = "MEDIA_DISTANCIA" LARGA_DISTANCIA = "LARGA_DISTANCIA" ALL = "ALL" class State(Enum): """Estados para filtros""" YES = "YES" NO = "NO" ALL = "ALL" class AdifClient: """Cliente para interactuar con la API de Adif""" # URLs base BASE_URL_STATIONS = "https://estaciones.api.adif.es" BASE_URL_CIRCULATION = "https://circulacion.api.adif.es" BASE_URL_ELCANOWEB = "https://elcanoweb.adif.es/api" BASE_URL_AVISA = "https://avisa.adif.es" # User keys USER_KEY_CIRCULATIONS = "f4ce9fbfa9d721e39b8984805901b5df" USER_KEY_STATIONS = "0d021447a2fd2ac64553674d5a0c1a6f" # Tokens REGISTRATION_TOKEN = "b9034774-c6e4-4663-a1a8-74bf7102651b" AVISA_BASIC_TOKEN = "YXZpc3RhX2NsaWVudF9hbmRyb2lkOjh5WzZKNyFmSjwhXypmYXE1NyNnOSohNElwa2MjWC1BTg==" SUBSCRIPTIONS_BASIC_TOKEN = "ZGVpbW9zOmRlaW1vc3R0" def __init__(self, debug: bool = False): """ Inicializar el cliente Args: debug: Si True, imprime información de depuración """ self.debug = debug self.session = requests.Session() def _get_headers_stations(self) -> Dict[str, str]: """Headers para endpoints de estaciones""" return { "Content-Type": "application/json;charset=utf-8", "User-key": self.USER_KEY_STATIONS } def _get_headers_circulations(self) -> Dict[str, str]: """Headers para endpoints de circulaciones""" return { "Content-Type": "application/json;charset=utf-8", "User-key": self.USER_KEY_CIRCULATIONS } def _get_headers_avisa(self) -> Dict[str, str]: """Headers para endpoints de Avisa""" return { "Content-Type": "application/json;charset=utf-8", "Authorization": f"Basic {self.AVISA_BASIC_TOKEN}" } def _log(self, message: str): """Log de depuración""" if self.debug: print(f"[DEBUG] {message}") def _request(self, method: str, url: str, headers: Dict[str, str], data: Optional[Dict] = None, params: Optional[Dict] = None) -> Optional[Dict]: """ Realizar petición HTTP Args: method: Método HTTP (GET, POST, etc.) url: URL completa headers: Headers HTTP data: Body JSON (opcional) params: Query parameters (opcional) Returns: Respuesta JSON o None si hay error """ try: self._log(f"{method} {url}") if data: self._log(f"Body: {json.dumps(data, indent=2)}") response = self.session.request( method=method, url=url, headers=headers, json=data, params=params, timeout=30 ) self._log(f"Status: {response.status_code}") if response.status_code == 200: return response.json() else: self._log(f"Error: {response.text}") return { "error": True, "status_code": response.status_code, "message": response.text } except Exception as e: self._log(f"Exception: {str(e)}") return {"error": True, "message": str(e)} # ==================== ESTACIONES ==================== def get_all_stations(self) -> Optional[Dict]: """ Obtener todas las estaciones Returns: Listado de estaciones """ url = f"{self.BASE_URL_STATIONS}/portroyalmanager/secure/stations/allstations/reducedinfo/{self.REGISTRATION_TOKEN}/" return self._request("GET", url, self._get_headers_stations()) def get_station_details(self, station_code: str) -> Optional[Dict]: """ Obtener detalles de una estación Args: station_code: Código de la estación Returns: Detalles de la estación """ url = f"{self.BASE_URL_STATIONS}/portroyalmanager/secure/stations/onestation/" data = {"stationCode": station_code} return self._request("POST", url, self._get_headers_stations(), data=data) def get_station_observations(self, station_code: str) -> Optional[Dict]: """ Obtener observaciones de una estación Args: station_code: Código de la estación Returns: Observaciones de la estación """ url = f"{self.BASE_URL_STATIONS}/portroyalmanager/secure/stationsobservations/" data = {"stationCode": station_code} return self._request("POST", url, self._get_headers_stations(), data=data) # ==================== CIRCULACIONES ==================== def get_departures(self, station_code: str, traffic_type: TrafficType = TrafficType.ALL, commercial_service: State = State.ALL, commercial_stop_type: State = State.ALL, page: int = 0, size: int = 20, origin_station: Optional[str] = None, destination_station: Optional[str] = None) -> Optional[Dict]: """ Obtener salidas desde una estación Args: station_code: Código de la estación traffic_type: Tipo de tráfico (CERCANIAS, MEDIA_DISTANCIA, etc.) commercial_service: Filtro de servicio comercial commercial_stop_type: Filtro de tipo de parada comercial page: Número de página size: Tamaño de página origin_station: Estación origen (opcional) destination_station: Estación destino (opcional) Returns: Salidas de trenes """ url = f"{self.BASE_URL_CIRCULATION}/portroyalmanager/secure/circulationpaths/departures/traffictype/" data = { "commercialService": commercial_service.value, "commercialStopType": commercial_stop_type.value, "stationCode": station_code, "page": { "page": page, "size": size }, "trafficType": traffic_type.value } if origin_station: data["originStationCode"] = origin_station if destination_station: data["destinationStationCode"] = destination_station return self._request("POST", url, self._get_headers_circulations(), data=data) def get_arrivals(self, station_code: str, traffic_type: TrafficType = TrafficType.ALL, commercial_service: State = State.ALL, commercial_stop_type: State = State.ALL, page: int = 0, size: int = 20, origin_station: Optional[str] = None, destination_station: Optional[str] = None) -> Optional[Dict]: """ Obtener llegadas a una estación Args: station_code: Código de la estación traffic_type: Tipo de tráfico commercial_service: Filtro de servicio comercial commercial_stop_type: Filtro de tipo de parada comercial page: Número de página size: Tamaño de página origin_station: Estación origen (opcional) destination_station: Estación destino (opcional) Returns: Llegadas de trenes """ url = f"{self.BASE_URL_CIRCULATION}/portroyalmanager/secure/circulationpaths/arrivals/traffictype/" data = { "commercialService": commercial_service.value, "commercialStopType": commercial_stop_type.value, "stationCode": station_code, "page": { "page": page, "size": size }, "trafficType": traffic_type.value } if origin_station: data["originStationCode"] = origin_station if destination_station: data["destinationStationCode"] = destination_station return self._request("POST", url, self._get_headers_circulations(), data=data) def get_between_stations(self, origin_station: str, destination_station: str, traffic_type: TrafficType = TrafficType.ALL, commercial_service: State = State.ALL, commercial_stop_type: State = State.ALL, page: int = 0, size: int = 20) -> Optional[Dict]: """ Obtener trenes entre dos estaciones Args: origin_station: Estación origen destination_station: Estación destino traffic_type: Tipo de tráfico commercial_service: Filtro de servicio comercial commercial_stop_type: Filtro de tipo de parada comercial page: Número de página size: Tamaño de página Returns: Trenes entre estaciones """ url = f"{self.BASE_URL_CIRCULATION}/portroyalmanager/secure/circulationpaths/betweenstations/traffictype/" data = { "commercialService": commercial_service.value, "commercialStopType": commercial_stop_type.value, "originStationCode": origin_station, "destinationStationCode": destination_station, "page": { "page": page, "size": size }, "trafficType": traffic_type.value } return self._request("POST", url, self._get_headers_circulations(), data=data) def get_path_details(self, commercial_number: Optional[str] = None, origin_station: Optional[str] = None, destination_station: Optional[str] = None, launching_date: Optional[int] = None, all_control_points: bool = False) -> Optional[Dict]: """ Obtener detalles de una ruta/tren específico Args: commercial_number: Número comercial del tren origin_station: Estación origen destination_station: Estación destino launching_date: Fecha de salida (timestamp en milisegundos) all_control_points: Si mostrar todos los puntos de control Returns: Detalles de la ruta """ url = f"{self.BASE_URL_CIRCULATION}/portroyalmanager/secure/circulationpathdetails/onepaths/" data = { "allControlPoints": all_control_points } if commercial_number: data["commercialNumber"] = commercial_number if origin_station: data["originStationCode"] = origin_station if destination_station: data["destinationStationCode"] = destination_station if launching_date: data["launchingDate"] = launching_date return self._request("POST", url, self._get_headers_circulations(), data=data) def get_composition(self, commercial_number: Optional[str] = None, origin_station: Optional[str] = None, destination_station: Optional[str] = None, launching_date: Optional[int] = None) -> Optional[Dict]: """ Obtener composición de un tren (vagones, etc.) Args: commercial_number: Número comercial del tren origin_station: Estación origen destination_station: Estación destino launching_date: Fecha de salida (timestamp en milisegundos) Returns: Composición del tren """ url = f"{self.BASE_URL_CIRCULATION}/portroyalmanager/secure/circulationpaths/compositions/path/" data = {} if commercial_number: data["commercialNumber"] = commercial_number if origin_station: data["originStationCode"] = origin_station if destination_station: data["destinationStationCode"] = destination_station if launching_date: data["launchingDate"] = launching_date return self._request("POST", url, self._get_headers_circulations(), data=data) # ==================== AVISA ==================== def avisa_get_stations(self) -> Optional[Dict]: """ Obtener estaciones de Avisa Returns: Estaciones de Avisa """ url = f"{self.BASE_URL_AVISA}/avisa-ws/api/v1/station" return self._request("GET", url, self._get_headers_avisa()) def avisa_get_categories(self) -> Optional[Dict]: """ Obtener categorías de estaciones Returns: Categorías """ url = f"{self.BASE_URL_AVISA}/avisa-ws/api/v1/category" return self._request("GET", url, self._get_headers_avisa()) def avisa_get_incidences(self) -> Optional[Dict]: """ Obtener incidencias Returns: Lista de incidencias """ url = f"{self.BASE_URL_AVISA}/avisa-ws/api/v1/incidence" return self._request("GET", url, self._get_headers_avisa()) def main(): """Ejemplo de uso""" print("=== Cliente Adif API ===\n") # Crear cliente con modo debug client = AdifClient(debug=True) # Ejemplo: Obtener todas las estaciones print("\n1. Intentando obtener todas las estaciones...") stations = client.get_all_stations() if stations and not stations.get("error"): print(f"✓ Encontradas {len(stations.get('stations', []))} estaciones") else: print(f"✗ Error: {stations}") # Ejemplo: Obtener salidas de Madrid Atocha (código: 10200) print("\n2. Intentando obtener salidas de Madrid Atocha...") departures = client.get_departures( station_code="10200", traffic_type=TrafficType.CERCANIAS, size=5 ) if departures and not departures.get("error"): print(f"✓ Obtenidas salidas") print(json.dumps(departures, indent=2, ensure_ascii=False)[:500] + "...") else: print(f"✗ Error: {departures}") # Ejemplo: Obtener estaciones de Avisa print("\n3. Intentando obtener estaciones de Avisa...") avisa_stations = client.avisa_get_stations() if avisa_stations and not avisa_stations.get("error"): print(f"✓ Obtenidas estaciones de Avisa") else: print(f"✗ Error: {avisa_stations}") if __name__ == "__main__": main()