From 8c7ee2416c8749e3f3e8fd1feb8e4f14f78fae59 Mon Sep 17 00:00:00 2001 From: stoney420 Date: Fri, 27 Jun 2025 23:32:25 +0200 Subject: [PATCH] Add scripts/report_generator.py --- scripts/report_generator.py | 133 ++++++++++++++++++++++++++++++++++++ 1 file changed, 133 insertions(+) create mode 100644 scripts/report_generator.py diff --git a/scripts/report_generator.py b/scripts/report_generator.py new file mode 100644 index 0000000..0842b72 --- /dev/null +++ b/scripts/report_generator.py @@ -0,0 +1,133 @@ +""" +Report Generator - Creates comprehensive reports and statistics +""" + +import logging +from datetime import datetime +from pathlib import Path +from typing import Dict, Optional + +class ReportGenerator: + """Generate comprehensive reports and statistics.""" + + def __init__(self, config): + self.config = config + self.logger = logging.getLogger(__name__) + self.reports_dir = Path("reports") + self.reports_dir.mkdir(exist_ok=True) + + def generate_markdown_report(self, stats: Dict, health_results: Dict = None) -> str: + """Generate a detailed Markdown report.""" + timestamp = datetime.now().strftime('%Y-%m-%d %H:%M:%S') + + report_lines = [ + "# IPTV Playlist Generation Report", + f"**Generated:** {timestamp}\n", + "## Summary Statistics", + f"- **Total channels processed:** {stats.get('total_channels', 0)}", + f"- **Valid channels:** {stats.get('valid_channels', 0)}", + f"- **Duplicates removed:** {stats.get('duplicates_removed', 0)}", + f"- **New channels imported:** {stats.get('imported_channels', 0)}", + f"- **Countries detected:** {stats.get('countries_detected', 0)}", + "" + ] + + # Health check section + if health_results: + healthy_count = sum(1 for is_healthy, _ in health_results.values() if is_healthy) + total_checked = len(health_results) + success_rate = (healthy_count / total_checked * 100) if total_checked > 0 else 0 + + report_lines.extend([ + "## Health Check Results", + f"- **Channels checked:** {total_checked}", + f"- **Healthy channels:** {healthy_count}", + f"- **Success rate:** {success_rate:.1f}%", + "" + ]) + + # Failed channels + failed_channels = [name for name, (is_healthy, status) in health_results.items() if not is_healthy] + if failed_channels: + report_lines.extend([ + "### Failed Channels", + *[f"- {channel}" for channel in failed_channels[:10]], # Limit to first 10 + "" + ]) + + # Country distribution + if 'country_distribution' in stats and stats['country_distribution']: + sorted_countries = sorted( + stats['country_distribution'].items(), + key=lambda x: x[1], + reverse=True + ) + + report_lines.extend([ + "## Channel Distribution by Country", + *[f"- **{country}:** {count} channels" for country, count in sorted_countries], + "" + ]) + + # Configuration summary + report_lines.extend([ + "## Configuration", + f"- **Remove duplicates:** {self.config.settings.get('remove_duplicates', 'Unknown')}", + f"- **Auto country detection:** {self.config.settings.get('auto_detect_country', 'Unknown')}", + f"- **Quality detection:** {self.config.settings.get('detect_quality', 'Unknown')}", + f"- **Adult content filtering:** {self.config.settings.get('skip_adult_content', 'Unknown')}", + f"- **Health check enabled:** {self.config.settings.get('enable_health_check', 'Unknown')}", + "" + ]) + + return "\n".join(report_lines) + + def save_report(self, stats: Dict, health_results: Dict = None) -> Optional[Path]: + """Save the report to a file.""" + try: + report_content = self.generate_markdown_report(stats, health_results) + timestamp = datetime.now().strftime('%Y%m%d_%H%M%S') + report_path = self.reports_dir / f"playlist_report_{timestamp}.md" + + with open(report_path, 'w', encoding='utf-8') as f: + f.write(report_content) + + self.logger.info(f"Report saved: {report_path}") + + # Cleanup old reports (keep last 10) + self._cleanup_old_reports() + + return report_path + + except Exception as e: + self.logger.error(f"Failed to save report: {e}") + return None + + def _cleanup_old_reports(self): + """Remove old reports, keeping only the most recent ones.""" + try: + report_files = sorted( + [f for f in self.reports_dir.glob("playlist_report_*.md") if f.is_file()], + key=lambda x: x.stat().st_mtime, + reverse=True + ) + + # Keep last 10 reports + for old_report in report_files[10:]: + old_report.unlink() + self.logger.debug(f"Removed old report: {old_report}") + + except Exception as e: + self.logger.warning(f"Could not cleanup old reports: {e}") + + def log_summary_stats(self, stats: Dict): + """Log a quick summary to console/log.""" + self.logger.info("=" * 50) + self.logger.info("PLAYLIST GENERATION SUMMARY") + self.logger.info("=" * 50) + self.logger.info(f"Total channels: {stats.get('total_channels', 0)}") + self.logger.info(f"Valid channels: {stats.get('valid_channels', 0)}") + self.logger.info(f"Duplicates removed: {stats.get('duplicates_removed', 0)}") + self.logger.info(f"New imports: {stats.get('imported_channels', 0)}") + self.logger.info(f"Countries: {stats.get('countries_detected', 0)}") + self.logger.info("=" * 50) \ No newline at end of file