210 lines
10 KiB
Python
210 lines
10 KiB
Python
#!/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() |