Update scripts/file_manager.py
All checks were successful
Generate M3U Playlist with Auto-Organization / build-and-organize (push) Successful in 36s
All checks were successful
Generate M3U Playlist with Auto-Organization / build-and-organize (push) Successful in 36s
This commit is contained in:
parent
71d03261c0
commit
4659edae06
1 changed files with 139 additions and 287 deletions
|
|
@ -1,289 +1,141 @@
|
||||||
# 📺 IPTV Playlist Manager
|
"""
|
||||||
|
File Manager - Handles file operations, backups, and channel loading
|
||||||
|
"""
|
||||||
|
|
||||||
[](../../actions)
|
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.
|
class FileManager:
|
||||||
|
"""Manage file operations with backup and rotation."""
|
||||||
|
|
||||||
## ✨ **What Makes This Special**
|
def __init__(self, config):
|
||||||
|
self.config = config
|
||||||
|
self.logger = logging.getLogger(__name__)
|
||||||
|
self.backup_dir = Path("backups")
|
||||||
|
self.backup_dir.mkdir(exist_ok=True)
|
||||||
|
|
||||||
- 🌍 **Smart Country Detection** - Automatically organizes channels by country using advanced pattern matching
|
def create_backup(self, file_path: str) -> Optional[Path]:
|
||||||
- 🎬 **Quality Recognition** - Detects and labels 4K, FHD, HD, and SD streams
|
"""Create timestamped backup with rotation."""
|
||||||
- 🔄 **Intelligent Deduplication** - Removes duplicates using signature-based matching
|
if not self.config.settings.get('create_backup', True):
|
||||||
- 📊 **Professional Reporting** - Detailed statistics and processing logs
|
return None
|
||||||
- 🛡️ **Content Filtering** - Optional adult content filtering
|
|
||||||
- ⚡ **Lightning Fast** - Processes 1000+ channels in seconds
|
|
||||||
|
|
||||||
## 🚀 Quick Start
|
file_path = Path(file_path)
|
||||||
|
if not file_path.exists():
|
||||||
|
return None
|
||||||
|
|
||||||
### 📥 **Download Your Playlist**
|
timestamp = datetime.now().strftime('%Y%m%d_%H%M%S')
|
||||||
- **[📺 Download playlist.m3u](./playlist.m3u)** - Ready to use in any IPTV player
|
backup_path = self.backup_dir / f"{file_path.stem}_{timestamp}{file_path.suffix}"
|
||||||
- **[📋 View Channel List](./channels.txt)** - See all available channels
|
|
||||||
- **[📊 Check Reports](./reports/daily/)** - View processing history and statistics
|
|
||||||
|
|
||||||
### ⚡ **Add Channels (3 Easy Ways)**
|
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
|
||||||
|
|
||||||
#### Method 1: Upload M3U File (Bulk Import) ⭐ **Recommended**
|
def _cleanup_old_backups(self, base_name: str):
|
||||||
1. Get your M3U file from your IPTV provider
|
"""Remove old backups, keeping only the most recent ones."""
|
||||||
2. Upload it to this repository as `bulk_import.m3u`
|
max_backups = self.config.settings.get('max_backups', 5)
|
||||||
3. Commit the file - **automatic processing begins!**
|
|
||||||
4. Check reports for import results with country detection
|
|
||||||
|
|
||||||
#### Method 2: Edit Channels Manually
|
backup_files = sorted(
|
||||||
1. Click **[channels.txt](./channels.txt)** and edit it
|
[f for f in self.backup_dir.glob(f"{base_name}_*") if f.is_file()],
|
||||||
2. Add channels using this format:
|
key=lambda x: x.stat().st_mtime,
|
||||||
```
|
reverse=True
|
||||||
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
|
for old_backup in backup_files[max_backups:]:
|
||||||
Stream name = CNN International
|
try:
|
||||||
Logo = https://example.com/cnn-logo.png
|
old_backup.unlink()
|
||||||
EPG id = cnn.us
|
self.logger.debug(f"Removed old backup: {old_backup}")
|
||||||
Stream URL = http://your-stream-url-here
|
except Exception as e:
|
||||||
```
|
self.logger.warning(f"Could not remove old backup {old_backup}: {e}")
|
||||||
3. Commit your changes
|
|
||||||
|
|
||||||
#### Method 3: Use the Template
|
def load_all_channels(self) -> List[Dict]:
|
||||||
1. Check **[templates/channel_template.txt](./templates/channel_template.txt)** for the exact format
|
"""Load all channels from the channels file."""
|
||||||
2. Copy the template and fill in your channel details
|
if not os.path.exists(self.config.channels_file):
|
||||||
3. Add to channels.txt
|
self.logger.info("No channels.txt file found")
|
||||||
|
return []
|
||||||
|
|
||||||
## 📁 Repository Structure
|
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())
|
||||||
📦 Your IPTV Repository
|
channels = []
|
||||||
├── 📺 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)
|
for block in channel_blocks:
|
||||||
|
if block.strip():
|
||||||
|
channel = self._parse_channel_block(block)
|
||||||
|
if channel:
|
||||||
|
channels.append(channel)
|
||||||
|
|
||||||
The system **intelligently** processes your channels through this pipeline:
|
self.logger.info(f"Loaded {len(channels)} channels from file")
|
||||||
|
return channels
|
||||||
|
|
||||||
```
|
except Exception as e:
|
||||||
📥 Input (M3U/Manual) → 🔍 Parse & Validate → 🌍 Country Detection
|
self.logger.error(f"Error loading channels: {e}")
|
||||||
↓
|
return []
|
||||||
🎬 Quality Detection → 🔄 Duplicate Removal → 📺 Generate Clean M3U
|
|
||||||
↓
|
|
||||||
📊 Create Reports → ✅ Done!
|
|
||||||
```
|
|
||||||
|
|
||||||
### **Advanced Features:**
|
def _parse_channel_block(self, block: str) -> Optional[Dict]:
|
||||||
1. **🌍 Smart Country Detection** - Uses 500+ patterns to detect countries from channel names
|
"""Parse a channel block from channels.txt."""
|
||||||
2. **🎬 Quality Recognition** - Automatically detects 4K, FHD, HD, SD quality
|
channel_data = {}
|
||||||
3. **🔍 Intelligent Deduplication** - Advanced signature matching prevents duplicates
|
lines = block.strip().split('\n')
|
||||||
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**
|
for line in lines:
|
||||||
|
if '=' in line:
|
||||||
|
key, value = line.split('=', 1)
|
||||||
|
channel_data[key.strip()] = value.strip()
|
||||||
|
|
||||||
The system automatically detects and organizes channels from:
|
return channel_data if channel_data else None
|
||||||
|
|
||||||
| Region | Countries | Examples |
|
def save_channels(self, channels: List[Dict]) -> bool:
|
||||||
|--------|-----------|----------|
|
"""Save channels to the channels.txt file."""
|
||||||
| **🇺🇸 North America** | USA, Canada | ESPN, CNN, CBC, TSN |
|
try:
|
||||||
| **🇪🇺 Europe** | UK, Germany, France, Spain, Italy, Netherlands | BBC, Sky, ARD, TF1 |
|
# Create backup first
|
||||||
| **🌏 Other** | Australia, Brazil, Arabic | ABC AU, Globo, MBC |
|
self.create_backup(self.config.channels_file)
|
||||||
|
|
||||||
**Detection Methods:**
|
with open(self.config.channels_file, 'w', encoding='utf-8') as f:
|
||||||
- **Prefixes**: `us:`, `uk:`, `[US]`, `(CA)`
|
for i, channel in enumerate(channels):
|
||||||
- **Channel Names**: ESPN → 🇺🇸, BBC → 🇬🇧, CBC → 🇨🇦
|
if i > 0:
|
||||||
- **Network Patterns**: Sky Sports → 🇬🇧, Fox Sports → 🇺🇸
|
f.write("\n\n")
|
||||||
- **Quality Tags**: Automatic 4K/FHD/HD/SD detection
|
f.write(self._convert_to_channels_txt_block(channel))
|
||||||
|
|
||||||
## 🛠️ Advanced Configuration
|
self.logger.info(f"Saved {len(channels)} channels to file")
|
||||||
|
return True
|
||||||
|
|
||||||
### **Smart Country Detection** (`config/patterns.json`)
|
except Exception as e:
|
||||||
The system includes **500+ detection patterns** for accurate categorization:
|
self.logger.error(f"Error saving channels: {e}")
|
||||||
```json
|
return False
|
||||||
{
|
|
||||||
"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`)
|
def append_channels(self, new_channels: List[Dict]) -> bool:
|
||||||
Force specific channels into custom groups:
|
"""Append new channels to existing channels file."""
|
||||||
```json
|
try:
|
||||||
{
|
# Load existing channels
|
||||||
"ESPN": "🇺🇸 United States",
|
existing_channels = self.load_all_channels()
|
||||||
"Fox Sports": "🇺🇸 United States",
|
|
||||||
"BBC News": "🇬🇧 United Kingdom"
|
|
||||||
}
|
|
||||||
```
|
|
||||||
|
|
||||||
### **System Settings** (`config/settings.json`)
|
# Combine with new channels
|
||||||
Customize processing behavior:
|
all_channels = existing_channels + new_channels
|
||||||
```json
|
|
||||||
{
|
|
||||||
"remove_duplicates": true,
|
|
||||||
"auto_detect_country": true,
|
|
||||||
"detect_quality": true,
|
|
||||||
"skip_adult_content": true,
|
|
||||||
"sort_channels": true,
|
|
||||||
"backup_before_import": true
|
|
||||||
}
|
|
||||||
```
|
|
||||||
|
|
||||||
## 📊 **Monitoring & Statistics**
|
# Save combined channels
|
||||||
|
return self.save_channels(all_channels)
|
||||||
|
|
||||||
### **📈 Live Reports**
|
except Exception as e:
|
||||||
- **[📊 Latest Report](./reports/daily/)** - Processing statistics and country breakdown
|
self.logger.error(f"Error appending channels: {e}")
|
||||||
- **[📋 Processing Logs](./reports/logs/playlist_update.log)** - Detailed operation logs
|
return False
|
||||||
- **[📥 Import History](./reports/logs/import_history.log)** - Import tracking
|
|
||||||
- **[❌ Error Tracking](./reports/logs/error.log)** - Issue monitoring
|
|
||||||
|
|
||||||
### **📈 What You Get**
|
def _convert_to_channels_txt_block(self, channel_data: Dict) -> str:
|
||||||
- **Channel Counts** - Total and per-country statistics
|
"""Convert to channels.txt format."""
|
||||||
- **Quality Distribution** - 4K/FHD/HD/SD breakdown
|
block = []
|
||||||
- **Processing Stats** - Import success rates, duplicates removed
|
block.append(f"Group = {channel_data.get('Group', 'Uncategorized')}")
|
||||||
- **Country Detection** - Accuracy and coverage metrics
|
block.append(f"Stream name = {channel_data.get('Stream name', 'Unknown Channel')}")
|
||||||
- **Performance Data** - Processing times and efficiency
|
block.append(f"Logo = {channel_data.get('Logo', '')}")
|
||||||
|
block.append(f"EPG id = {channel_data.get('EPG id', '')}")
|
||||||
## 🎮 **Using Your Playlist**
|
block.append(f"Stream URL = {channel_data.get('Stream URL', '')}")
|
||||||
|
return "\n".join(block)
|
||||||
### **📱 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*
|
|
||||||
Loading…
Add table
Add a link
Reference in a new issue