#!/usr/bin/env python3 """ Multi-Country Radio M3U Playlist Generator (resilient version) -------------------------------------------------------------- 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. """ import sys import os import argparse import time import random import requests from datetime import datetime from pyradios import RadioBrowser class RateLimitError(Exception): """Exception raised when rate limit is hit""" pass # === Nové: automatická inicializace 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", ] for mirror in mirrors: try: rb = RadioBrowser(base_url=mirror) rb.countries() # test funkčnosti print(f"✅ Using RadioBrowser mirror: {mirror}") return rb except Exception as e: print(f"⚠️ Mirror {mirror} failed: {e}") continue raise RuntimeError("❌ No working RadioBrowser mirrors found.") 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"Output file will be created at: {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 exceeded. Waiting for {wait_time:.1f}s (retry {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...") time.sleep(backoff_time) backoff_time *= 1.5 raise RateLimitError(f"Failed to fetch stations for {country_code} after {max_retries} retries") 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 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}") 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)}") country_codes = [c for c in country_codes if c.upper() in country_dict] if not country_codes: print("No valid country codes. Exiting.") 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)}] Fetching {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}.") continue total_found += len(valid_stations) print(f"✅ {len(valid_stations)} stations found for {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: {e}") failed_countries.append(country_code) except Exception as e: print(f"❌ Error fetching {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.") if failed_countries: print(f"⚠️ Failed countries: {', '.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 created: {output_file} ({len(unique_stations)} stations)") 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("\nProcess interrupted.") sys.exit(1) if __name__ == "__main__": main()