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

This commit is contained in:
stoney420 2025-06-29 02:47:11 +02:00
parent 0f904c6e78
commit 2943aa881f

View file

@ -1,219 +1,106 @@
#!/usr/bin/env python3 #!/usr/bin/env python3
""" """
IPTV Country + Platform Organizer IPTV Traditional Broadcasters Only - Simplified Version
Groups channels by country first, then platform within country Clean, minimal code with no redundancy
Example: 🇺🇸 USA, 🇺🇸 USA - Plex, 🇺🇸 USA - Pluto, 🇨🇦 Canada, 🇨🇦 Canada - Plex
""" """
import os import os
import sys
import shutil import shutil
from datetime import datetime from datetime import datetime
from pathlib import Path from pathlib import Path
# FIXED: Ensure we're in the right directory # Ensure correct directory
script_dir = Path(__file__).parent script_dir = Path(__file__).parent
root_dir = script_dir.parent root_dir = script_dir.parent
os.chdir(root_dir) os.chdir(root_dir)
def detect_country_and_platform(channel_name, epg_id="", logo_url="", stream_url=""): def is_streaming_platform(all_text):
"""Enhanced country + platform detection.""" """Check if channel is from streaming platform (simplified)."""
all_text = f"{channel_name.lower().strip()} {epg_id.lower().strip()} {logo_url.lower().strip()} {stream_url.lower().strip()}" platforms = ["plex.tv", "pluto.tv", "jmp2.uk/plu-", "sam-", "samsung", "tubi", "xumo", "roku", "youtube"]
channel_lower = channel_name.lower() return any(platform in all_text for platform in platforms)
# STEP 1: Detect the country first
country = detect_base_country(channel_name, epg_id, logo_url, stream_url)
# STEP 2: Detect platform
platform = detect_platform(all_text)
# STEP 3: Combine country + platform
if platform:
return f"{country} - {platform}"
else:
return country
def detect_base_country(channel_name, epg_id="", logo_url="", stream_url=""): def detect_traditional_broadcaster_country(channel_name, epg_id="", logo_url="", stream_url=""):
"""Detect the base country of the channel.""" """Detect country for traditional broadcasters only."""
all_text = f"{channel_name.lower().strip()} {epg_id.lower().strip()} {logo_url.lower().strip()} {stream_url.lower().strip()}" all_text = f"{channel_name.lower()} {epg_id.lower()} {logo_url.lower()} {stream_url.lower()}"
channel_lower = channel_name.lower() channel_lower = channel_name.lower()
# PRIORITY 1: EPG ID suffix detection (most reliable) # Skip streaming platforms
if ".ca" in epg_id.lower(): if is_streaming_platform(all_text):
return "🇨🇦 Canada" return "Uncategorized"
elif ".us" in epg_id.lower():
return "🇺🇸 United States"
elif ".uk" in epg_id.lower():
return "🇬🇧 United Kingdom"
elif ".ph" in epg_id.lower():
return "🇵🇭 Philippines"
elif ".au" in epg_id.lower():
return "🇦🇺 Australia"
elif ".jp" in epg_id.lower():
return "🇯🇵 Japan"
elif ".my" in epg_id.lower():
return "🇲🇾 Malaysia"
elif ".de" in epg_id.lower():
return "🇩🇪 Germany"
elif ".fr" in epg_id.lower():
return "🇫🇷 France"
# PRIORITY 2: Specific channel fixes for misclassified channels # EPG ID detection (most reliable)
epg_countries = {
# Canadian channels ".ca": "🇨🇦 Canada", ".us": "🇺🇸 United States", ".uk": "🇬🇧 United Kingdom",
if any(x in channel_lower for x in ["tsn 1", "tsn 2", "tsn 3", "tsn 4", "tsn 5", "tsn1", "tsn2", "tsn3", "tsn4", "tsn5"]): ".ph": "🇵🇭 Philippines", ".au": "🇦🇺 Australia", ".jp": "🇯🇵 Japan",
return "🇨🇦 Canada" ".my": "🇲🇾 Malaysia", ".de": "🇩🇪 Germany", ".fr": "🇫🇷 France",
if any(x in channel_lower for x in ["cbc news", "cbc news toronto", "cbc news british columbia"]): ".es": "🇪🇸 Spain", ".it": "🇮🇹 Italy", ".br": "🇧🇷 Brazil"
return "🇨🇦 Canada"
if "w network" in channel_lower and "canada" in all_text:
return "🇨🇦 Canada"
# 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"]):
return "🇺🇸 United States"
if any(x in channel_lower for x in ["cbs", "nbc", "abc", "fox news", "cnn", "espn", "discovery channel", "cartoon network"]):
return "🇺🇸 United States"
# UK channels
if "come dine with me" in channel_lower:
return "🇬🇧 United Kingdom"
if any(x in channel_lower for x in ["bbc", "itv", "sky news", "channel 4", "channel 5"]):
return "🇬🇧 United Kingdom"
# Philippines channels
if any(x in channel_lower for x in ["anc global", "anc ph"]):
return "🇵🇭 Philippines"
# Malaysia channels
if "malaysia" in channel_lower or "bein sports 1 malaysia" in channel_lower:
return "🇲🇾 Malaysia"
# Japan channels
if "animax" in channel_lower:
return "🇯🇵 Japan"
# PRIORITY 3: Special platform handling with country detection
# Pluto TV regional detection
if "pluto.tv" in all_text or "images.pluto.tv" in all_text or "jmp2.uk/plu-" in all_text:
pluto_countries = {
"cbc news toronto": "🇨🇦 Canada",
"cbc news british columbia": "🇨🇦 Canada",
"come dine with me": "🇬🇧 United Kingdom"
}
for channel_pattern, country in pluto_countries.items():
if channel_pattern in channel_lower:
return country
# Default Pluto TV to US
return "🇺🇸 United States"
# Samsung TV Plus regional detection
if "samsung" in all_text or "sam-" in stream_url:
if "cabc" in epg_id.lower(): # Canadian Samsung channels
return "🇨🇦 Canada"
return "🇺🇸 United States" # Default Samsung to US
# Plex TV (mostly US unless specifically regional)
if "plex.tv" in all_text or "provider-static.plex.tv" in all_text:
return "🇺🇸 United States"
# PRIORITY 4: Pattern matching by keywords
country_patterns = {
"🇺🇸 United States": ["usa", "america", "united states", "c-span", "newsmax", "newsnation"],
"🇨🇦 Canada": ["canada", "canadian", "ctv", "global", "sportsnet"],
"🇬🇧 United Kingdom": ["uk", "british", "britain", "england"],
"🇵🇭 Philippines": ["philippines", "filipino"],
"🇦🇺 Australia": ["australia", "australian"],
"🇯🇵 Japan": ["japan", "japanese", "nhk"],
"🇲🇾 Malaysia": ["malaysia", "malaysian"],
"🇩🇪 Germany": ["germany", "german", "deutschland"],
"🇫🇷 France": ["france", "french"],
"🇪🇸 Spain": ["spain", "spanish"],
"🇮🇹 Italy": ["italy", "italian"],
"🇧🇷 Brazil": ["brazil", "brazilian"],
"🇲🇽 Mexico": ["mexico", "mexican"],
"🇷🇺 Russia": ["russia", "russian"]
} }
for country, keywords in country_patterns.items(): for domain, country in epg_countries.items():
if domain in epg_id.lower():
return country
# Essential broadcaster patterns (simplified)
broadcasters = {
"🇨🇦 Canada": ["tsn", "cbc", "ctv", "global", "sportsnet", "w network"],
"🇺🇸 United States": ["cbs", "nbc", "abc", "fox", "cnn", "espn", "discovery", "hgtv", "mtv", "c-span"],
"🇬🇧 United Kingdom": ["bbc", "itv", "sky", "channel 4", "channel 5"],
"🇵🇭 Philippines": ["abs-cbn", "gma", "anc", "tv5"],
"🇦🇺 Australia": ["abc australia", "nine network", "seven network"],
"🇯🇵 Japan": ["nhk", "fuji tv", "animax"],
"🇲🇾 Malaysia": ["tv1", "tv2", "astro"],
"🇩🇪 Germany": ["ard", "zdf", "rtl", "sat.1"],
"🇫🇷 France": ["tf1", "france 2", "canal+"],
"🇪🇸 Spain": ["antena 3", "telecinco", "tve"],
"🇮🇹 Italy": ["rai", "mediaset", "canale 5"],
"🇧🇷 Brazil": ["globo", "sbt", "record"],
"🇲🇽 Mexico": ["televisa", "tv azteca"],
"🇷🇺 Russia": ["первый", "россия", "нтв"]
}
for country, keywords in broadcasters.items():
if any(keyword in all_text for keyword in keywords): if any(keyword in all_text for keyword in keywords):
return country return country
return "🌍 International" return "Uncategorized"
def detect_platform(all_text):
"""Detect streaming platform."""
# Platform detection patterns
if "pluto.tv" in all_text or "images.pluto.tv" in all_text or "jmp2.uk/plu-" in all_text:
return "Pluto TV"
elif "plex.tv" in all_text or "provider-static.plex.tv" in all_text:
return "Plex TV"
elif "samsung" in all_text or "sam-" in all_text:
return "Samsung TV+"
elif "tubi" in all_text:
return "Tubi"
elif "xumo" in all_text:
return "Xumo"
elif "crackle" in all_text:
return "Crackle"
elif "roku" in all_text:
return "Roku Channel"
elif "youtube" in all_text:
return "YouTube"
elif "peacock" in all_text:
return "Peacock"
elif "paramount+" in all_text or "paramount plus" in all_text:
return "Paramount+"
return None # No platform detected = traditional broadcaster
def load_channels(): def load_channels():
"""Load existing channels from channels.txt.""" """Load channels from channels.txt."""
channels = []
if not os.path.exists('channels.txt'): if not os.path.exists('channels.txt'):
print("❌ No existing channels.txt found") print("❌ No channels.txt found")
return channels return []
try: try:
with open('channels.txt', 'r', encoding='utf-8') as f: with open('channels.txt', 'r', encoding='utf-8') as f:
content = f.read() content = f.read()
print(f"📄 channels.txt size: {len(content)} characters") channels = []
for block in content.split('\n\n'):
blocks = content.split('\n\n')
for block in blocks:
if not block.strip(): if not block.strip():
continue continue
lines = block.strip().split('\n')
channel_data = {}
for line in lines: channel_data = {}
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() channel_data[key.strip()] = value.strip()
if channel_data and channel_data.get('Stream name'): if channel_data.get('Stream name'):
channels.append(channel_data) channels.append(channel_data)
print(f"✅ Loaded {len(channels)} existing channels") print(f"✅ Loaded {len(channels)} 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 channels
def reorganize_channels_by_country_platform(channels): def reorganize_channels(channels):
"""Reorganize channels by country, then platform within country.""" """Reorganize channels by country (traditional only)."""
print("🌍 Reorganizing channels by country + platform...") print("📺 Organizing traditional broadcasters by country...")
changes = 0 changes = 0
group_stats = {} stats = {}
for channel in channels: for channel in channels:
old_group = channel.get('Group', 'Uncategorized') old_group = channel.get('Group', 'Uncategorized')
@ -222,144 +109,112 @@ def reorganize_channels_by_country_platform(channels):
logo = channel.get('Logo', '') logo = channel.get('Logo', '')
stream_url = channel.get('Stream URL', '') stream_url = channel.get('Stream URL', '')
# Apply country + platform detection new_group = detect_traditional_broadcaster_country(stream_name, epg_id, logo, stream_url)
new_group = detect_country_and_platform(stream_name, epg_id, logo, stream_url)
if old_group != new_group: if old_group != new_group:
print(f"🔄 Reorg: '{stream_name}' {old_group}{new_group}") if new_group == "Uncategorized":
print(f"📱 Platform: '{stream_name}' → Uncategorized")
else:
print(f"📺 Fixed: '{stream_name}' {old_group}{new_group}")
channel['Group'] = new_group channel['Group'] = new_group
changes += 1 changes += 1
# Count groups stats[new_group] = stats.get(new_group, 0) + 1
group_stats[new_group] = group_stats.get(new_group, 0) + 1
print(f"✅ Reorganized {changes} channel classifications") print(f"\n✅ Changes made: {changes}")
print(f"📺 Traditional broadcasters: {sum(v for k, v in stats.items() if k != 'Uncategorized')}")
# Show organization results print(f"📱 Streaming/Unknown: {stats.get('Uncategorized', 0)}")
print(f"\n🗂️ NEW ORGANIZATION:")
sorted_groups = sorted(group_stats.items(), key=lambda x: (x[0].split(' - ')[0], x[0]))
for group, count in sorted_groups:
print(f" {group}: {count} channels")
return channels return channels
def save_channels(channels): def save_channels(channels):
"""Save channels to channels.txt.""" """Save channels to file."""
# Backup
if os.path.exists('channels.txt'): if os.path.exists('channels.txt'):
backup_name = f"channels_backup_{datetime.now().strftime('%Y%m%d_%H%M%S')}.txt" backup = f"channels_backup_{datetime.now().strftime('%Y%m%d_%H%M%S')}.txt"
shutil.copy2('channels.txt', backup_name) shutil.copy2('channels.txt', backup)
print(f"📋 Created backup: {backup_name}") print(f"📋 Backup: {backup}")
try: try:
with open('channels.txt', 'w', encoding='utf-8') as f: with open('channels.txt', 'w', encoding='utf-8') as f:
for i, channel in enumerate(channels): for i, channel in enumerate(channels):
if i > 0: if i > 0:
f.write("\n\n") f.write("\n\n")
f.write(f"Group = {channel.get('Group', 'Uncategorized')}\n") f.write(f"Group = {channel.get('Group', 'Uncategorized')}\n")
f.write(f"Stream name = {channel.get('Stream name', 'Unknown')}\n") f.write(f"Stream name = {channel.get('Stream name', 'Unknown')}\n")
f.write(f"Logo = {channel.get('Logo', '')}\n") f.write(f"Logo = {channel.get('Logo', '')}\n")
f.write(f"EPG id = {channel.get('EPG id', '')}\n") f.write(f"EPG id = {channel.get('EPG id', '')}\n")
f.write(f"Stream URL = {channel.get('Stream URL', '')}\n") f.write(f"Stream URL = {channel.get('Stream URL', '')}\n")
print(f"✅ Saved {len(channels)} channels to channels.txt") print(f"✅ Saved {len(channels)} channels")
return True return True
except Exception as e: except Exception as e:
print(f"Error saving channels: {e}") print(f"Save error: {e}")
return False return False
def generate_m3u(channels): def generate_m3u(channels):
"""Generate M3U playlist with country + platform organization.""" """Generate M3U playlist."""
try: try:
with open('playlist.m3u', 'w', encoding='utf-8') as f: with open('playlist.m3u', 'w', encoding='utf-8') as f:
f.write('#EXTM3U\n') f.write('#EXTM3U\n')
valid_channels = 0
group_stats = {}
for channel in channels: for channel in channels:
stream_name = channel.get('Stream name', '') name = channel.get('Stream name', '')
group = channel.get('Group', 'Uncategorized') group = channel.get('Group', 'Uncategorized')
logo = channel.get('Logo', '') logo = channel.get('Logo', '')
epg_id = channel.get('EPG id', '') epg_id = channel.get('EPG id', '')
url = channel.get('Stream URL', '') url = channel.get('Stream URL', '')
if stream_name and url: if name and url:
f.write(f'#EXTINF:-1 group-title="{group}"') f.write(f'#EXTINF:-1 group-title="{group}"')
if logo: if logo:
f.write(f' tvg-logo="{logo}"') f.write(f' tvg-logo="{logo}"')
if epg_id: if epg_id:
f.write(f' tvg-id="{epg_id}"') f.write(f' tvg-id="{epg_id}"')
f.write(f',{stream_name}\n') f.write(f',{name}\n{url}\n')
f.write(f'{url}\n')
valid_channels += 1
group_stats[group] = group_stats.get(group, 0) + 1
print(f"📺 Generated playlist.m3u with {valid_channels} channels")
# Show organized groups
print("🌍 Organized Groups:")
sorted_groups = sorted(group_stats.items(), key=lambda x: (x[0].split(' - ')[0], x[0]))
for group, count in sorted_groups[:15]: # Show top 15
print(f" {group}: {count} channels")
print("✅ Generated playlist.m3u")
return True return True
except Exception as e: except Exception as e:
print(f"Error generating playlist: {e}") print(f"❌ M3U error: {e}")
return False return False
def main(): def main():
"""Main execution function.""" """Main function."""
print("🌍 IPTV Country + Platform Organizer") print("📺 Traditional Broadcasters Only - Country Organization")
print("=" * 60)
print("Organizing channels: Country first, then platform within country")
print("Example: 🇺🇸 USA, 🇺🇸 USA - Plex, 🇨🇦 Canada, 🇨🇦 Canada - Plex")
print("=" * 60) print("=" * 60)
# Load existing channels
channels = load_channels() channels = load_channels()
if not channels: if not channels:
print("❌ No channels found to process")
return False return False
# Reorganize by country + platform # Reorganize
reorganized_channels = reorganize_channels_by_country_platform(channels) channels = reorganize_channels(channels)
# Sort channels: Country first, then platform within country, then channel name # Sort: Countries first, then Uncategorized last
print("📝 Sorting channels by country + platform...") channels.sort(key=lambda x: (
reorganized_channels.sort(key=lambda x: ( "zzz" if x.get('Group') == "Uncategorized" else x.get('Group', ''),
x.get('Group', '').split(' - ')[0], # Country first x.get('Stream name', '')
x.get('Group', ''), # Then platform within country
x.get('Stream name', '') # Then channel name
)) ))
# Save reorganized channels # Save and generate
if not save_channels(reorganized_channels): if not save_channels(channels):
return False return False
# Generate playlist if not generate_m3u(channels):
if not generate_m3u(reorganized_channels):
return False return False
# Clear import file # Clear import
try: try:
with open('bulk_import.m3u', 'w', encoding='utf-8') as f: with open('bulk_import.m3u', 'w', encoding='utf-8') as f:
f.write('#EXTM3U\n# Import processed\n') f.write('#EXTM3U\n')
print("🧹 Cleared import file") print("🧹 Cleared import file")
except: except:
pass pass
print("\n🎉 COUNTRY + PLATFORM ORGANIZATION COMPLETED!") print("\n🎉 COMPLETED!")
print("✅ Channels organized by country first, then platform") print("✅ Traditional broadcasters organized by country")
print("✅ TSN channels → Canada") print("✅ All streaming platforms → Uncategorized")
print("✅ CBC News → Canada")
print("✅ TV Land → USA")
print("✅ Plex/Pluto channels organized within their countries")
print("✅ Clean country-based organization achieved!")
return True return True