#!/usr/bin/env python3 """ Multi-Country Radio M3U Playlist Generator ---------------------------------------- This script generates an M3U playlist file containing radio stations from multiple specified countries using the radio-browser.info API via the pyradios library. Features include duplicate removal, default logos, rate limit handling, and country-based grouping. """ 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 def get_workspace_safe_path(output_file): """ Ensure output path is safe for CI/CD environments. If GITHUB_WORKSPACE is set and output_file is relative, use workspace as base. """ if not output_file: return None # If already absolute path, return as-is if os.path.isabs(output_file): return output_file # Check if we're in a CI environment with GITHUB_WORKSPACE workspace = os.environ.get('GITHUB_WORKSPACE') if workspace and os.path.exists(workspace): return os.path.join(workspace, output_file) # Fallback to current working directory for relative paths return os.path.abspath(output_file) def generate_safe_output_path(output_file, country_codes): """Generate a safe output path for the playlist file""" 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" # Make sure the path is workspace-safe 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): """ Fetch stations with retry logic for rate limiting Args: rb: RadioBrowser instance country_code: Country code to fetch stations for max_retries: Maximum number of retry attempts initial_backoff: Initial backoff time in seconds Returns: List of stations """ 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: # Too Many Requests retry_count += 1 jitter = random.uniform(0.8, 1.2) # Add some randomness to prevent thundering herd wait_time = backoff_time * jitter print(f"Rate limit exceeded. Waiting for {wait_time:.1f} seconds before retry {retry_count}/{max_retries}...") time.sleep(wait_time) # Exponential backoff: double the wait time for next attempt backoff_time *= 2 else: # For other HTTP errors, raise the exception raise # If we've exhausted all retries raise RateLimitError(f"Failed to fetch stations for {country_code} after {max_retries} retries due to rate limiting") 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): """ Create an M3U playlist for radio stations from multiple specified countries. Args: country_codes (list): List of two-letter country codes (ISO 3166-1 alpha-2) output_file (str, optional): Path to output file. If None, generates a default name group_title (str, optional): Group title for stations in the playlist default_logo_url (str): Default logo URL to use when a station has no logo use_country_as_group (bool): If True, use full country name as the group-title Returns: str: Path to the created playlist file """ # Initialize RadioBrowser client rb = RadioBrowser() # Generate safe output filename output_file = generate_safe_output_path(output_file, country_codes) # Get all country information for validation and display try: countries_info = rb.countries() country_dict = {country['iso_3166_1']: country['name'] for country in countries_info} except requests.exceptions.HTTPError as e: if e.response.status_code == 429: # Too Many Requests print("Rate limit exceeded when fetching country list. Waiting 30 seconds and retrying...") time.sleep(30) countries_info = rb.countries() country_dict = {country['iso_3166_1']: country['name'] for country in countries_info} else: raise # Validate country codes invalid_codes = [code.upper() for code in country_codes if code.upper() not in country_dict] if invalid_codes: print(f"Warning: The following country codes are invalid: {', '.join(invalid_codes)}") country_codes = [code for code in country_codes if code.upper() in country_dict] if not country_codes: print("No valid country codes provided. Exiting.") sys.exit(1) # Dictionary to track unique stations and avoid duplicates # We'll use a combination of station name and URL as the key unique_stations = {} total_found = 0 failed_countries = [] # Process each country and collect stations for i, country_code in enumerate(country_codes): country_code = country_code.upper() country_name = country_dict.get(country_code, "Unknown Country") print(f"[{i+1}/{len(country_codes)}] Fetching stations for {country_name} ({country_code})...") try: # Use our retry function instead of direct API call 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} ({country_code}).") continue total_found += len(valid_stations) print(f"Found {len(valid_stations)} stations for {country_name}.") # Add each station to our unique stations dictionary for station in valid_stations: # Create a unique key based on the station name and URL # This helps identify genuinely unique stations station_key = f"{station['name'].lower()}_{station['url']}" if station_key not in unique_stations: # Add country code and name to the station data station['country_code'] = country_code station['country_name'] = country_name unique_stations[station_key] = station except RateLimitError as e: print(f"Error: {str(e)}") failed_countries.append(country_code) continue except Exception as e: print(f"Error fetching stations for {country_name}: {str(e)}") failed_countries.append(country_code) continue # Add a small delay between countries to avoid hitting rate limits if i < len(country_codes) - 1: time.sleep(random.uniform(0.5, 1.5)) print(f"Found {total_found} total stations, {len(unique_stations)} unique after removing duplicates.") if failed_countries: print(f"Failed to process these countries: {', '.join(failed_countries)}") # In CI environment, don't ask for user input - just log and continue if os.environ.get('GITHUB_WORKSPACE') or os.environ.get('CI'): print("Running in CI environment - skipping retry prompt") else: retry_option = input("Would you like to retry failed countries? (y/n): ") if retry_option.lower() == 'y': print("Retrying failed countries...") # Wait a bit longer before retrying failed countries time.sleep(30) return create_multi_country_playlist(failed_countries, f"retry_{os.path.basename(output_file)}", group_title, default_logo_url, use_country_as_group) # Create M3U playlist with open(output_file, 'w', encoding='utf-8') as f: # Write M3U header f.write("#EXTM3U\n") # Add each unique station to the playlist for station in unique_stations.values(): # Clean station name to avoid issues with special characters station_name = station['name'].replace(',', ' ').strip() # Create the formatted station name with country code prefix country_code = station['country_code'] formatted_name = f"{country_code} | {station_name}" # Get station logo if available, otherwise use the default logo logo_url = station.get('favicon', '') if not logo_url or logo_url == "null": logo_url = default_logo_url # Determine which group title to use if use_country_as_group: station_group = station['country_name'] else: station_group = group_title # Write the entry in the requested format f.write(f'#EXTINF:-1 group-title="{station_group}" tvg-logo="{logo_url}",{formatted_name}\n') f.write(f"{station['url']}\n") print(f"Playlist created: {output_file}") print(f"Total unique stations in playlist: {len(unique_stations)}") return output_file def parse_country_codes(countries_str): """Parse a comma-separated string of country codes into a list""" # Split by comma, strip whitespace, and filter out empty strings return [code.strip() for code in countries_str.split(',') if code.strip()] def main(): """Main function to handle command line arguments""" parser = argparse.ArgumentParser(description="Generate an M3U playlist for radio stations from multiple countries") # Add a different way to specify countries as a comma-separated string parser.add_argument("--countries", type=str, help="Comma-separated list of two-letter country codes (e.g., 'CZ, SK, DE, FR, US')") # Keep the positional argument for backward compatibility, but make it optional parser.add_argument("country_codes", nargs="*", help="Two-letter country codes (e.g., US CZ DE FR)") parser.add_argument("-o", "--output", help="Output file path") parser.add_argument("-g", "--group", default="Radio Stations", help="Group title for stations in the playlist") parser.add_argument("--logo", default="https://amz.odjezdy.online/xbackbone/VUne9/HilOMeka75.png/raw", help="Default logo URL for stations without logos") parser.add_argument("--use-country-as-group", "-ucag", action="store_true", help="Use full country name as the group-title instead of a single group") args = parser.parse_args() # Determine which method of specifying countries was used if args.countries: country_codes = parse_country_codes(args.countries) elif args.country_codes: country_codes = args.country_codes else: parser.print_help() print("\nError: You must specify country codes either with positional arguments or with --countries") sys.exit(1) try: create_multi_country_playlist(country_codes, args.output, args.group, args.logo, args.use_country_as_group) except KeyboardInterrupt: print("\nProcess interrupted by user. Exiting...") sys.exit(1) if __name__ == "__main__": main()