my-private-iptv-m3u/scripts/generate_playlist.py
stoney420 202773f4e5
All checks were successful
Generate M3U Playlist with Auto-Organization / build-and-organize (push) Successful in 22s
Update scripts/generate_playlist.py
2025-06-29 03:33:41 +02:00

290 lines
No EOL
11 KiB
Python
Raw Blame History

This file contains invisible Unicode characters

This file contains invisible Unicode characters that are indistinguishable to humans but may be processed differently by a computer. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.

This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.

#!/usr/bin/env python3
"""
IPTV Enhanced Country Detection - Simplified No-Hang Version
Uses your existing 3-point analysis but removes network calls that cause hanging
"""
import os
import shutil
import re
from datetime import datetime
from pathlib import Path
# Ensure correct directory
script_dir = Path(__file__).parent
root_dir = script_dir.parent
os.chdir(root_dir)
def is_streaming_platform(all_text):
"""Check if channel is from streaming platform."""
platforms = [
"plex.tv", "pluto.tv", "jmp2.uk/plu-", "sam-", "samsung",
"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=""):
"""Enhanced 3-point country detection for traditional broadcasters."""
all_text = f"{channel_name.lower()} {epg_id.lower()} {logo_url.lower()} {stream_url.lower()}"
channel_lower = channel_name.lower()
# Skip streaming platforms entirely
if is_streaming_platform(all_text):
return "Uncategorized"
# PRIORITY 1: EPG ID detection (most reliable)
epg_countries = {
".ca": "🇨🇦 Canada", ".us": "🇺🇸 United States", ".uk": "🇬🇧 United Kingdom",
".ph": "🇵🇭 Philippines", ".au": "🇦🇺 Australia", ".jp": "🇯🇵 Japan",
".my": "🇲🇾 Malaysia", ".de": "🇩🇪 Germany", ".fr": "🇫🇷 France",
".es": "🇪🇸 Spain", ".it": "🇮🇹 Italy", ".br": "🇧🇷 Brazil",
".mx": "🇲🇽 Mexico", ".ar": "🇦🇷 Argentina"
}
for domain, country in epg_countries.items():
if domain in epg_id.lower():
return country
# PRIORITY 2: Specific channel fixes from your original code
# Canadian sports channels (TSN series)
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:
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"
# UK shows/channels
if "come dine with me" in channel_lower:
return "🇬🇧 United Kingdom"
# Philippines news channels
if any(x in channel_lower for x in ["anc global", "anc ph"]):
return "🇵🇭 Philippines"
# Japan anime channels
if "animax" in channel_lower:
return "🇯🇵 Japan"
# PRIORITY 3: Logo URL detection (your 3-point analysis)
logo_patterns = {
"/canada/": "🇨🇦 Canada", "/ca/": "🇨🇦 Canada", "canada": "🇨🇦 Canada",
"/usa/": "🇺🇸 United States", "/us/": "🇺🇸 United States", "america": "🇺🇸 United States",
"/uk/": "🇬🇧 United Kingdom", "/britain/": "🇬🇧 United Kingdom", "british": "🇬🇧 United Kingdom",
"/germany/": "🇩🇪 Germany", "/de/": "🇩🇪 Germany",
"/france/": "🇫🇷 France", "/fr/": "🇫🇷 France",
"/spain/": "🇪🇸 Spain", "/es/": "🇪🇸 Spain",
"/italy/": "🇮🇹 Italy", "/it/": "🇮🇹 Italy"
}
for pattern, country in logo_patterns.items():
if pattern in logo_url.lower():
return country
# PRIORITY 4: Essential broadcaster patterns
broadcasters = {
"🇨🇦 Canada": ["tsn", "cbc", "ctv", "global", "sportsnet", "w network", "city tv", "aptn"],
"🇺🇸 United States": ["cbs", "nbc", "abc", "fox", "cnn", "espn", "discovery", "hgtv", "mtv", "c-span", "msnbc", "comedy central"],
"🇬🇧 United Kingdom": ["bbc", "itv", "sky", "channel 4", "channel 5", "dave", "e4", "film4"],
"🇵🇭 Philippines": ["abs-cbn", "gma", "anc", "tv5", "pbo", "net 25"],
"🇦🇺 Australia": ["abc australia", "nine network", "seven network", "ten network", "sbs"],
"🇯🇵 Japan": ["nhk", "fuji tv", "animax", "tbs", "tv asahi"],
"🇲🇾 Malaysia": ["tv1", "tv2", "astro", "rtm", "8tv"],
"🇩🇪 Germany": ["ard", "zdf", "rtl", "sat.1", "pro7", "vox"],
"🇫🇷 France": ["tf1", "france 2", "canal+", "m6", "arte"],
"🇪🇸 Spain": ["antena 3", "telecinco", "tve", "la sexta"],
"🇮🇹 Italy": ["rai", "mediaset", "canale 5", "la7"],
"🇧🇷 Brazil": ["globo", "sbt", "record", "band"],
"🇲🇽 Mexico": ["televisa", "tv azteca", "canal 5"],
"🇷🇺 Russia": ["первый", "россия", "нтв", "тнт"]
}
for country, keywords in broadcasters.items():
if any(keyword in all_text for keyword in keywords):
return country
return "Uncategorized"
def load_channels():
"""Load channels from channels.txt."""
if not os.path.exists('channels.txt'):
print("❌ No channels.txt found")
return []
try:
with open('channels.txt', 'r', encoding='utf-8') as f:
content = f.read()
channels = []
for block in content.split('\n\n'):
if not block.strip():
continue
channel_data = {}
for line in block.strip().split('\n'):
if '=' in line:
key, value = line.split('=', 1)
channel_data[key.strip()] = value.strip()
if channel_data.get('Stream name'):
channels.append(channel_data)
print(f"✅ Loaded {len(channels)} channels")
return channels
except Exception as e:
print(f"❌ Error loading channels: {e}")
return []
def reorganize_channels(channels):
"""Reorganize channels by country (traditional only)."""
print("📺 Organizing traditional broadcasters by country...")
changes = 0
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_traditional_broadcaster_country(stream_name, epg_id, logo, stream_url)
if 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
changes += 1
stats[new_group] = stats.get(new_group, 0) + 1
print(f"\n✅ Changes made: {changes}")
print(f"📺 Traditional broadcasters: {sum(v for k, v in stats.items() if k != 'Uncategorized')}")
print(f"📱 Streaming/Unknown: {stats.get('Uncategorized', 0)}")
return channels
def save_channels(channels):
"""Save channels to file."""
# Backup
if os.path.exists('channels.txt'):
backup = f"channels_backup_{datetime.now().strftime('%Y%m%d_%H%M%S')}.txt"
shutil.copy2('channels.txt', backup)
print(f"📋 Backup: {backup}")
try:
with open('channels.txt', 'w', encoding='utf-8') as f:
for i, channel in enumerate(channels):
if i > 0:
f.write("\n\n")
f.write(f"Group = {channel.get('Group', 'Uncategorized')}\n")
f.write(f"Stream name = {channel.get('Stream name', 'Unknown')}\n")
f.write(f"Logo = {channel.get('Logo', '')}\n")
f.write(f"EPG id = {channel.get('EPG id', '')}\n")
f.write(f"Stream URL = {channel.get('Stream URL', '')}\n")
print(f"✅ Saved {len(channels)} channels")
return True
except Exception as e:
print(f"❌ Save error: {e}")
return False
def generate_m3u(channels):
"""Generate M3U playlist."""
try:
with open('playlist.m3u', 'w', encoding='utf-8') as f:
f.write('#EXTM3U\n')
for channel in channels:
name = channel.get('Stream name', '')
group = channel.get('Group', 'Uncategorized')
logo = channel.get('Logo', '')
epg_id = channel.get('EPG id', '')
url = channel.get('Stream URL', '')
if name and url:
f.write(f'#EXTINF:-1 group-title="{group}"')
if logo:
f.write(f' tvg-logo="{logo}"')
if epg_id:
f.write(f' tvg-id="{epg_id}"')
f.write(f',{name}\n{url}\n')
print("✅ Generated playlist.m3u")
return True
except Exception as e:
print(f"❌ M3U error: {e}")
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():
"""Main function."""
print("📺 Enhanced Country Detection - No Network Calls")
print("=" * 60)
# Create logs directory
create_logs()
channels = load_channels()
if not channels:
print(" No channels to process")
return True
# Reorganize
channels = reorganize_channels(channels)
# Sort: Countries first, then Uncategorized last
channels.sort(key=lambda x: (
"zzz" if x.get('Group') == "Uncategorized" else x.get('Group', ''),
x.get('Stream name', '')
))
# Save and generate
if not save_channels(channels):
return False
if not generate_m3u(channels):
return False
# Clear import
try:
with open('bulk_import.m3u', 'w', encoding='utf-8') as f:
f.write('#EXTM3U\n')
print("🧹 Cleared import file")
except:
pass
print("\n🎉 COMPLETED!")
print("✅ Traditional broadcasters organized by country")
print("✅ All streaming platforms → Uncategorized")
print("✅ No network calls - fast execution")
return True
if __name__ == "__main__":
success = main()
exit(0 if success else 1)