Update .forgejo/scripts/radio_country_export.py

This commit is contained in:
Vlastimil Novotny / Ch-last / mxnticek 2025-06-15 01:48:05 +02:00
parent cf366635e0
commit 9d7c1471af

View file

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