#!/usr/bin/env python3 """ Flightradar24 API Client для sandbox и production Поддержка обоих окружений для разработки прототипа карты шумового загрязнения """ import os import sys import json import requests from datetime import datetime, timedelta from typing import Optional, Dict, Any, Tuple class Flightradar24Client: """Клиент для работы с Flightradar24 API (sandbox и production)""" def __init__(self, use_sandbox: bool = True, api_key: Optional[str] = None): """ Инициализация клиента Args: use_sandbox: Использовать sandbox окружение (True) или production (False) api_key: Ключ API (если None, берётся из переменных окружения) """ self.use_sandbox = use_sandbox if use_sandbox: self.base_url = "https://fr24api.flightradar24.com/api" self.api_key = api_key or os.getenv("FLIGHTRADAR24_SANDBOX_KEY") or os.getenv("FLIGHTRADAR24_API_KEY") self.default_headers = { "Accept": "application/json", "Accept-Version": "v1" } else: self.base_url = "https://api.flightradar24.com/common/v1" self.api_key = api_key or os.getenv("FLIGHTRADAR24_PRODUCTION_KEY") or os.getenv("FLIGHTRADAR24_API_KEY") self.default_headers = { "Content-Type": "application/json" } if not self.api_key: raise ValueError( "API ключ не предоставлен. Установите переменные окружения:\n" " - FLIGHTRADAR24_SANDBOX_KEY (для sandbox)\n" " - FLIGHTRADAR24_PRODUCTION_KEY (для production)\n" " - Или FLIGHTRADAR24_API_KEY (по умолчанию для sandbox)" ) self.session = requests.Session() self.session.headers.update({ "Authorization": f"Bearer {self.api_key}", **self.default_headers }) print(f"🔧 Инициализирован клиент для {'sandbox' if use_sandbox else 'production'}") print(f" Base URL: {self.base_url}") print(f" Key: {self.api_key[:15]}...{self.api_key[-10:] if len(self.api_key) > 25 else '***'}") def _make_request(self, endpoint: str, params: Optional[Dict] = None) -> Dict[str, Any]: """Базовый метод для выполнения запросов к API""" url = f"{self.base_url}{endpoint}" try: response = self.session.get(url, params=params, timeout=30) response.raise_for_status() return response.json() except requests.exceptions.RequestException as e: print(f"⚠️ Ошибка при запросе к {url}: {e}") if hasattr(e, 'response') and e.response: print(f" Статус: {e.response.status_code}") print(f" Ответ: {e.response.text[:500]}") raise def get_airport_info_light(self, airport_code: str) -> Dict[str, Any]: """Получить базовую информацию об аэропорте (light endpoint)""" return self._make_request(f"/airport/light/{airport_code}") def get_airline_info_light(self, airline_code: str) -> Dict[str, Any]: """Получить базовую информацию об авиакомпании (light endpoint)""" return self._make_request(f"/airline/light/{airline_code}") def get_live_flight_positions(self, bounds: Optional[str] = None, limit: int = 10) -> Dict[str, Any]: """ Получить live позиции самолетов Args: bounds: Границы в формате 'lat1,lon1,lat2,lon2' (опционально) limit: Ограничение количества результатов """ params = {"limit": limit} if bounds: params["bounds"] = bounds return self._make_request("/flight/list", params) def get_flight_details(self, flight_id: str) -> Dict[str, Any]: """Получить детальную информацию о конкретном рейсе""" return self._make_request(f"/flight/{flight_id}") def get_historical_flight_events(self, flight_id: str, start_time: Optional[str] = None, end_time: Optional[str] = None) -> Dict[str, Any]: """ Получить исторические события полета Args: flight_id: Идентификатор рейса start_time: Начальное время в формате ISO (опционально) end_time: Конечное время в формате ISO (опционально) """ params = {} if start_time: params["start"] = start_time if end_time: params["end"] = end_time return self._make_request(f"/flight/{flight_id}/history", params) def get_flight_tracks(self, flight_id: str) -> Dict[str, Any]: """Получить треки полета (playback)""" return self._make_request(f"/flight/{flight_id}/playback") def get_usage_report(self) -> Dict[str, Any]: """Получить отчет об использовании API (использованные кредиты)""" return self._make_request("/usage") def search_flights(self, query: str, limit: int = 20) -> Dict[str, Any]: """ Поиск рейсов по различным критериям Args: query: Строка поиска (номер рейса, callsign, регистрация) limit: Ограничение количества результатов """ params = {"query": query, "limit": limit} return self._make_request("/search", params) def print_flight_info(flight_data: Dict[str, Any]): """Красиво вывести информацию о рейсе""" if "data" in flight_data and flight_data["data"]: flight = flight_data["data"][0] print(f"Рейс: {flight.get('callsign', 'N/A')}") print(f"Номер: {flight.get('flight', 'N/A')}") print(f"Самолет: {flight.get('type', 'N/A')} ({flight.get('reg', 'N/A')})") print(f"Откуда: {flight.get('orig_icao', 'N/A')} -> Куда: {flight.get('dest_icao', 'N/A')}") print(f"Высота: {flight.get('alt', 'N/A')} ft, Скорость: {flight.get('gspeed', 'N/A')} kts") if "lat" in flight and "lon" in flight: print(f"Координаты: {flight['lat']:.4f}, {flight['lon']:.4f}") print(f"Источник данных: {flight.get('source', 'N/A')}") print("-" * 50) def main(): """Основная функция демонстрации""" print("=== Flightradar24 Explorer API Demo ===") print("Тариф Explorer предоставляет доступ к следующим endpoint'ам:") print("1. Статические данные (аэропорты/авиакомпании light)") print("2. Live позиции самолетов (light и full)") print("3. Исторические события полетов") print("4. Треки полетов (playback)") print("5. Поиск рейсов") print("6. Отчет об использовании API") print() # Проверка API ключа api_key = os.getenv("FLIGHTRADAR24_API_KEY") if not api_key: print("❌ API ключ не найден в переменных окружения.") print("Установите его командой: export FLIGHTRADAR24_API_KEY='your_key_here'") print("Или передайте как аргумент командной строки.") return try: # Инициализация клиента client = Flightradar24Explorer(api_key) print("✅ API клиент инициализирован успешно") # Демонстрация возможностей (закомментирована, т.к. требует реальных запросов) print("\nДля использования раскомментируйте нужные вызовы в коде:") print("# 1. Получить информацию об аэропорте") print("# airport_info = client.get_airport_info_light('SVO')") print("# print(json.dumps(airport_info, indent=2, ensure_ascii=False))") print() print("# 2. Получить live позиции самолетов") print("# flights = client.get_live_flight_positions(limit=5)") print("# print_flight_info(flights)") print() print("# 3. Поиск рейсов") print("# results = client.search_flights('SU100')") print("# print(json.dumps(results, indent=2, ensure_ascii=False))") print() print("# 4. Получить отчет об использовании") print("# usage = client.get_usage_report()") print("# print(json.dumps(usage, indent=2, ensure_ascii=False))") print("\n📋 Примеры запросов подготовлены.") print("Для начала работы укажите конкретные задачи:") print("- Мониторинг конкретных рейсов") print("- Анализ маршрутов авиакомпаний") print("- Построение дашбордов") print("- Сбор статистики") except Exception as e: print(f"❌ Ошибка: {e}") if __name__ == "__main__": main()