From 71e2eb1d39d6f2cb2dacfd73b959c28858ccb5a4 Mon Sep 17 00:00:00 2001 From: stoney420 Date: Sat, 28 Jun 2025 07:19:42 +0200 Subject: [PATCH] Update scripts/playlist_builder.py --- scripts/playlist_builder.py | 157 +++++++++++++++++++----------------- 1 file changed, 81 insertions(+), 76 deletions(-) diff --git a/scripts/playlist_builder.py b/scripts/playlist_builder.py index 84f54b7..438d8a9 100644 --- a/scripts/playlist_builder.py +++ b/scripts/playlist_builder.py @@ -1,83 +1,88 @@ -#!/usr/bin/env python3 """ -IPTV Repository Monthly Maintenance -Run automatically by workflow +Playlist Builder - Generates the final M3U playlist """ -import os -import shutil -import gzip -from pathlib import Path -from datetime import datetime, timedelta +import logging +from typing import Dict, List, Tuple -def monthly_maintenance(): - """Run monthly maintenance tasks.""" - print("šŸ§¹ IPTV Repository Monthly Maintenance") - print("=" * 40) +class PlaylistBuilder: + """Generate M3U playlist files.""" - root_path = Path.cwd() - actions = [] + def __init__(self, config): + self.config = config + self.logger = logging.getLogger(__name__) - # 1. Compress old backups - print("1. Compressing old backups...") - backups_dir = root_path / 'backups' - cutoff_date = datetime.now() - timedelta(days=7) + def generate_m3u(self, channels: List[Dict]) -> Tuple[int, Dict]: + """Generate the M3U playlist file and return stats.""" + m3u_lines = ["#EXTM3U"] + valid_channels = 0 + country_stats = {} + + for channel in channels: + stream_name = channel.get('Stream name', '') + group_name = channel.get('Group', 'Uncategorized') + logo_url = channel.get('Logo', '') + epg_id = channel.get('EPG id', '') + stream_url = channel.get('Stream URL', '') + + if not stream_name or not stream_url: + continue + + # Build EXTINF line with all attributes + extinf_attrs = [ + f'tvg-id="{epg_id}"', + f'tvg-logo="{logo_url}"', + f'group-title="{group_name}"', + f'tvg-name="{stream_name}"' + ] + + extinf_line = f"#EXTINF:-1 {' '.join(extinf_attrs)},{stream_name}" + m3u_lines.append(extinf_line) + m3u_lines.append(stream_url) + valid_channels += 1 + + # Update country statistics + country_stats[group_name] = country_stats.get(group_name, 0) + 1 + + try: + with open(self.config.playlist_file, 'w', encoding='utf-8') as f: + for line in m3u_lines: + f.write(line + '\n') + + self.logger.info(f"Generated {self.config.playlist_file} with {valid_channels} channels") + + # Log top countries + sorted_stats = dict(sorted(country_stats.items(), key=lambda x: x[1], reverse=True)) + top_countries = dict(list(sorted_stats.items())[:5]) + self.logger.info(f"Top countries: {top_countries}") + + return valid_channels, country_stats + + except Exception as e: + self.logger.error(f"Error writing playlist: {e}") + return 0, {} - if backups_dir.exists(): - for backup_file in backups_dir.glob('*.txt'): - try: - file_date = datetime.fromtimestamp(backup_file.stat().st_mtime) - if file_date < cutoff_date: - compressed_path = backup_file.with_suffix('.txt.gz') - if not compressed_path.exists(): - with open(backup_file, 'rb') as f_in: - with gzip.open(compressed_path, 'wb') as f_out: - shutil.copyfileobj(f_in, f_out) - backup_file.unlink() - actions.append(f"Compressed: {backup_file.name}") - except Exception as e: - print(f" Warning: {e}") - - # 2. Archive old reports - print("2. Archiving old reports...") - reports_dir = root_path / 'reports' / 'daily' - archive_dir = root_path / 'reports' / 'archive' - - cutoff_date = datetime.now() - timedelta(days=30) - - if reports_dir.exists(): - for report_file in reports_dir.glob('*.md'): - try: - file_date = datetime.fromtimestamp(report_file.stat().st_mtime) - if file_date < cutoff_date: - month_folder = archive_dir / file_date.strftime('%Y-%m') - month_folder.mkdir(parents=True, exist_ok=True) - - new_path = month_folder / report_file.name - shutil.move(str(report_file), str(new_path)) - actions.append(f"Archived: {report_file.name}") - except Exception as e: - print(f" Warning: {e}") - - # 3. Clean temporary files - print("3. Cleaning temporary files...") - patterns = ['*_temp*', '*.tmp', '*~', '*.swp'] - - for pattern in patterns: - for temp_file in root_path.rglob(pattern): - if temp_file.is_file() and '.git' not in str(temp_file): - try: - temp_file.unlink() - actions.append(f"Removed: {temp_file.relative_to(root_path)}") - except Exception as e: - print(f" Warning: {e}") - - print(f"\nāœ… Monthly maintenance complete! {len(actions)} actions taken") - if actions: - for action in actions[:5]: - print(f" āœ… {action}") - if len(actions) > 5: - print(f" ... and {len(actions) - 5} more") - -if __name__ == "__main__": - monthly_maintenance() \ No newline at end of file + def validate_m3u_structure(self) -> bool: + """Validate the generated M3U file structure.""" + try: + with open(self.config.playlist_file, 'r', encoding='utf-8') as f: + content = f.read() + + lines = content.strip().split('\n') + + if not lines or lines[0] != '#EXTM3U': + self.logger.error("M3U file missing #EXTM3U header") + return False + + extinf_count = sum(1 for line in lines if line.startswith('#EXTINF:')) + url_count = sum(1 for line in lines if line.startswith(('http://', 'https://', 'rtmp://'))) + + if extinf_count != url_count: + self.logger.warning(f"M3U structure mismatch: {extinf_count} EXTINF lines vs {url_count} URLs") + + self.logger.info(f"M3U validation complete: {extinf_count} channels validated") + return True + + except Exception as e: + self.logger.error(f"Error validating M3U: {e}") + return False \ No newline at end of file