diff --git a/scripts/file_manager.py b/scripts/file_manager.py index 09c8fc3..4117d46 100644 --- a/scripts/file_manager.py +++ b/scripts/file_manager.py @@ -1,289 +1,141 @@ -# 📺 IPTV Playlist Manager +""" +File Manager - Handles file operations, backups, and channel loading +""" -[![Build Status](https://img.shields.io/badge/build-passing-brightgreen)](../../actions) -[![Playlist Status](https://img.shields.io/badge/playlist-active-blue)](#) -[![Last Updated](https://img.shields.io/badge/updated-auto-green)](#) -[![Countries](https://img.shields.io/badge/countries-auto--detected-orange)](#) +import os +import re +import shutil +import logging +from datetime import datetime +from pathlib import Path +from typing import Dict, List, Optional -> **Automated IPTV playlist management system** with intelligent channel organization that processes M3U files, manages channel databases, and generates clean playlists via Forgejo Actions. - -## ✨ **What Makes This Special** - -- 🌍 **Smart Country Detection** - Automatically organizes channels by country using advanced pattern matching -- 🎬 **Quality Recognition** - Detects and labels 4K, FHD, HD, and SD streams -- 🔄 **Intelligent Deduplication** - Removes duplicates using signature-based matching -- 📊 **Professional Reporting** - Detailed statistics and processing logs -- 🛡️ **Content Filtering** - Optional adult content filtering -- ⚡ **Lightning Fast** - Processes 1000+ channels in seconds - -## 🚀 Quick Start - -### 📥 **Download Your Playlist** -- **[📺 Download playlist.m3u](./playlist.m3u)** - Ready to use in any IPTV player -- **[📋 View Channel List](./channels.txt)** - See all available channels -- **[📊 Check Reports](./reports/daily/)** - View processing history and statistics - -### ⚡ **Add Channels (3 Easy Ways)** - -#### Method 1: Upload M3U File (Bulk Import) ⭐ **Recommended** -1. Get your M3U file from your IPTV provider -2. Upload it to this repository as `bulk_import.m3u` -3. Commit the file - **automatic processing begins!** -4. Check reports for import results with country detection - -#### Method 2: Edit Channels Manually -1. Click **[channels.txt](./channels.txt)** and edit it -2. Add channels using this format: -``` -Group = Sports -Stream name = ESPN HD -Logo = https://example.com/espn-logo.png -EPG id = espn.us -Stream URL = http://your-stream-url-here - -Group = News -Stream name = CNN International -Logo = https://example.com/cnn-logo.png -EPG id = cnn.us -Stream URL = http://your-stream-url-here -``` -3. Commit your changes - -#### Method 3: Use the Template -1. Check **[templates/channel_template.txt](./templates/channel_template.txt)** for the exact format -2. Copy the template and fill in your channel details -3. Add to channels.txt - -## 📁 Repository Structure - -``` -📦 Your IPTV Repository -├── 📺 playlist.m3u # 🎯 Generated playlist (your main file!) -├── 📝 channels.txt # 🎯 Channel database (edit this!) -├── 📥 bulk_import.m3u # 🎯 Drop M3U files here for import -├── 📖 README.md # This guide -├── 📁 scripts/ # 🧠 Processing engine (Python) -│ ├── generate_playlist.py # Main orchestrator -│ ├── channel_processor.py # Smart processing & detection -│ ├── playlist_builder.py # M3U generation -│ └── report_generator.py # Statistics & reporting -├── 📁 config/ # ⚙️ Configuration files -│ ├── settings.json # System settings -│ ├── patterns.json # Country detection patterns -│ └── group_overrides.json # Manual overrides -├── 📁 reports/ # 📊 Processing reports & statistics -│ ├── daily/ # Latest reports -│ └── logs/ # System logs -├── 📁 templates/ # 📋 Channel entry templates -├── 📁 backups/ # 💾 Automatic backups -└── 📁 .forgejo/workflows/ # 🤖 Automation workflows -``` - -## 🧠 How It Works (The Smart Stuff) - -The system **intelligently** processes your channels through this pipeline: - -``` -📥 Input (M3U/Manual) → 🔍 Parse & Validate → 🌍 Country Detection - ↓ -🎬 Quality Detection → 🔄 Duplicate Removal → 📺 Generate Clean M3U - ↓ -📊 Create Reports → ✅ Done! -``` - -### **Advanced Features:** -1. **🌍 Smart Country Detection** - Uses 500+ patterns to detect countries from channel names -2. **🎬 Quality Recognition** - Automatically detects 4K, FHD, HD, SD quality -3. **🔍 Intelligent Deduplication** - Advanced signature matching prevents duplicates -4. **📊 Data Validation** - Ensures all channels have required information -5. **🏷️ Auto-Organization** - Groups channels by detected country -6. **📝 Comprehensive Logging** - Tracks all changes and imports -7. **✨ Clean Output** - Generates perfectly formatted M3U files -8. **💾 Automatic Backups** - Never lose your channel data - -## 🌍 **Supported Countries & Detection** - -The system automatically detects and organizes channels from: - -| Region | Countries | Examples | -|--------|-----------|----------| -| **🇺🇸 North America** | USA, Canada | ESPN, CNN, CBC, TSN | -| **🇪🇺 Europe** | UK, Germany, France, Spain, Italy, Netherlands | BBC, Sky, ARD, TF1 | -| **🌏 Other** | Australia, Brazil, Arabic | ABC AU, Globo, MBC | - -**Detection Methods:** -- **Prefixes**: `us:`, `uk:`, `[US]`, `(CA)` -- **Channel Names**: ESPN → 🇺🇸, BBC → 🇬🇧, CBC → 🇨🇦 -- **Network Patterns**: Sky Sports → 🇬🇧, Fox Sports → 🇺🇸 -- **Quality Tags**: Automatic 4K/FHD/HD/SD detection - -## 🛠️ Advanced Configuration - -### **Smart Country Detection** (`config/patterns.json`) -The system includes **500+ detection patterns** for accurate categorization: -```json -{ - "country_patterns": { - "🇺🇸 United States": ["espn", "cnn", "fox", "nbc", "cbs"], - "🇬🇧 United Kingdom": ["bbc", "sky", "itv", "channel 4"], - "🇨🇦 Canada": ["cbc", "tsn", "sportsnet", "ctv"] - } -} -``` - -### **Manual Overrides** (`config/group_overrides.json`) -Force specific channels into custom groups: -```json -{ - "ESPN": "🇺🇸 United States", - "Fox Sports": "🇺🇸 United States", - "BBC News": "🇬🇧 United Kingdom" -} -``` - -### **System Settings** (`config/settings.json`) -Customize processing behavior: -```json -{ - "remove_duplicates": true, - "auto_detect_country": true, - "detect_quality": true, - "skip_adult_content": true, - "sort_channels": true, - "backup_before_import": true -} -``` - -## 📊 **Monitoring & Statistics** - -### **📈 Live Reports** -- **[📊 Latest Report](./reports/daily/)** - Processing statistics and country breakdown -- **[📋 Processing Logs](./reports/logs/playlist_update.log)** - Detailed operation logs -- **[📥 Import History](./reports/logs/import_history.log)** - Import tracking -- **[❌ Error Tracking](./reports/logs/error.log)** - Issue monitoring - -### **📈 What You Get** -- **Channel Counts** - Total and per-country statistics -- **Quality Distribution** - 4K/FHD/HD/SD breakdown -- **Processing Stats** - Import success rates, duplicates removed -- **Country Detection** - Accuracy and coverage metrics -- **Performance Data** - Processing times and efficiency - -## 🎮 **Using Your Playlist** - -### **📱 Popular IPTV Players:** -- **VLC Media Player**: File → Open Network Stream → Paste URL -- **Kodi**: Add-ons → PVR → Simple Client → M3U URL -- **Perfect Player**: Settings → Playlist → Add playlist → M3U URL -- **IPTV Smarters**: Add playlist → M3U URL -- **TiviMate**: Add playlist → M3U Playlist → URL - -### **🔗 Your Playlist URL:** -``` -https://forgejo.plainrock127.xyz/[your-username]/[repo-name]/raw/branch/main/playlist.m3u -``` - -## 🚨 **Troubleshooting** - -### **Common Issues:** - -**❓ Import file not processing?** -- ✅ Make sure file is named exactly `bulk_import.m3u` -- ✅ Check file format is valid M3U with `#EXTM3U` header -- ✅ Look at reports for detailed error information - -**❓ Channels missing after import?** -- ✅ Check reports for duplicate removal notices -- ✅ Verify channel format in original M3U -- ✅ Look for validation errors in processing logs - -**❓ Country detection not working?** -- ✅ Check if channel names match patterns in `config/patterns.json` -- ✅ Add custom patterns for your specific channels -- ✅ Use manual overrides in `config/group_overrides.json` - -**❓ Playlist not updating?** -- ✅ Check Actions tab for workflow status -- ✅ Ensure you committed your changes -- ✅ Review error logs for workflow issues - -**❓ Need help?** -- 📊 Check the reports in the `reports/daily/` folder -- 📋 Review logs in `reports/logs/` folder -- 📖 Review templates in `templates/` folder -- 🐛 Create an issue if problems persist - -## 🎯 **Pro Tips** - -1. **📱 Mobile Friendly**: Repository works great on mobile browsers for quick edits -2. **🔄 Auto-Sync**: Playlist updates automatically when you make changes -3. **💾 Never Lose Data**: Your channel data is version controlled with automatic backups -4. **🏷️ Smart Organization**: Let the system detect countries automatically for better organization -5. **📊 Monitor Health**: Check reports regularly to track system performance -6. **🎬 Quality Labels**: System automatically detects and labels stream quality -7. **🔍 Bulk Processing**: Import hundreds of channels at once with intelligent processing - -## 🔧 **For Developers** - -### **🐍 Technology Stack:** -- **Python 3.11+** - Core processing engine -- **Forgejo Actions** - CI/CD automation -- **JSON Configuration** - Flexible, editable settings -- **Markdown Reporting** - Human-readable reports - -### **🏗️ Architecture:** -```bash -scripts/ -├── generate_playlist.py # 🎯 Main orchestrator -├── channel_processor.py # 🧠 Smart processing & country detection -├── playlist_builder.py # 📺 M3U generation & formatting -├── health_checker.py # 🏥 Optional stream validation -├── report_generator.py # 📊 Statistics & reporting -└── config_manager.py # ⚙️ Configuration management -``` - -### **🚀 Running Locally:** -```bash -# Process channels and generate playlist -python3 scripts/generate_playlist.py - -# Check system configuration -python3 -c "from scripts.config_manager import ConfigManager; print(ConfigManager().get_detection_summary())" -``` - -### **🤝 Contributing:** -1. Fork the repository -2. Create feature branch (`git checkout -b feature/amazing-feature`) -3. Test your changes locally -4. Commit changes (`git commit -m 'Add amazing feature'`) -5. Submit pull request - ---- - -## 📈 **Current Stats** -- **Channels**: Auto-counted from playlist generation -- **Countries**: Auto-detected from smart pattern matching -- **Quality Streams**: 4K/FHD/HD automatically identified -- **Processing Speed**: 1000+ channels per second -- **Detection Accuracy**: 95%+ for known patterns -- **Last Updated**: Auto-timestamp from last workflow run -- **Build Status**: Live from Forgejo Actions - ---- - -## 🏆 **Why This System Rocks** - -✅ **Fully Automated** - Set it and forget it -✅ **Intelligent Processing** - Smarter than basic M3U tools -✅ **Professional Quality** - Enterprise-level features -✅ **Easy to Use** - Simple for beginners, powerful for experts -✅ **Highly Configurable** - Customize everything -✅ **Well Documented** - Comprehensive guides and examples -✅ **Version Controlled** - Never lose your work -✅ **Performance Optimized** - Fast and efficient - ---- - -**🎉 Enjoy your professional-grade automated IPTV playlist management system!** - -*Built with ❤️ for quality IPTV management* \ No newline at end of file +class FileManager: + """Manage file operations with backup and rotation.""" + + def __init__(self, config): + self.config = config + self.logger = logging.getLogger(__name__) + self.backup_dir = Path("backups") + self.backup_dir.mkdir(exist_ok=True) + + def create_backup(self, file_path: str) -> Optional[Path]: + """Create timestamped backup with rotation.""" + if not self.config.settings.get('create_backup', True): + return None + + file_path = Path(file_path) + if not file_path.exists(): + return None + + timestamp = datetime.now().strftime('%Y%m%d_%H%M%S') + backup_path = self.backup_dir / f"{file_path.stem}_{timestamp}{file_path.suffix}" + + try: + shutil.copy2(file_path, backup_path) + self.logger.info(f"Created backup: {backup_path}") + self._cleanup_old_backups(file_path.stem) + return backup_path + except Exception as e: + self.logger.error(f"Failed to create backup: {e}") + return None + + def _cleanup_old_backups(self, base_name: str): + """Remove old backups, keeping only the most recent ones.""" + max_backups = self.config.settings.get('max_backups', 5) + + backup_files = sorted( + [f for f in self.backup_dir.glob(f"{base_name}_*") if f.is_file()], + key=lambda x: x.stat().st_mtime, + reverse=True + ) + + for old_backup in backup_files[max_backups:]: + try: + old_backup.unlink() + self.logger.debug(f"Removed old backup: {old_backup}") + except Exception as e: + self.logger.warning(f"Could not remove old backup {old_backup}: {e}") + + def load_all_channels(self) -> List[Dict]: + """Load all channels from the channels file.""" + if not os.path.exists(self.config.channels_file): + self.logger.info("No channels.txt file found") + return [] + + try: + with open(self.config.channels_file, 'r', encoding='utf-8') as f: + content = f.read() + + channel_blocks = re.split(r'\n\s*\n+', content.strip()) + channels = [] + + for block in channel_blocks: + if block.strip(): + channel = self._parse_channel_block(block) + if channel: + channels.append(channel) + + self.logger.info(f"Loaded {len(channels)} channels from file") + return channels + + except Exception as e: + self.logger.error(f"Error loading channels: {e}") + return [] + + def _parse_channel_block(self, block: str) -> Optional[Dict]: + """Parse a channel block from channels.txt.""" + channel_data = {} + lines = block.strip().split('\n') + + for line in lines: + if '=' in line: + key, value = line.split('=', 1) + channel_data[key.strip()] = value.strip() + + return channel_data if channel_data else None + + def save_channels(self, channels: List[Dict]) -> bool: + """Save channels to the channels.txt file.""" + try: + # Create backup first + self.create_backup(self.config.channels_file) + + with open(self.config.channels_file, 'w', encoding='utf-8') as f: + for i, channel in enumerate(channels): + if i > 0: + f.write("\n\n") + f.write(self._convert_to_channels_txt_block(channel)) + + self.logger.info(f"Saved {len(channels)} channels to file") + return True + + except Exception as e: + self.logger.error(f"Error saving channels: {e}") + return False + + def append_channels(self, new_channels: List[Dict]) -> bool: + """Append new channels to existing channels file.""" + try: + # Load existing channels + existing_channels = self.load_all_channels() + + # Combine with new channels + all_channels = existing_channels + new_channels + + # Save combined channels + return self.save_channels(all_channels) + + except Exception as e: + self.logger.error(f"Error appending channels: {e}") + return False + + def _convert_to_channels_txt_block(self, channel_data: Dict) -> str: + """Convert to channels.txt format.""" + block = [] + block.append(f"Group = {channel_data.get('Group', 'Uncategorized')}") + block.append(f"Stream name = {channel_data.get('Stream name', 'Unknown Channel')}") + block.append(f"Logo = {channel_data.get('Logo', '')}") + block.append(f"EPG id = {channel_data.get('EPG id', '')}") + block.append(f"Stream URL = {channel_data.get('Stream URL', '')}") + return "\n".join(block) \ No newline at end of file