my-private-iptv-m3u/scripts/playlist_builder.py

177 lines
6.3 KiB
Python
Raw Normal View History

2025-06-28 06:51:03 +02:00
#!/usr/bin/env python3
"""
IPTV Repository Monthly Maintenance
Run automatically by workflow
"""
import os
import shutil
import gzip
from pathlib import Path
from datetime import datetime, timedelta
def monthly_maintenance():
"""Run monthly maintenance tasks."""
print("🧹 IPTV Repository Monthly Maintenance")
print("=" * 40)
root_path = Path.cwd()
actions = []
# 1. Compress old backups
print("1. Compressing old backups...")
backups_dir = root_path / 'backups'
cutoff_date = datetime.now() - timedelta(days=7)
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()
playlist_builder.py
--------------------
2025-06-27 23:30:51 +02:00
"""
Playlist Builder - Generates the final M3U playlist
"""
import logging
from typing import Dict, List, Tuple
class PlaylistBuilder:
"""Generate M3U playlist files."""
def __init__(self, config):
self.config = config
self.logger = logging.getLogger(__name__)
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, {}
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