Add scripts/report_generator.py

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)