CZ-SK-streams/.forgejo/scripts/radio.py

#!/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.
Podporuje --tvh-compatible formát pro Tvheadend.
"""

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,
                                  tvh_compatible=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
            stream_url = s['url']

            # Blok pro formátování URL pro Tvheadend
            if tvh_compatible:
                # Připravíme metadata - odstraníme mezery a problematické znaky
                service_name = name.replace(" ", "_").replace('"', "'").replace("=", "-").replace(":", "-")
                provider_name = "RadioBrowser"  # Jednoduchý string místo URL
                
                # Escapujeme URL pro bezpečné použití v shellu
                escaped_url = stream_url.replace('"', '\\"').replace("'", "\\'")
                
                url_to_write = (
                    f'pipe://ffmpeg -loglevel fatal -i "{escaped_url}" -vn -acodec copy -flags +global_header '
                    f'-strict -2 -metadata service_provider="{provider_name}" '
                    f'-metadata service_name="{service_name}" -f mpegts -mpegts_service_type digital_radio pipe:1'
                )
            else:
                url_to_write = stream_url

            f.write(f'#EXTINF:-1 group-title="{group}" tvg-logo="{logo}",{s["country_code"]} | {name}\n')
            f.write(f"{url_to_write}\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/xbackend/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")
    parser.add_argument("--tvh-compatible", "-tvh", action="store_true",
                        help="Generovat M3U kompatibilní s Tvheadend (pipe://ffmpeg)")
    
    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, args.tvh_compatible)
    except KeyboardInterrupt:
        print("\n🛑 Přerušeno uživatelem.")
        sys.exit(1)

if __name__ == "__main__":
    main()