88 lines
3.2 KiB
Python
88 lines
3.2 KiB
Python
import os
|
|
import requests
|
|
from typing import List, Dict, Optional
|
|
|
|
RADIO_BROWSER_INSTANCE = os.getenv("RADIO_BROWSER_INSTANCE", "https://de1.api.radio-browser.info")
|
|
|
|
|
|
def fetch_stations(country: Optional[str] = None, genre: Optional[str] = None, limit: int = 50) -> List[Dict]:
|
|
try:
|
|
url = f"{RADIO_BROWSER_INSTANCE}/json/stations/getTopReverseGeoByCountryHnattrialphabetic"
|
|
params = {"limit": limit, "order": "votes", "reverse": "true"}
|
|
|
|
if country:
|
|
params["countrycode"] = country
|
|
if genre:
|
|
params["tag"] = genre
|
|
|
|
response = requests.get(url, params=params, timeout=10)
|
|
if response.status_code != 200:
|
|
# Fallback: get top stations by votes
|
|
url = f"{RADIO_BROWSER_INSTANCE}/json/stations/getTopByVotes"
|
|
response = requests.get(url, params=params, timeout=10)
|
|
if response.status_code != 200:
|
|
return []
|
|
|
|
stations = response.json()
|
|
return [_format_station(s) for s in stations]
|
|
except (requests.RequestException, KeyError):
|
|
return []
|
|
|
|
|
|
def fetch_nearby_stations(lat: float, lon: float, radius: int = 100) -> List[Dict]:
|
|
try:
|
|
url = f"{RADIO_BROWSER_INSTANCE}/json/stations/search"
|
|
params = {
|
|
"name": "",
|
|
"order": "votes",
|
|
"reverse": "true",
|
|
"limit": 50,
|
|
"offset": 0,
|
|
"break_on_circular": "true",
|
|
"hidebroken": "true",
|
|
}
|
|
|
|
response = requests.get(url, params=params, timeout=10)
|
|
if response.status_code != 200:
|
|
return []
|
|
|
|
stations = response.json()
|
|
nearby = []
|
|
for s in stations:
|
|
s_lat = s.get("geo_lat")
|
|
s_lon = s.get("geo_long")
|
|
if s_lat and s_lon:
|
|
distance = _haversine(lat, lon, float(s_lat), float(s_lon))
|
|
if distance <= radius:
|
|
nearby.append(_format_station(s))
|
|
|
|
return sorted(nearby, key=lambda x: x.get("distance", 999))[:20]
|
|
except (requests.RequestException, KeyError):
|
|
return []
|
|
|
|
|
|
def _format_station(s: Dict) -> Dict:
|
|
return {
|
|
"id": s.get("stationuuid") or s.get("name", ""),
|
|
"name": s.get("name", "Unknown"),
|
|
"frequency": s.get("codec"),
|
|
"stream_url": s.get("url_resolved") or s.get("url", ""),
|
|
"location_lat": float(s["geo_lat"]) if s.get("geo_lat") else None,
|
|
"location_lon": float(s["geo_long"]) if s.get("geo_long") else None,
|
|
"genre": s.get("tag", ""),
|
|
"country": s.get("countryname", ""),
|
|
"language": s.get("language", ""),
|
|
"bitrate": int(s["bitrate"]) if s.get("bitrate") else None,
|
|
"tags": s.get("tag", "").split(",") if s.get("tag") else [],
|
|
"votes": int(s["votes"]) if s.get("votes") else 0,
|
|
}
|
|
|
|
|
|
def _haversine(lat1: float, lon1: float, lat2: float, lon2: float) -> float:
|
|
import math
|
|
R = 6371
|
|
d_lat = math.radians(lat2 - lat1)
|
|
d_lon = math.radians(lon2 - lon1)
|
|
a = math.sin(d_lat / 2) ** 2 + math.cos(math.radians(lat1)) * math.cos(math.radians(lat2)) * math.sin(d_lon / 2) ** 2
|
|
c = 2 * math.atan2(math.sqrt(a), math.sqrt(1 - a))
|
|
return R * c |