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

This commit is contained in:
stoney420 2025-06-29 02:35:11 +02:00
parent a867930605
commit 30f1e9f11b

View file

@ -1,7 +1,8 @@
#!/usr/bin/env python3
"""
IPTV Playlist Generator - Enhanced Country Detection
FIXED: Properly handles working directory for Forgejo
IPTV Country + Platform Organizer
Groups channels by country first, then platform within country
Example: 🇺🇸 USA, 🇺🇸 USA - Plex, 🇺🇸 USA - Pluto, 🇨🇦 Canada, 🇨🇦 Canada - Plex
"""
import os
@ -13,19 +14,28 @@ from pathlib import Path
# FIXED: Ensure we're in the right directory
script_dir = Path(__file__).parent
root_dir = script_dir.parent
# Change to root directory where channels.txt should be
os.chdir(root_dir)
def setup_directories():
"""Create required directories."""
os.makedirs('reports/daily', exist_ok=True)
os.makedirs('backups', exist_ok=True)
os.makedirs('logs', exist_ok=True)
def detect_country_and_platform(channel_name, epg_id="", logo_url="", stream_url=""):
"""Enhanced country + platform detection."""
all_text = f"{channel_name.lower().strip()} {epg_id.lower().strip()} {logo_url.lower().strip()} {stream_url.lower().strip()}"
channel_lower = channel_name.lower()
def detect_country_enhanced(channel_name, epg_id="", logo_url=""):
"""Enhanced country detection with all the fixes."""
all_text = f"{channel_name.lower().strip()} {epg_id.lower().strip()} {logo_url.lower().strip()}"
# 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=""):
"""Detect the base country of the channel."""
all_text = f"{channel_name.lower().strip()} {epg_id.lower().strip()} {logo_url.lower().strip()} {stream_url.lower().strip()}"
channel_lower = channel_name.lower()
# PRIORITY 1: EPG ID suffix detection (most reliable)
@ -41,99 +51,129 @@ def detect_country_enhanced(channel_name, epg_id="", logo_url=""):
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
# Canadian sports channels (TSN series)
# Canadian channels
if any(x in channel_lower for x in ["tsn 1", "tsn 2", "tsn 3", "tsn 4", "tsn 5", "tsn1", "tsn2", "tsn3", "tsn4", "tsn5"]):
return "🇨🇦 Canada"
# CBC News Toronto (Canadian)
if "cbc news toronto" in channel_lower:
if any(x in channel_lower for x in ["cbc news", "cbc news toronto", "cbc news british columbia"]):
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 shows/channels
# 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 news channels
# Philippines channels
if any(x in channel_lower for x in ["anc global", "anc ph"]):
return "🇵🇭 Philippines"
# Japan anime channels
# 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: Platform-based detection
# Pluto TV special handling
# 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_overrides = {
pluto_countries = {
"cbc news toronto": "🇨🇦 Canada",
"cbc news british columbia": "🇨🇦 Canada",
"come dine with me": "🇬🇧 United Kingdom"
}
for channel_pattern, country in pluto_overrides.items():
for channel_pattern, country in pluto_countries.items():
if channel_pattern in channel_lower:
return country
return "🇺🇸 United States" # Default Pluto TV to US
# Default Pluto TV to US
return "🇺🇸 United States"
# Plex TV handling (mostly US)
# 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
patterns = {
"🇺🇸 United States": ["usa", "us ", "america", "cbs", "nbc", "abc", "fox", "espn", "cnn", "amc", "mtv", "comedy central", "nickelodeon", "disney", "hgtv", "syfy", "bravo", "tlc", "lifetime", "paramount", "weather channel", "tmz", "wgn"],
"🇨🇦 Canada": ["canada", "canadian", "cbc", "ctv", "global", "tsn", "sportsnet", "w network", "much", "teletoon"],
"🇬🇧 United Kingdom": ["uk", "british", "bbc", "itv", "sky", "channel 4", "channel 5", "dave", "quest", "bt sport", "premier league"],
"🇵🇭 Philippines": ["philippines", "filipino", "abs-cbn", "gma", "anc", "cnn philippines"],
"🇦🇺 Australia": ["australia", "australian", "abc australia", "nine network", "seven network", "ten network"],
"🇯🇵 Japan": ["japan", "japanese", "nhk", "fuji tv", "animax"],
"🇮🇳 India": ["india", "indian", "hindi", "zee", "star", "sony", "colors"],
"🇩🇪 Germany": ["germany", "german", "ard", "zdf", "rtl", "sat.1", "pro7"],
"🇫🇷 France": ["france", "french", "tf1", "france 2", "m6", "canal+"],
"🇪🇸 Spain": ["spain", "spanish", "antena 3", "telecinco", "tve"],
"🇮🇹 Italy": ["italy", "italian", "rai", "mediaset", "canale 5"],
"🇳🇱 Netherlands": ["netherlands", "dutch", "npo", "rtl 4"],
"🇧🇷 Brazil": ["brazil", "brazilian", "globo", "sbt", "record"],
"🇲🇽 Mexico": ["mexico", "mexican", "televisa", "tv azteca"],
"🇷🇺 Russia": ["russia", "russian", "первый", "россия", "нтв"]
# 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 patterns.items():
for country, keywords in country_patterns.items():
if any(keyword in all_text for keyword in keywords):
return country
return "🌍 International"
def debug_current_directory():
"""Debug what files are available in current directory."""
current_dir = os.getcwd()
print(f"🗂️ Current working directory: {current_dir}")
def detect_platform(all_text):
"""Detect streaming platform."""
files = os.listdir('.')
print(f"📁 Files in directory: {len(files)} items")
# 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+"
# Check for our key files
key_files = ['channels.txt', 'playlist.m3u', 'bulk_import.m3u']
for file in key_files:
if os.path.exists(file):
size = os.path.getsize(file)
print(f"✅ Found {file} ({size} bytes)")
else:
print(f"❌ Missing {file}")
return None # No platform detected = traditional broadcaster
def load_channels():
"""Load existing channels from channels.txt."""
channels = []
# Debug first
debug_current_directory()
if not os.path.exists('channels.txt'):
print("❌ No existing channels.txt found")
return channels
@ -168,26 +208,39 @@ def load_channels():
return channels
def update_channel_countries(channels):
"""Update all channels with enhanced country detection."""
print("🌍 Updating channel countries with enhanced detection...")
def reorganize_channels_by_country_platform(channels):
"""Reorganize channels by country, then platform within country."""
print("🌍 Reorganizing channels by country + platform...")
changes = 0
group_stats = {}
for channel in channels:
old_group = channel.get('Group', 'Uncategorized')
stream_name = channel.get('Stream name', '')
epg_id = channel.get('EPG id', '')
logo = channel.get('Logo', '')
stream_url = channel.get('Stream URL', '')
new_group = detect_country_enhanced(stream_name, epg_id, logo)
# Apply country + platform detection
new_group = detect_country_and_platform(stream_name, epg_id, logo, stream_url)
if old_group != new_group:
print(f"🔄 Fix: '{stream_name}' {old_group}{new_group}")
print(f"🔄 Reorg: '{stream_name}' {old_group}{new_group}")
channel['Group'] = new_group
changes += 1
print(f"✅ Updated {changes} channel classifications")
# Count groups
group_stats[new_group] = group_stats.get(new_group, 0) + 1
print(f"✅ Reorganized {changes} channel classifications")
# Show organization results
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
def save_channels(channels):
@ -217,13 +270,13 @@ def save_channels(channels):
return False
def generate_m3u(channels):
"""Generate M3U playlist."""
"""Generate M3U playlist with country + platform organization."""
try:
with open('playlist.m3u', 'w', encoding='utf-8') as f:
f.write('#EXTM3U\n')
valid_channels = 0
country_stats = {}
group_stats = {}
for channel in channels:
stream_name = channel.get('Stream name', '')
@ -242,16 +295,15 @@ def generate_m3u(channels):
f.write(f'{url}\n')
valid_channels += 1
country_stats[group] = country_stats.get(group, 0) + 1
group_stats[group] = group_stats.get(group, 0) + 1
print(f"📺 Generated playlist.m3u with {valid_channels} channels")
# Show top countries
sorted_countries = sorted(country_stats.items(), key=lambda x: x[1], reverse=True)
print("🌍 Top Countries:")
for country, count in sorted_countries[:10]:
percentage = (count / valid_channels * 100) if valid_channels > 0 else 0
print(f" {country}: {count} ({percentage:.1f}%)")
# 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")
return True
@ -259,45 +311,13 @@ def generate_m3u(channels):
print(f"❌ Error generating playlist: {e}")
return False
def create_report(channels):
"""Create a simple report."""
try:
timestamp = datetime.now().strftime('%Y%m%d_%H%M%S')
report_file = f"reports/daily/report_{timestamp}.md"
with open(report_file, 'w', encoding='utf-8') as f:
f.write("# 🌍 Enhanced Country Detection Report\n")
f.write(f"**Generated:** {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}\n\n")
f.write(f"## 📊 Statistics\n")
f.write(f"- **Total Channels:** {len(channels)}\n\n")
# Count by country
country_stats = {}
for channel in channels:
group = channel.get('Group', 'Uncategorized')
country_stats[group] = country_stats.get(group, 0) + 1
f.write("## 🌍 Country Distribution\n")
sorted_countries = sorted(country_stats.items(), key=lambda x: x[1], reverse=True)
for country, count in sorted_countries:
percentage = (count / len(channels) * 100) if len(channels) > 0 else 0
f.write(f"- **{country}:** {count} channels ({percentage:.1f}%)\n")
f.write("\n---\n")
f.write("*Enhanced country detection with 99%+ accuracy*\n")
print(f"📊 Report created: {report_file}")
except Exception as e:
print(f"⚠️ Could not create report: {e}")
def main():
"""Main execution function."""
print("🚀 IPTV Playlist Generator - Enhanced Country Detection")
print("🌍 IPTV Country + Platform Organizer")
print("=" * 60)
print("Organizing channels: Country first, then platform within country")
print("Example: 🇺🇸 USA, 🇺🇸 USA - Plex, 🇨🇦 Canada, 🇨🇦 Canada - Plex")
print("=" * 60)
# Setup
setup_directories()
# Load existing channels
channels = load_channels()
@ -306,23 +326,25 @@ def main():
print("❌ No channels found to process")
return False
# Update countries with enhanced detection
updated_channels = update_channel_countries(channels)
# Reorganize by country + platform
reorganized_channels = reorganize_channels_by_country_platform(channels)
# Sort channels
updated_channels.sort(key=lambda x: (x.get('Group', ''), x.get('Stream name', '')))
# Sort channels: Country first, then platform within country, then channel name
print("📝 Sorting channels by country + platform...")
reorganized_channels.sort(key=lambda x: (
x.get('Group', '').split(' - ')[0], # Country first
x.get('Group', ''), # Then platform within country
x.get('Stream name', '') # Then channel name
))
# Save updated channels
if not save_channels(updated_channels):
# Save reorganized channels
if not save_channels(reorganized_channels):
return False
# Generate playlist
if not generate_m3u(updated_channels):
if not generate_m3u(reorganized_channels):
return False
# Create report
create_report(updated_channels)
# Clear import file
try:
with open('bulk_import.m3u', 'w', encoding='utf-8') as f:
@ -331,12 +353,13 @@ def main():
except:
pass
print("\n🎉 ENHANCED COUNTRY DETECTION COMPLETED!")
print("✅ All TSN channels should now be in Canada")
print("✅ TV Land, We TV should now be in USA")
print("✅ ANC channels should now be in Philippines")
print("✅ Come Dine with Me should now be in UK")
print("✅ Animax should now be in Japan")
print("\n🎉 COUNTRY + PLATFORM ORGANIZATION COMPLETED!")
print("✅ Channels organized by country first, then platform")
print("✅ TSN channels → Canada")
print("✅ CBC News → Canada")
print("✅ TV Land → USA")
print("✅ Plex/Pluto channels organized within their countries")
print("✅ Clean country-based organization achieved!")
return True