diff --git a/scripts/generate_playlist.py b/scripts/generate_playlist.py index a1cecb5..7f4bdff 100644 --- a/scripts/generate_playlist.py +++ b/scripts/generate_playlist.py @@ -1,219 +1,106 @@ #!/usr/bin/env python3 """ -IPTV Country + Platform Organizer -Groups channels by country first, then platform within country -Example: ๐Ÿ‡บ๐Ÿ‡ธ USA, ๐Ÿ‡บ๐Ÿ‡ธ USA - Plex, ๐Ÿ‡บ๐Ÿ‡ธ USA - Pluto, ๐Ÿ‡จ๐Ÿ‡ฆ Canada, ๐Ÿ‡จ๐Ÿ‡ฆ Canada - Plex +IPTV Traditional Broadcasters Only - Simplified Version +Clean, minimal code with no redundancy """ import os -import sys import shutil from datetime import datetime from pathlib import Path -# FIXED: Ensure we're in the right directory +# Ensure correct directory script_dir = Path(__file__).parent root_dir = script_dir.parent os.chdir(root_dir) -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() - - # 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 is_streaming_platform(all_text): + """Check if channel is from streaming platform (simplified).""" + platforms = ["plex.tv", "pluto.tv", "jmp2.uk/plu-", "sam-", "samsung", "tubi", "xumo", "roku", "youtube"] + return any(platform in all_text for platform in platforms) -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()}" +def detect_traditional_broadcaster_country(channel_name, epg_id="", logo_url="", stream_url=""): + """Detect country for traditional broadcasters only.""" + all_text = f"{channel_name.lower()} {epg_id.lower()} {logo_url.lower()} {stream_url.lower()}" channel_lower = channel_name.lower() - # PRIORITY 1: EPG ID suffix detection (most reliable) - if ".ca" in epg_id.lower(): - return "๐Ÿ‡จ๐Ÿ‡ฆ Canada" - 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" + # Skip streaming platforms + if is_streaming_platform(all_text): + return "Uncategorized" - # PRIORITY 2: Specific channel fixes for misclassified channels - - # 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" - 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 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"] + # 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" } - 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): return country - return "๐ŸŒ International" - -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 + return "Uncategorized" def load_channels(): - """Load existing channels from channels.txt.""" - channels = [] - + """Load channels from channels.txt.""" if not os.path.exists('channels.txt'): - print("โŒ No existing channels.txt found") - return channels + print("โŒ No channels.txt found") + return [] try: with open('channels.txt', 'r', encoding='utf-8') as f: content = f.read() - print(f"๐Ÿ“„ channels.txt size: {len(content)} characters") - - blocks = content.split('\n\n') - - for block in blocks: + channels = [] + for block in content.split('\n\n'): if not block.strip(): continue - - lines = block.strip().split('\n') - channel_data = {} - for line in lines: + 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 and channel_data.get('Stream name'): + if channel_data.get('Stream name'): channels.append(channel_data) - print(f"โœ… Loaded {len(channels)} existing channels") + print(f"โœ… Loaded {len(channels)} channels") + return channels except Exception as e: print(f"โŒ Error loading channels: {e}") - - return channels + return [] -def reorganize_channels_by_country_platform(channels): - """Reorganize channels by country, then platform within country.""" - print("๐ŸŒ Reorganizing channels by country + platform...") +def reorganize_channels(channels): + """Reorganize channels by country (traditional only).""" + print("๐Ÿ“บ Organizing traditional broadcasters by country...") changes = 0 - group_stats = {} + stats = {} for channel in channels: old_group = channel.get('Group', 'Uncategorized') @@ -222,144 +109,112 @@ def reorganize_channels_by_country_platform(channels): logo = channel.get('Logo', '') stream_url = channel.get('Stream URL', '') - # Apply country + platform detection - new_group = detect_country_and_platform(stream_name, epg_id, logo, stream_url) + new_group = detect_traditional_broadcaster_country(stream_name, epg_id, logo, stream_url) 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 changes += 1 - # Count groups - group_stats[new_group] = group_stats.get(new_group, 0) + 1 + stats[new_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") + 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 channels.txt.""" + """Save channels to file.""" + # Backup if os.path.exists('channels.txt'): - backup_name = f"channels_backup_{datetime.now().strftime('%Y%m%d_%H%M%S')}.txt" - shutil.copy2('channels.txt', backup_name) - print(f"๐Ÿ“‹ Created backup: {backup_name}") + 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 to channels.txt") + print(f"โœ… Saved {len(channels)} channels") return True - except Exception as e: - print(f"โŒ Error saving channels: {e}") + print(f"โŒ Save error: {e}") return False def generate_m3u(channels): - """Generate M3U playlist with country + platform organization.""" + """Generate M3U playlist.""" try: with open('playlist.m3u', 'w', encoding='utf-8') as f: f.write('#EXTM3U\n') - valid_channels = 0 - group_stats = {} - for channel in channels: - stream_name = channel.get('Stream name', '') + 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 stream_name and 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',{stream_name}\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") + f.write(f',{name}\n{url}\n') + print("โœ… Generated playlist.m3u") return True - except Exception as e: - print(f"โŒ Error generating playlist: {e}") + print(f"โŒ M3U error: {e}") return False def main(): - """Main execution function.""" - print("๐ŸŒ IPTV Country + Platform Organizer") - print("=" * 60) - print("Organizing channels: Country first, then platform within country") - print("Example: ๐Ÿ‡บ๐Ÿ‡ธ USA, ๐Ÿ‡บ๐Ÿ‡ธ USA - Plex, ๐Ÿ‡จ๐Ÿ‡ฆ Canada, ๐Ÿ‡จ๐Ÿ‡ฆ Canada - Plex") + """Main function.""" + print("๐Ÿ“บ Traditional Broadcasters Only - Country Organization") print("=" * 60) - # Load existing channels channels = load_channels() - if not channels: - print("โŒ No channels found to process") return False - # Reorganize by country + platform - reorganized_channels = reorganize_channels_by_country_platform(channels) + # Reorganize + channels = reorganize_channels(channels) - # 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 + # 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 reorganized channels - if not save_channels(reorganized_channels): + # Save and generate + if not save_channels(channels): return False - # Generate playlist - if not generate_m3u(reorganized_channels): + if not generate_m3u(channels): return False - # Clear import file + # Clear import try: 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") except: pass - 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!") + print("\n๐ŸŽ‰ COMPLETED!") + print("โœ… Traditional broadcasters organized by country") + print("โœ… All streaming platforms โ†’ Uncategorized") return True