#!/usr/bin/env python3 """ Multi-Country Radio M3U Playlist Generator (resilient + dynamic mirrors) ----------------------------------------------------------------------- Generuje M3U playlist s radii z více zemí pomocí radio-browser.info API. Automaticky detekuje dostupné servery přes DNS a přepíná mezi nimi. """ import sys import os import argparse import time import random import requests import socket from datetime import datetime from pyradios import RadioBrowser class RateLimitError(Exception): """Exception raised when rate limit is hit""" pass # === Funkce pro detekci aktuálních mirrorů === def get_dynamic_mirrors(): """Získá aktuální dostupné RadioBrowser servery z DNS""" hosts = [] try: infos = socket.getaddrinfo('all.api.radio-browser.info', 80, 0, 0, socket.IPPROTO_TCP) for info in infos: ip = info[4][0] try: host = socket.gethostbyaddr(ip)[0] if host not in hosts: hosts.append(host) except socket.herror: continue hosts.sort() mirrors = [f"https://{h}" for h in hosts] if mirrors: print(f"🌍 Načteno {len(mirrors)} mirrorů z DNS:") for m in mirrors: print(" •", m) return mirrors except Exception as e: print("⚠️ Nepodařilo se zjistit servery přes DNS:", e) return [] # === Inicializace RadioBrowseru s fallbacky === def init_radio_browser(): mirrors = get_dynamic_mirrors() if not mirrors: mirrors = [ "https://de1.api.radio-browser.info", "https://de2.api.radio-browser.info", "https://fi1.api.radio-browser.info", "https://nl1.api.radio-browser.info", "https://at1.api.radio-browser.info", ] print("⚠️ Používám fallback seznam mirrorů.") random.shuffle(mirrors) for mirror in mirrors: try: rb = RadioBrowser(base_url=mirror) rb.countries() # test funkčnosti print(f"✅ Používám RadioBrowser mirror: {mirror}") return rb except Exception as e: print(f"⚠️ Mirror {mirror} selhal: {e}") continue raise RuntimeError("❌ Žádný funkční RadioBrowser mirror nebyl nalezen.") # === Pomocné funkce === def get_workspace_safe_path(output_file): if not output_file: return None if os.path.isabs(output_file): return output_file workspace = os.environ.get('GITHUB_WORKSPACE') if workspace and os.path.exists(workspace): return os.path.join(workspace, output_file) return os.path.abspath(output_file) def generate_safe_output_path(output_file, country_codes): if not output_file: country_string = "_".join(country_codes[:5]).upper() if len(country_codes) > 5: country_string += "_etc" output_file = f"radio_playlist_{country_string}_{datetime.now().strftime('%Y%m%d')}.m3u" safe_path = get_workspace_safe_path(output_file) print(f"📄 Výstupní soubor: {safe_path}") return safe_path def fetch_stations_with_retry(rb, country_code, max_retries=5, initial_backoff=10): retry_count = 0 backoff_time = initial_backoff while retry_count < max_retries: try: return rb.stations_by_countrycode(country_code) except requests.exceptions.HTTPError as e: if e.response.status_code == 429: retry_count += 1 jitter = random.uniform(0.8, 1.2) wait_time = backoff_time * jitter print(f"⏳ Rate limit – čekám {wait_time:.1f}s (pokus {retry_count}/{max_retries})...") time.sleep(wait_time) backoff_time *= 2 else: raise except requests.exceptions.RequestException as e: retry_count += 1 print(f"⚠️ Síťová chyba ({e}). Opakuji za {backoff_time:.1f}s...") time.sleep(backoff_time) backoff_time *= 1.5 raise RateLimitError(f"❌ Nezdařilo se načíst stanice pro {country_code} po {max_retries} pokusech") # === Hlavní logika === def create_multi_country_playlist(country_codes, output_file=None, group_title="Radio Stations", default_logo_url="https://amz.odjezdy.online/xbackbone/VUne9/HilOMeka75.png/raw", use_country_as_group=False): rb = init_radio_browser() # použije autodetekovaný mirror output_file = generate_safe_output_path(output_file, country_codes) try: countries_info = rb.countries() country_dict = {c['iso_3166_1']: c['name'] for c in countries_info} except Exception as e: print(f"❌ Nepodařilo se načíst seznam zemí: {e}") return invalid_codes = [code.upper() for code in country_codes if code.upper() not in country_dict] if invalid_codes: print(f"⚠️ Neplatné kódy zemí: {', '.join(invalid_codes)}") country_codes = [c for c in country_codes if c.upper() in country_dict] if not country_codes: print("❌ Žádné platné kódy zemí. Ukončuji.") sys.exit(1) unique_stations = {} total_found = 0 failed_countries = [] for i, country_code in enumerate(country_codes): country_code = country_code.upper() country_name = country_dict.get(country_code, "Unknown") print(f"[{i+1}/{len(country_codes)}] 📻 {country_name} ({country_code})") try: stations = fetch_stations_with_retry(rb, country_code) valid_stations = [s for s in stations if s.get('url')] if not valid_stations: print(f"⚠️ Žádné stanice pro {country_name}.") continue total_found += len(valid_stations) print(f"✅ {len(valid_stations)} stanic načteno pro {country_name}.") for s in valid_stations: key = f"{s['name'].lower()}_{s['url']}" if key not in unique_stations: s['country_code'] = country_code s['country_name'] = country_name unique_stations[key] = s except RateLimitError as e: print(f"❌ Rate limit pro {country_name}: {e}") failed_countries.append(country_code) except Exception as e: print(f"❌ Chyba při načítání {country_name}: {e}") failed_countries.append(country_code) if i < len(country_codes) - 1: time.sleep(random.uniform(0.5, 1.5)) print(f"\n📊 Celkem nalezeno {total_found} (unikátních {len(unique_stations)}).") if failed_countries: print(f"⚠️ Neúspěšné země: {', '.join(failed_countries)}") with open(output_file, 'w', encoding='utf-8') as f: f.write("#EXTM3U\n") for s in unique_stations.values(): name = s['name'].replace(',', ' ').strip() logo = s.get('favicon') or default_logo_url if not logo or logo == "null": logo = default_logo_url group = s['country_name'] if use_country_as_group else group_title f.write(f'#EXTINF:-1 group-title="{group}" tvg-logo="{logo}",{s["country_code"]} | {name}\n') f.write(f"{s['url']}\n") print(f"🎧 Playlist vytvořen: {output_file} ({len(unique_stations)} stanic)") return output_file def parse_country_codes(countries_str): return [c.strip() for c in countries_str.split(',') if c.strip()] def main(): parser = argparse.ArgumentParser(description="Generate an M3U playlist for multiple countries") parser.add_argument("--countries", type=str, help="Comma-separated list (e.g. 'CZ,SK,DE')") parser.add_argument("country_codes", nargs="*", help="Two-letter country codes") parser.add_argument("-o", "--output", help="Output file path") parser.add_argument("-g", "--group", default="Radio Stations", help="Group title") parser.add_argument("--logo", default="https://amz.odjezdy.online/xbackbone/VUne9/HilOMeka75.png/raw", help="Default logo URL") parser.add_argument("--use-country-as-group", "-ucag", action="store_true", help="Use full country name as group-title") args = parser.parse_args() if args.countries: codes = parse_country_codes(args.countries) elif args.country_codes: codes = args.country_codes else: parser.print_help() sys.exit(1) try: create_multi_country_playlist(codes, args.output, args.group, args.logo, args.use_country_as_group) except KeyboardInterrupt: print("\n🛑 Přerušeno uživatelem.") sys.exit(1) if __name__ == "__main__": main()