From f0854ce3ae8641a1da00499ada30f317a9bdb35c Mon Sep 17 00:00:00 2001 From: mxnticek Date: Sat, 25 Oct 2025 23:14:07 +0200 Subject: [PATCH] Update .forgejo/scripts/radio.py --- .forgejo/scripts/radio.py | 98 ++++++++++++++++++++++++++------------- 1 file changed, 67 insertions(+), 31 deletions(-) diff --git a/.forgejo/scripts/radio.py b/.forgejo/scripts/radio.py index 5b2bddf..dc0b307 100644 --- a/.forgejo/scripts/radio.py +++ b/.forgejo/scripts/radio.py @@ -1,10 +1,9 @@ #!/usr/bin/env python3 """ -Multi-Country Radio M3U Playlist Generator (resilient version) --------------------------------------------------------------- +Multi-Country Radio M3U Playlist Generator (resilient + dynamic mirrors) +----------------------------------------------------------------------- Generuje M3U playlist s radii z více zemí pomocí radio-browser.info API. -Tahle verze automaticky přepíná mezi dostupnými servery (mirrory) -a zvládá výpadky i HTTP 502 chyby. +Automaticky detekuje dostupné servery přes DNS a přepíná mezi nimi. """ import sys @@ -13,6 +12,7 @@ import argparse import time import random import requests +import socket from datetime import datetime from pyradios import RadioBrowser @@ -20,24 +20,60 @@ class RateLimitError(Exception): """Exception raised when rate limit is hit""" pass -# === Nové: automatická inicializace s fallbacky === +# === 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 = [ - "https://de1.api.radio-browser.info", - "https://de2.api.radio-browser.info", - "https://fi1.api.radio-browser.info", - ] + 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"✅ Using RadioBrowser mirror: {mirror}") + print(f"✅ Používám RadioBrowser mirror: {mirror}") return rb except Exception as e: - print(f"⚠️ Mirror {mirror} failed: {e}") + print(f"⚠️ Mirror {mirror} selhal: {e}") continue - raise RuntimeError("❌ No working RadioBrowser mirrors found.") + raise RuntimeError("❌ Žádný funkční RadioBrowser mirror nebyl nalezen.") + +# === Pomocné funkce === def get_workspace_safe_path(output_file): if not output_file: return None @@ -55,7 +91,7 @@ def generate_safe_output_path(output_file, country_codes): 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"Output file will be created at: {safe_path}") + print(f"📄 Výstupní soubor: {safe_path}") return safe_path def fetch_stations_with_retry(rb, country_code, max_retries=5, initial_backoff=10): @@ -69,38 +105,38 @@ def fetch_stations_with_retry(rb, country_code, max_retries=5, initial_backoff=1 retry_count += 1 jitter = random.uniform(0.8, 1.2) wait_time = backoff_time * jitter - print(f"Rate limit exceeded. Waiting for {wait_time:.1f}s (retry {retry_count}/{max_retries})...") + 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"Network error ({e}). Retrying in {backoff_time}s...") + print(f"⚠️ Síťová chyba ({e}). Opakuji za {backoff_time:.1f}s...") time.sleep(backoff_time) backoff_time *= 1.5 - raise RateLimitError(f"Failed to fetch stations for {country_code} after {max_retries} retries") + 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 odolnou inicializaci - + 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"❌ Failed to fetch country list: {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"⚠️ Warning: invalid country codes: {', '.join(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("No valid country codes. Exiting.") + print("❌ Žádné platné kódy zemí. Ukončuji.") sys.exit(1) unique_stations = {} @@ -110,15 +146,15 @@ def create_multi_country_playlist(country_codes, output_file=None, group_title=" 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)}] Fetching {country_name} ({country_code})...") + 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"No stations found for {country_name}.") + print(f"⚠️ Žádné stanice pro {country_name}.") continue total_found += len(valid_stations) - print(f"✅ {len(valid_stations)} stations found for {country_name}.") + 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: @@ -126,17 +162,17 @@ def create_multi_country_playlist(country_codes, output_file=None, group_title=" s['country_name'] = country_name unique_stations[key] = s except RateLimitError as e: - print(f"❌ Rate limit: {e}") + print(f"❌ Rate limit pro {country_name}: {e}") failed_countries.append(country_code) except Exception as e: - print(f"❌ Error fetching {country_name}: {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📊 Found {total_found} total, {len(unique_stations)} unique stations.") + print(f"\n📊 Celkem nalezeno {total_found} (unikátních {len(unique_stations)}).") if failed_countries: - print(f"⚠️ Failed countries: {', '.join(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") @@ -149,7 +185,7 @@ def create_multi_country_playlist(country_codes, output_file=None, 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 created: {output_file} ({len(unique_stations)} stations)") + print(f"🎧 Playlist vytvořen: {output_file} ({len(unique_stations)} stanic)") return output_file def parse_country_codes(countries_str): @@ -178,7 +214,7 @@ def main(): try: create_multi_country_playlist(codes, args.output, args.group, args.logo, args.use_country_as_group) except KeyboardInterrupt: - print("\nProcess interrupted.") + print("\n🛑 Přerušeno uživatelem.") sys.exit(1) if __name__ == "__main__":