248 lines
No EOL
10 KiB
Python
248 lines
No EOL
10 KiB
Python
#!/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): # <<< ZMĚNA: Přidán nový argument
|
||
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']
|
||
|
||
# <<< ZMĚNA: 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 = stream_url # Poskytovatel je URL, jak je v textu
|
||
|
||
url_to_write = (
|
||
f"pipe://ffmpeg -loglevel fatal -i {stream_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
|
||
# <<< KONEC ZMĚNY
|
||
|
||
f.write(f'#EXTINF:-1 group-title="{group}" tvg-logo="{logo}",{s["country_code"]} | {name}\n')
|
||
f.write(f"{url_to_write}\n") # Použijeme upravenou nebo původní URL
|
||
|
||
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")
|
||
|
||
# <<< ZMĚNA: Přidán argument pro TVH kompatibilitu
|
||
parser.add_argument("--tvh-compatible", "-tvh", action="store_true",
|
||
help="Generovat M3U kompatibilní s Tvheadend (pipe://ffmpeg)")
|
||
# <<< KONEC ZMĚNY
|
||
|
||
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:
|
||
# <<< ZMĚNA: Předání nového argumentu
|
||
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() |