Update scripts/generate_playlist.py
All checks were successful
📺 Generate M3U Playlist / build (push) Successful in 1m15s

This commit is contained in:
stoney420 2025-06-27 18:55:40 +02:00
parent 1a38bef290
commit 602021a94a

View file

@ -59,174 +59,41 @@ def load_group_overrides():
def detect_country_from_channel(channel_name, epg_id="", logo_url=""):
"""
Auto-detect country from channel information using cascading approach.
Returns the detected country group or 'Uncategorized' if no match found.
Simple country detection that will definitely work.
"""
# Convert to lowercase for easier matching
name_lower = channel_name.lower()
epg_lower = epg_id.lower()
# Normalize inputs for better matching
channel_lower = channel_name.lower().strip()
epg_lower = epg_id.lower().strip()
logo_lower = logo_url.lower().strip()
log_message(f"Detecting country for: '{channel_name}' (EPG: '{epg_id}')", "DEBUG")
# 1. BROADCASTER NAMES (Most reliable)
broadcaster_country = {
# US Major Networks
"cbs": "🇺🇸 United States", "nbc": "🇺🇸 United States",
"abc": "🇺🇸 United States", "fox": "🇺🇸 United States",
"espn": "🇺🇸 United States", "cnn": "🇺🇸 United States",
"hbo": "🇺🇸 United States", "mtv": "🇺🇸 United States",
"discovery": "🇺🇸 United States", "cartoon network": "🇺🇸 United States",
"showtime": "🇺🇸 United States", "starz": "🇺🇸 United States",
"tnt": "🇺🇸 United States", "tbs": "🇺🇸 United States",
# UK Detection
if "sky" in name_lower or ".uk" in epg_lower or "british" in name_lower or "bbc" in name_lower or "itv" in name_lower:
log_message(f"Detected UK for: {channel_name}", "INFO")
return "🇬🇧 United Kingdom"
# UK Networks
"bbc": "🇬🇧 United Kingdom", "itv": "🇬🇧 United Kingdom",
"channel 4": "🇬🇧 United Kingdom", "sky": "🇬🇧 United Kingdom",
"e4": "🇬🇧 United Kingdom", "film4": "🇬🇧 United Kingdom",
"more4": "🇬🇧 United Kingdom", "dave": "🇬🇧 United Kingdom",
# US Detection
if "usa" in name_lower or "us " in name_lower or ".us" in epg_lower or "america" in name_lower or "cnn" in name_lower or "espn" in name_lower or "fox" in name_lower:
log_message(f"Detected US for: {channel_name}", "INFO")
return "🇺🇸 United States"
# Canadian Networks
"cbc": "🇨🇦 Canada", "ctv": "🇨🇦 Canada", "global": "🇨🇦 Canada",
"tvo": "🇨🇦 Canada", "aptn": "🇨🇦 Canada",
# Canada Detection
if "canada" in name_lower or "cbc" in name_lower or ".ca" in epg_lower or "ctv" in name_lower:
log_message(f"Detected Canada for: {channel_name}", "INFO")
return "🇨🇦 Canada"
# German Networks
"ard": "🇩🇪 Germany", "zdf": "🇩🇪 Germany", "rtl": "🇩🇪 Germany",
"sat.1": "🇩🇪 Germany", "pro7": "🇩🇪 Germany", "vox": "🇩🇪 Germany",
"kabel": "🇩🇪 Germany", "n24": "🇩🇪 Germany",
# Germany Detection
if "german" in name_lower or ".de" in epg_lower or "ard" in name_lower or "zdf" in name_lower:
log_message(f"Detected Germany for: {channel_name}", "INFO")
return "🇩🇪 Germany"
# French Networks
"tf1": "🇫🇷 France", "france 2": "🇫🇷 France", "m6": "🇫🇷 France",
"canal+": "🇫🇷 France", "arte": "🇫🇷 France",
# France Detection
if "france" in name_lower or ".fr" in epg_lower or "tf1" in name_lower:
log_message(f"Detected France for: {channel_name}", "INFO")
return "🇫🇷 France"
# Spanish Networks
"tve": "🇪🇸 Spain", "antena 3": "🇪🇸 Spain", "telecinco": "🇪🇸 Spain",
"cuatro": "🇪🇸 Spain", "la sexta": "🇪🇸 Spain",
# Italian Networks
"rai": "🇮🇹 Italy", "mediaset": "🇮🇹 Italy", "canale 5": "🇮🇹 Italy",
"italia 1": "🇮🇹 Italy", "rete 4": "🇮🇹 Italy",
# Other Countries
"globo": "🇧🇷 Brazil", "band": "🇧🇷 Brazil", "sbt": "🇧🇷 Brazil",
"televisa": "🇲🇽 Mexico", "tv azteca": "🇲🇽 Mexico",
"al jazeera": "🇸🇦 Arabic", "mbc": "🇸🇦 Arabic", "lbc": "🇸🇦 Arabic",
"rt": "🇷🇺 Russia", "channel one": "🇷🇺 Russia",
"cctv": "🇨🇳 China", "phoenix": "🇨🇳 China",
"nhk": "🇯🇵 Japan", "fuji": "🇯🇵 Japan",
"kbs": "🇰🇷 South Korea", "sbs": "🇰🇷 South Korea", "mbc": "🇰🇷 South Korea",
"abc au": "🇦🇺 Australia", "seven": "🇦🇺 Australia", "nine": "🇦🇺 Australia",
"npo": "🇳🇱 Netherlands", "rtl nl": "🇳🇱 Netherlands"
}
# Check for exact broadcaster matches
for broadcaster, country in broadcaster_country.items():
if broadcaster in channel_lower:
return country
# 2. COUNTRY NAME PATTERNS (Very reliable)
country_patterns = {
"🇺🇸 United States": [
"usa", "united states", "america", "american", " us ", "us:", "(us)", "us hd"
],
"🇬🇧 United Kingdom": [
" uk ", "uk:", "(uk)", "united kingdom", "britain", "british", "england", "english"
],
"🇨🇦 Canada": [
"canada", "canadian", " ca ", "ca:", "(ca)"
],
"🇩🇪 Germany": [
"germany", "german", "deutschland", "deutsch", " de ", "de:", "(de)"
],
"🇫🇷 France": [
"france", "french", "français", " fr ", "fr:", "(fr)"
],
"🇪🇸 Spain": [
"spain", "spanish", "españa", "español", " es ", "es:", "(es)"
],
"🇮🇹 Italy": [
"italy", "italian", "italia", "italiano", " it ", "it:", "(it)"
],
"🇧🇷 Brazil": [
"brazil", "brazilian", "brasil", "português", " br ", "br:", "(br)"
],
"🇲🇽 Mexico": [
"mexico", "mexican", "méxico", " mx ", "mx:", "(mx)"
],
"🇦🇺 Australia": [
"australia", "australian", "aussie", " au ", "au:", "(au)"
],
"🇳🇱 Netherlands": [
"netherlands", "dutch", "holland", "nederland", " nl ", "nl:", "(nl)"
],
"🇷🇺 Russia": [
"russia", "russian", "россия", " ru ", "ru:", "(ru)"
],
"🇨🇳 China": [
"china", "chinese", "中国", " cn ", "cn:", "(cn)"
],
"🇯🇵 Japan": [
"japan", "japanese", "日本", " jp ", "jp:", "(jp)"
],
"🇰🇷 South Korea": [
"korea", "korean", "한국", " kr ", "kr:", "(kr)", "south korea"
],
"🇸🇦 Arabic": [
"arabic", "arab", "middle east", "العربية", "al ", "aljazeera"
],
"🇮🇳 India": [
"india", "indian", "hindi", "bollywood", "zee", "star plus"
],
"🇵🇹 Portugal": [
"portugal", "portuguese", " pt ", "pt:", "(pt)"
],
"🇹🇷 Turkey": [
"turkey", "turkish", "türkiye", " tr ", "tr:", "(tr)"
]
}
# Check channel name for country patterns
for country, patterns in country_patterns.items():
for pattern in patterns:
if pattern in channel_lower:
return country
# 3. EPG ID ANALYSIS (Good for country codes)
epg_country_map = {
"🇺🇸 United States": [".us", "usa.", ".com"],
"🇬🇧 United Kingdom": [".uk", ".gb", "british"],
"🇨🇦 Canada": [".ca", "canada."],
"🇩🇪 Germany": [".de", "german."],
"🇫🇷 France": [".fr", "france."],
"🇪🇸 Spain": [".es", "spain."],
"🇮🇹 Italy": [".it", "italy."],
"🇧🇷 Brazil": [".br", "brazil."],
"🇲🇽 Mexico": [".mx", "mexico."],
"🇦🇺 Australia": [".au", "australia."],
"🇳🇱 Netherlands": [".nl", "netherlands."],
"🇷🇺 Russia": [".ru", "russia."],
"🇨🇳 China": [".cn", "china."],
"🇯🇵 Japan": [".jp", "japan."],
"🇰🇷 South Korea": [".kr", "korea."],
"🇮🇳 India": [".in", "india."],
"🇵🇹 Portugal": [".pt", "portugal."],
"🇹🇷 Turkey": [".tr", "turkey."]
}
# Check EPG ID for country indicators
if epg_id:
for country, patterns in epg_country_map.items():
for pattern in patterns:
if pattern in epg_lower:
return country
# 4. LOGO URL ANALYSIS (Sometimes helpful)
if logo_url:
for country, patterns in epg_country_map.items():
for pattern in patterns:
if pattern in logo_lower:
return country
# If no match found, return Uncategorized
# No match found
log_message(f"No country detected for: {channel_name}", "DEBUG")
return "Uncategorized"
def apply_auto_country_detection(channel, group_overrides, settings):
@ -351,53 +218,64 @@ def remove_duplicates(channels, settings):
def update_existing_channels_with_country_detection():
"""Re-process existing channels.txt to apply country detection to old channels."""
if not os.path.exists(CHANNELS_FILE):
log_message("No channels.txt file found", "WARNING")
return
settings = load_settings()
group_overrides = load_group_overrides()
if not settings.get('auto_detect_country', True):
return
log_message("Re-detecting countries for existing channels...", "INFO")
log_message("Starting to re-detect countries for ALL existing channels...", "INFO")
# Read existing channels
with open(CHANNELS_FILE, 'r', encoding='utf-8') as f:
content = f.read()
log_message(f"Read {len(content)} characters from channels.txt", "DEBUG")
channel_blocks = re.split(r'\n\s*\n+', content.strip())
log_message(f"Found {len(channel_blocks)} channel blocks", "INFO")
updated_channels = []
changes_made = 0
for block in channel_blocks:
for i, block in enumerate(channel_blocks):
if block.strip():
channel = parse_channel_block(block)
if channel:
old_group = channel.get('Group', 'Uncategorized')
stream_name = channel.get('Stream name', 'Unknown')
epg_id = channel.get('EPG id', '')
# Apply auto-detection (this will update the group)
channel = apply_auto_country_detection(channel, group_overrides, settings)
log_message(f"Processing channel {i+1}: '{stream_name}' (currently in '{old_group}')", "DEBUG")
new_group = channel.get('Group', 'Uncategorized')
# Force apply auto-detection regardless of current group
detected_country = detect_country_from_channel(stream_name, epg_id, "")
# Log if group changed
if old_group != new_group:
# Always update if we detected something specific
if detected_country != "Uncategorized":
channel['Group'] = detected_country
changes_made += 1
log_message(f"Updated: '{channel.get('Stream name')}' from '{old_group}' to '{new_group}'", "INFO")
log_message(f"CHANGED: '{stream_name}' from '{old_group}' to '{detected_country}'", "INFO")
else:
log_message(f"NO CHANGE: '{stream_name}' stays as '{old_group}'", "DEBUG")
updated_channels.append(channel)
# If changes were made, rewrite the file
if changes_made > 0:
log_message(f"Updating {changes_made} channels with new country detection...", "INFO")
# Always rewrite the file if we have channels
if updated_channels:
log_message(f"Rewriting channels.txt with {len(updated_channels)} channels ({changes_made} changes made)", "INFO")
# Backup original
# Create backup
backup_name = f"{CHANNELS_FILE}.backup.{datetime.now().strftime('%Y%m%d_%H%M%S')}"
try:
import shutil
shutil.copy2(CHANNELS_FILE, backup_name)
log_message(f"Created backup: {backup_name}", "INFO")
except Exception as e:
log_message(f"Could not create backup: {e}", "WARNING")
# Write updated channels
try:
with open(CHANNELS_FILE, 'w', encoding='utf-8') as f:
for i, channel in enumerate(updated_channels):
if i > 0:
@ -406,9 +284,11 @@ def update_existing_channels_with_country_detection():
block_content = convert_to_channels_txt_block(channel)
f.write(block_content)
log_message(f"Successfully updated {changes_made} channels with country detection", "INFO")
log_message(f"Successfully rewrote channels.txt with country detection", "INFO")
except Exception as e:
log_message(f"ERROR writing channels.txt: {e}", "ERROR")
else:
log_message("No channels needed country detection updates", "INFO")
log_message("No channels found to update", "WARNING")
def process_import():
"""Process bulk import file."""
@ -523,6 +403,9 @@ def generate_playlist():
settings = load_settings()
group_overrides = load_group_overrides()
log_message(f"Settings loaded: {settings}", "INFO")
log_message(f"Group overrides loaded: {group_overrides}", "INFO")
# FIRST: Update existing channels with country detection
update_existing_channels_with_country_detection()