Update scripts/generate_playlist.py
All checks were successful
Generate M3U Playlist with Auto-Organization / build-and-organize (push) Successful in 16s

This commit is contained in:
stoney420 2025-06-29 03:51:38 +02:00
parent cc5b952bed
commit 4856bc09ad

View file

@ -1,12 +1,12 @@
#!/usr/bin/env python3 #!/usr/bin/env python3
""" """
IPTV Enhanced Country Detection - Simplified No-Hang Version IPTV Enhanced Country Detection - Updated Version
Uses your existing 3-point analysis but removes network calls that cause hanging Uses 3-point analysis: Channel Name + EPG ID + Logo URL
Then filters to keep only legitimate countries
""" """
import os import os
import shutil import shutil
import re
from datetime import datetime from datetime import datetime
from pathlib import Path from pathlib import Path
@ -15,53 +15,114 @@ script_dir = Path(__file__).parent
root_dir = script_dir.parent root_dir = script_dir.parent
os.chdir(root_dir) os.chdir(root_dir)
def is_streaming_platform(all_text): def detect_country_from_channel_content(channel_name, epg_id="", logo_url="", stream_url=""):
"""Check if channel is from streaming platform.""" """
platforms = [ Enhanced country detection using 3-point analysis
"plex.tv", "pluto.tv", "jmp2.uk/plu-", "sam-", "samsung", Priority: EPG ID > Logo URL > Channel Name > Stream URL
"tubi", "xumo", "roku", "youtube", "daddylive", "drew247tv", """
"moveonjoy", "247tv", "livetv"
]
return any(platform in all_text for platform in platforms)
def detect_traditional_broadcaster_country(channel_name, epg_id="", logo_url="", stream_url=""): # Combine all text for analysis
"""Enhanced 3-point country detection for traditional broadcasters."""
all_text = f"{channel_name.lower()} {epg_id.lower()} {logo_url.lower()} {stream_url.lower()}" all_text = f"{channel_name.lower()} {epg_id.lower()} {logo_url.lower()} {stream_url.lower()}"
channel_lower = channel_name.lower()
# Skip streaming platforms entirely # STEP 1: Check for streaming services first (these go to Uncategorized)
if is_streaming_platform(all_text): streaming_services = [
"plex", "pluto", "tubi", "samsung", "xumo", "stirr", "crackle", "imdb tv",
"daddylive", "drew247", "aixmedia", "moveonjoy", "drewlive24", "udptv",
"a1xs.vip", "zekonew", "forcedtoplay", "cdn1host", "tvpass.org",
"jmp2.uk/plu-", "provider-static.plex.tv", "images.pluto.tv"
]
for service in streaming_services:
if service in all_text:
return "Uncategorized" return "Uncategorized"
# PRIORITY 1: EPG ID detection (most reliable) # STEP 2: EPG ID detection (most reliable) - Enhanced
epg_countries = { epg_patterns = {
".ca": "🇨🇦 Canada", ".us": "🇺🇸 United States", ".uk": "🇬🇧 United Kingdom", ".ca": "🇨🇦 Canada",
".ph": "🇵🇭 Philippines", ".au": "🇦🇺 Australia", ".jp": "🇯🇵 Japan", ".us": "🇺🇸 United States",
".my": "🇲🇾 Malaysia", ".de": "🇩🇪 Germany", ".fr": "🇫🇷 France", ".uk": "🇬🇧 United Kingdom",
".es": "🇪🇸 Spain", ".it": "🇮🇹 Italy", ".br": "🇧🇷 Brazil", ".ph": "🇵🇭 Philippines",
".mx": "🇲🇽 Mexico", ".ar": "🇦🇷 Argentina" ".au": "🇦🇺 Australia",
".jp": "🇯🇵 Japan",
".my": "🇲🇾 Malaysia",
".de": "🇩🇪 Germany",
".fr": "🇫🇷 France",
".es": "🇪🇸 Spain",
".it": "🇮🇹 Italy",
".br": "🇧🇷 Brazil",
".nl": "🇳🇱 Netherlands"
} }
for domain, country in epg_countries.items(): for domain, country in epg_patterns.items():
if domain in epg_id.lower(): if domain in epg_id.lower():
return country return country
# PRIORITY 2: Specific channel fixes from your original code # Enhanced Canadian EPG detection
canadian_epg_patterns = [
"cbc.", "ctv.", "global.", "tsn.", "sportsnet.", "citytv.", "aptn.",
".ab.ca", ".bc.ca", ".mb.ca", ".nb.ca", ".nl.ca", ".ns.ca", ".nt.ca",
".nu.ca", ".on.ca", ".pe.ca", ".qc.ca", ".sk.ca", ".yt.ca",
"cfcn", "cky", "ctfo", "cjoh", "ckws"
]
# Canadian sports channels (TSN series) for pattern in canadian_epg_patterns:
if any(x in channel_lower for x in ["tsn 1", "tsn 2", "tsn 3", "tsn 4", "tsn 5", "tsn1", "tsn2", "tsn3", "tsn4", "tsn5"]): if pattern in epg_id.lower() or pattern in all_text:
return "🇨🇦 Canada" return "🇨🇦 Canada"
# CBC News Toronto (Canadian) # STEP 3: Enhanced specific channel fixes
if "cbc news toronto" in channel_lower: channel_lower = channel_name.lower()
# Enhanced Canadian channels detection
canadian_indicators = [
# TSN variations
"tsn 1", "tsn 2", "tsn 3", "tsn 4", "tsn 5", "tsn1", "tsn2", "tsn3", "tsn4", "tsn5",
# CBC variations
"cbc news", "cbc toronto", "cbc calgary", "cbc vancouver", "cbc winnipeg", "cbc montreal",
# CTV variations
"ctv calgary", "ctv vancouver", "ctv toronto", "ctv winnipeg", "ctv ottawa", "ctv montreal",
"ctv atlantic", "ctv edmonton", "ctv saskatoon", "ctv regina", "ctv kitchener",
# Regional station calls
"cfcn", "cky", "ctfo", "cjoh", "ckws", "cfrn", "cfqc", "ckck", "chch",
# Other Canadian broadcasters
"sportsnet", "global tv", "citytv", "aptn", "omni", "tvo", "télé-québec"
]
for indicator in canadian_indicators:
if indicator in channel_lower:
return "🇨🇦 Canada" return "🇨🇦 Canada"
# Enhanced BBC handling (distinguish US vs UK)
if "bbc" in channel_lower:
# BBC America is US
if "bbc america" in channel_lower:
return "🇺🇸 United States"
# Most other BBC channels are UK
elif any(x in channel_lower for x in ["bbc one", "bbc two", "bbc three", "bbc four",
"bbc news", "bbc iplayer", "bbc scotland", "bbc wales",
"bbc comedy", "bbc drama", "bbc earth"]):
# Check if it's specifically UK version
if not any(x in all_text for x in ["america", ".us", "usa"]):
return "🇬🇧 United Kingdom"
# US channels that were misclassified # US channels that were misclassified
if any(x in channel_lower for x in ["tv land", "tvland", "we tv", "wetv", "all weddings we tv", "cheaters", "cheers", "christmas 365"]): if any(x in channel_lower for x in ["tv land", "tvland", "we tv", "wetv", "all weddings we tv", "cheaters", "cheers", "christmas 365"]):
return "🇺🇸 United States" return "🇺🇸 United States"
# UK shows/channels # Enhanced US network detection
if "come dine with me" in channel_lower: us_networks = [
"cbs", "nbc", "abc", "fox", "cnn", "espn", "hbo", "showtime", "starz", "cinemax",
"mtv", "vh1", "comedy central", "cartoon network", "nickelodeon", "disney channel",
"discovery", "history", "tlc", "hgtv", "food network", "travel channel",
"lifetime", "hallmark", "e!", "bravo", "oxygen", "syfy", "usa network",
"tnt", "tbs", "fx", "fxx", "amc", "ifc", "tcm", "turner classic"
]
for network in us_networks:
if network in channel_lower and not any(x in all_text for x in ["canada", ".ca", "uk", ".uk"]):
return "🇺🇸 United States"
# UK channels (but not BBC America)
if "come dine with me" in channel_lower or "itv" in channel_lower:
return "🇬🇧 United Kingdom" return "🇬🇧 United Kingdom"
# Philippines news channels # Philippines news channels
@ -72,47 +133,121 @@ def detect_traditional_broadcaster_country(channel_name, epg_id="", logo_url="",
if "animax" in channel_lower: if "animax" in channel_lower:
return "🇯🇵 Japan" return "🇯🇵 Japan"
# PRIORITY 3: Logo URL detection (your 3-point analysis) # STEP 4: Logo URL analysis
logo_patterns = { logo_patterns = {
"/canada/": "🇨🇦 Canada", "/ca/": "🇨🇦 Canada", "canada": "🇨🇦 Canada", "🇨🇦 Canada": ["/canada/", "/ca/", "canada.", "canadian"],
"/usa/": "🇺🇸 United States", "/us/": "🇺🇸 United States", "america": "🇺🇸 United States", "🇺🇸 United States": ["/usa/", "/us/", "united-states", "american"],
"/uk/": "🇬🇧 United Kingdom", "/britain/": "🇬🇧 United Kingdom", "british": "🇬🇧 United Kingdom", "🇬🇧 United Kingdom": ["/uk/", "/united-kingdom/", "british", "england"],
"/germany/": "🇩🇪 Germany", "/de/": "🇩🇪 Germany", "🇩🇪 Germany": ["/germany/", "/de/", "german", "deutschland"],
"/france/": "🇫🇷 France", "/fr/": "🇫🇷 France", "🇫🇷 France": ["/france/", "/fr/", "french", "français"],
"/spain/": "🇪🇸 Spain", "/es/": "🇪🇸 Spain", "🇮🇹 Italy": ["/italy/", "/it/", "italian", "italiano"],
"/italy/": "🇮🇹 Italy", "/it/": "🇮🇹 Italy" "🇪🇸 Spain": ["/spain/", "/es/", "spanish", "español"],
"🇳🇱 Netherlands": ["/netherlands/", "/nl/", "dutch", "nederland"],
"🇦🇺 Australia": ["/australia/", "/au/", "australian", "aussie"],
"🇯🇵 Japan": ["/japan/", "/jp/", "japanese", "日本"],
"🇰🇷 South Korea": ["/korea/", "/kr/", "korean", "한국"],
"🇮🇳 India": ["/india/", "/in/", "indian", "भारत"],
"🇧🇷 Brazil": ["/brazil/", "/br/", "brazilian", "brasil"],
"🇲🇽 Mexico": ["/mexico/", "/mx/", "mexican", "méxico"],
"🇦🇷 Argentina": ["/argentina/", "/ar/", "argentinian", "argentina"],
"🇵🇭 Philippines": ["/philippines/", "/ph/", "filipino", "pilipinas"]
} }
for pattern, country in logo_patterns.items(): for country, patterns in logo_patterns.items():
for pattern in patterns:
if pattern in logo_url.lower(): if pattern in logo_url.lower():
return country return country
# PRIORITY 4: Essential broadcaster patterns # STEP 5: Enhanced broadcaster patterns
broadcasters = { broadcaster_patterns = {
"🇨🇦 Canada": ["tsn", "cbc", "ctv", "global", "sportsnet", "w network", "city tv", "aptn"], "🇨🇦 Canada": [
"🇺🇸 United States": ["cbs", "nbc", "abc", "fox", "cnn", "espn", "discovery", "hgtv", "mtv", "c-span", "msnbc", "comedy central"], "cbc", "tsn", "ctv", "global", "sportsnet", "citytv", "aptn", "teletoon", "ytv",
"🇬🇧 United Kingdom": ["bbc", "itv", "sky", "channel 4", "channel 5", "dave", "e4", "film4"], "discovery canada", "history canada", "slice", "w network", "oln", "hgtv canada",
"🇵🇭 Philippines": ["abs-cbn", "gma", "anc", "tv5", "pbo", "net 25"], "food network canada", "showcase", "crave", "super channel", "hollywood suite"
"🇦🇺 Australia": ["abc australia", "nine network", "seven network", "ten network", "sbs"], ],
"🇯🇵 Japan": ["nhk", "fuji tv", "animax", "tbs", "tv asahi"], "🇺🇸 United States": [
"🇲🇾 Malaysia": ["tv1", "tv2", "astro", "rtm", "8tv"], "cbs", "nbc", "abc", "fox", "cnn", "espn", "amc", "mtv", "comedy central",
"🇩🇪 Germany": ["ard", "zdf", "rtl", "sat.1", "pro7", "vox"], "discovery usa", "history usa", "tlc usa", "hgtv usa", "food network usa", "paramount",
"🇫🇷 France": ["tf1", "france 2", "canal+", "m6", "arte"], "nickelodeon usa", "cartoon network usa", "disney usa", "lifetime", "e!", "bravo usa"
"🇪🇸 Spain": ["antena 3", "telecinco", "tve", "la sexta"], ],
"🇮🇹 Italy": ["rai", "mediaset", "canale 5", "la7"], "🇬🇧 United Kingdom": [
"🇧🇷 Brazil": ["globo", "sbt", "record", "band"], "bbc", "itv", "channel 4", "channel 5", "sky", "dave", "really", "yesterday",
"🇲🇽 Mexico": ["televisa", "tv azteca", "canal 5"], "discovery uk", "history uk", "tlc uk", "living", "alibi", "gold", "drama"
"🇷🇺 Russia": ["первый", "россия", "нтв", "тнт"] ],
"🇩🇪 Germany": [
"ard", "zdf", "rtl", "pro7", "sat.1", "vox", "kabel eins", "super rtl", "rtl2",
"discovery germany", "history germany", "tlc germany", "dmax", "sixx", "tele 5"
],
"🇫🇷 France": [
"tf1", "france 2", "france 3", "france 5", "m6", "canal+", "arte", "w9", "tmc",
"discovery france", "history france", "tlc france", "planete+", "ushuaia tv"
],
"🇮🇹 Italy": [
"rai", "canale 5", "italia 1", "rete 4", "la7", "tv8", "nove", "20 mediaset",
"discovery italia", "history italia", "dmax italia", "real time", "giallo"
],
"🇪🇸 Spain": [
"tve", "la 1", "la 2", "antena 3", "cuatro", "telecinco", "la sexta", "nova",
"discovery spain", "history spain", "dmax spain", "mega", "neox", "clan"
],
"🇳🇱 Netherlands": [
"npo", "rtl 4", "rtl 5", "rtl 7", "sbs6", "veronica", "net5", "rtl z",
"discovery netherlands", "history netherlands", "tlc netherlands"
],
"🇦🇺 Australia": [
"abc australia", "nine network", "seven network", "ten", "foxtel",
"discovery australia", "history australia", "lifestyle"
],
"🇯🇵 Japan": [
"nhk", "fuji tv", "tbs", "tv asahi", "tv tokyo", "nippon tv", "animax"
],
"🇰🇷 South Korea": [
"kbs", "mbc", "sbs", "jtbc", "tvn", "ocn"
],
"🇮🇳 India": [
"zee", "star plus", "colors", "sony tv", "& tv", "discovery india"
],
"🇧🇷 Brazil": [
"globo", "sbt", "record", "band", "discovery brasil"
],
"🇲🇽 Mexico": [
"televisa", "tv azteca", "once tv", "discovery mexico"
],
"🇦🇷 Argentina": [
"telefe", "canal 13", "america tv", "discovery argentina"
],
"🇵🇭 Philippines": [
"abs-cbn", "gma", "anc", "tv5", "pba rush"
]
} }
for country, keywords in broadcasters.items(): for country, keywords in broadcaster_patterns.items():
if any(keyword in all_text for keyword in keywords): for keyword in keywords:
if keyword in all_text:
return country return country
return "Uncategorized" return "Uncategorized"
def is_valid_country_group(group_name):
"""Check if group name is a valid country (not a streaming service)"""
valid_countries = [
"🇺🇸 United States", "🇨🇦 Canada", "🇬🇧 United Kingdom", "🇩🇪 Germany",
"🇫🇷 France", "🇮🇹 Italy", "🇪🇸 Spain", "🇳🇱 Netherlands", "🇧🇪 Belgium",
"🇦🇹 Austria", "🇨🇭 Switzerland", "🇸🇪 Sweden", "🇳🇴 Norway", "🇩🇰 Denmark",
"🇫🇮 Finland", "🇵🇱 Poland", "🇨🇿 Czech Republic", "🇭🇺 Hungary", "🇵🇹 Portugal",
"🇬🇷 Greece", "🇷🇴 Romania", "🇧🇬 Bulgaria", "🇭🇷 Croatia", "🇷🇸 Serbia",
"🇦🇺 Australia", "🇯🇵 Japan", "🇰🇷 South Korea", "🇮🇳 India", "🇨🇳 China",
"🇧🇷 Brazil", "🇲🇽 Mexico", "🇦🇷 Argentina", "🇨🇱 Chile", "🇨🇴 Colombia",
"🇷🇺 Russia", "🇹🇷 Turkey", "🇸🇦 Saudi Arabia", "🇦🇪 UAE", "🇪🇬 Egypt",
"🇿🇦 South Africa", "🇳🇬 Nigeria", "🇰🇪 Kenya", "🇮🇱 Israel", "🇹🇭 Thailand",
"🇻🇳 Vietnam", "🇵🇭 Philippines", "🇮🇩 Indonesia", "🇲🇾 Malaysia", "🇸🇬 Singapore"
]
return group_name in valid_countries
def load_channels(): def load_channels():
"""Load channels from channels.txt.""" """Load channels from channels.txt with enhanced parsing for malformed entries."""
if not os.path.exists('channels.txt'): if not os.path.exists('channels.txt'):
print("❌ No channels.txt found") print("❌ No channels.txt found")
return [] return []
@ -130,24 +265,44 @@ def load_channels():
for line in block.strip().split('\n'): for line in block.strip().split('\n'):
if '=' in line: if '=' in line:
key, value = line.split('=', 1) key, value = line.split('=', 1)
channel_data[key.strip()] = value.strip() key = key.strip()
value = value.strip()
if channel_data.get('Stream name'): # Clean up malformed values (fix the quote issues we saw)
if key == "Stream name" and value.startswith('"') and value.count('"') > 2:
# Handle malformed entries like: ".AB.ca",.AB.ca" tvg-logo=...
# Extract just the actual channel name
parts = value.split(',')
if len(parts) > 1:
value = parts[-1].strip().strip('"')
channel_data[key] = value
# Only add channels with valid stream names
if channel_data.get('Stream name') and len(channel_data.get('Stream name', '')) > 1:
channels.append(channel_data) channels.append(channel_data)
print(f"✅ Loaded {len(channels)} channels") print(f"✅ Loaded {len(channels)} channels (with enhanced parsing)")
return channels return channels
except Exception as e: except Exception as e:
print(f"❌ Error loading channels: {e}") print(f"❌ Error loading channels: {e}")
return [] return []
def reorganize_channels(channels): def reorganize_channels(channels):
"""Reorganize channels by country (traditional only).""" """Enhanced reorganization with 3-point analysis."""
print("📺 Organizing traditional broadcasters by country...") print("🔍 Enhanced Country Detection with 3-Point Analysis")
print("📊 Analyzing: Channel Name + EPG ID + Logo URL")
print("-" * 60)
changes = 0 changes = 0
stats = {} stats = {
'country_detected': 0,
'sent_to_uncategorized': 0,
'kept_existing_country': 0
}
country_counts = {}
for channel in channels: for channel in channels:
old_group = channel.get('Group', 'Uncategorized') old_group = channel.get('Group', 'Uncategorized')
@ -156,24 +311,46 @@ def reorganize_channels(channels):
logo = channel.get('Logo', '') logo = channel.get('Logo', '')
stream_url = channel.get('Stream URL', '') stream_url = channel.get('Stream URL', '')
new_group = detect_traditional_broadcaster_country(stream_name, epg_id, logo, stream_url) # Detect country using enhanced 3-point analysis
detected_country = detect_country_from_channel_content(stream_name, epg_id, logo, stream_url)
if old_group != new_group: # Decide final group
if new_group == "Uncategorized": if is_valid_country_group(old_group) and detected_country != "Uncategorized":
print(f"📱 Platform: '{stream_name}' → Uncategorized") # Keep existing valid country
final_group = old_group
stats['kept_existing_country'] += 1
elif detected_country != "Uncategorized":
# Use detected country
final_group = detected_country
stats['country_detected'] += 1
if old_group != detected_country:
print(f"🔍 Fixed: '{stream_name}' {old_group}{detected_country}")
changes += 1
else: else:
print(f"📺 Fixed: '{stream_name}' {old_group}{new_group}") # Send to Uncategorized
channel['Group'] = new_group final_group = "Uncategorized"
stats['sent_to_uncategorized'] += 1
if old_group != "Uncategorized":
print(f"📱 Platform: '{stream_name}' → Uncategorized")
changes += 1 changes += 1
stats[new_group] = stats.get(new_group, 0) + 1 channel['Group'] = final_group
country_counts[final_group] = country_counts.get(final_group, 0) + 1
print(f"\n✅ Changes made: {changes}") print(f"\n📊 PROCESSING RESULTS:")
print(f"📺 Traditional broadcasters: {sum(v for k, v in stats.items() if k != 'Uncategorized')}") print(f"✅ Changes made: {changes}")
print(f"📱 Streaming/Unknown: {stats.get('Uncategorized', 0)}") print(f"🔍 Country detected: {stats['country_detected']}")
print(f"✅ Kept existing countries: {stats['kept_existing_country']}")
print(f"📱 Sent to Uncategorized: {stats['sent_to_uncategorized']}")
print(f"\n🌍 FINAL GROUP DISTRIBUTION:")
sorted_countries = sorted(country_counts.items(), key=lambda x: (x[0] == "Uncategorized", -x[1]))
for country, count in sorted_countries:
print(f" {country}: {count} channels")
return channels return channels
def save_channels(channels): def save_channels(channels):
"""Save channels to file.""" """Save channels to file."""
# Backup # Backup
@ -199,6 +376,7 @@ def save_channels(channels):
print(f"❌ Save error: {e}") print(f"❌ Save error: {e}")
return False return False
def generate_m3u(channels): def generate_m3u(channels):
"""Generate M3U playlist.""" """Generate M3U playlist."""
try: try:
@ -226,38 +404,23 @@ def generate_m3u(channels):
print(f"❌ M3U error: {e}") print(f"❌ M3U error: {e}")
return False return False
def create_logs():
"""Create logs directory and simple log."""
try:
os.makedirs('logs', exist_ok=True)
with open('logs/playlist_update.log', 'w', encoding='utf-8') as f:
f.write(f"Enhanced IPTV Playlist Generator Log\n")
f.write(f"Generated: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}\n")
f.write(f"Status: Completed successfully\n")
print("✅ Created logs")
except Exception as e:
print(f"⚠️ Log creation failed: {e}")
def main(): def main():
"""Main function.""" """Main function."""
print("📺 Enhanced Country Detection - No Network Calls") print("🎯 Enhanced IPTV Country Detection - 3-Point Analysis")
print("=" * 60) print("=" * 70)
print("🔍 Analyzes: Channel Name + EPG ID + Logo URL")
# Create logs directory print("🎯 Filters: Only countries remain, streaming services → Uncategorized")
create_logs() print("=" * 70)
channels = load_channels() channels = load_channels()
if not channels: if not channels:
print(" No channels to process") return False
return True
# Reorganize # Enhanced reorganization
channels = reorganize_channels(channels) channels = reorganize_channels(channels)
# Sort: Countries first, then Uncategorized last # Sort: Countries first (alphabetically), then Uncategorized last
channels.sort(key=lambda x: ( channels.sort(key=lambda x: (
"zzz" if x.get('Group') == "Uncategorized" else x.get('Group', ''), "zzz" if x.get('Group') == "Uncategorized" else x.get('Group', ''),
x.get('Stream name', '') x.get('Stream name', '')
@ -278,13 +441,15 @@ def main():
except: except:
pass pass
print("\n🎉 COMPLETED!") print("\n🎉 ENHANCED PROCESSING COMPLETE!")
print("✅ Traditional broadcasters organized by country") print("✅ 3-point analysis applied to all channels")
print("✅ All streaming platforms → Uncategorized") print("✅ Countries detected from EPG ID, Logo URL, and Channel Names")
print("✅ No network calls - fast execution") print("✅ Streaming services filtered to Uncategorized")
print("✅ Clean country-organized playlist generated")
return True return True
if __name__ == "__main__": if __name__ == "__main__":
success = main() success = main()
exit(0 if success else 1) exit(0 if success else 1)