#!/usr/bin/env python3 """ Export failed download logs for error classification analysis. Outputs a JSON file that can be fed to Claude to improve error recognition patterns. Usage: python scripts/export_failed_logs.py [options] Options: --limit N Maximum number of records to export (default: 50) --days N Only include failures from last N days (default: 30) --error-type TYPE Filter by specific error type (e.g., not_found, auth_error) --output FILE Output file path (default: failed_logs_export.json) --include-success Also include some successful "no new content" for comparison """ import argparse import json import sys from datetime import datetime, timedelta from pathlib import Path # Add parent directory to path for imports sys.path.insert(0, str(Path(__file__).parent.parent)) from sqlalchemy import create_engine, desc, or_ from sqlalchemy.orm import sessionmaker from app.models.download import Download, DownloadStatus from app.config import get_settings def export_failed_logs( limit: int = 50, days: int = 30, error_type: str | None = None, include_success: bool = False, output_path: str = "failed_logs_export.json", ): """Export failed download logs to JSON for analysis.""" settings = get_settings() engine = create_engine(settings.database_url) Session = sessionmaker(bind=engine) session = Session() try: cutoff_date = datetime.utcnow() - timedelta(days=days) # Build query for failed downloads query = session.query(Download).filter( Download.status == DownloadStatus.FAILED, Download.created_at >= cutoff_date, ) if error_type: query = query.filter(Download.error_type == error_type) # Order by most recent first failed_downloads = query.order_by(desc(Download.created_at)).limit(limit).all() # Optionally include some successful NO_NEW_CONTENT for comparison success_downloads = [] if include_success: success_query = session.query(Download).filter( Download.status == DownloadStatus.COMPLETED, Download.created_at >= cutoff_date, Download.file_count == 0, # No new content cases ).order_by(desc(Download.created_at)).limit(limit // 4) success_downloads = success_query.all() # Format for export exports = [] for download in failed_downloads: exports.append(format_download_for_export(download, "failed")) for download in success_downloads: exports.append(format_download_for_export(download, "success_no_new_content")) # Build output document output = { "export_info": { "exported_at": datetime.utcnow().isoformat(), "total_records": len(exports), "failed_count": len(failed_downloads), "success_comparison_count": len(success_downloads), "days_included": days, "error_type_filter": error_type, }, "analysis_prompt": generate_analysis_prompt(), "downloads": exports, } # Write to file with open(output_path, "w", encoding="utf-8") as f: json.dump(output, f, indent=2, default=str) print(f"Exported {len(exports)} records to {output_path}") print(f" - Failed: {len(failed_downloads)}") if include_success: print(f" - Success (no new content): {len(success_downloads)}") # Print summary by error type error_types = {} for download in failed_downloads: et = download.error_type or "unknown" error_types[et] = error_types.get(et, 0) + 1 print("\nError type breakdown:") for et, count in sorted(error_types.items(), key=lambda x: -x[1]): print(f" {et}: {count}") return output_path finally: session.close() def format_download_for_export(download: Download, classification: str) -> dict: """Format a download record for export.""" # Extract logs from metadata metadata = download.metadata_ or {} stdout = metadata.get("stdout") or "" stderr = metadata.get("stderr") or "" # Truncate very long logs but keep enough for analysis max_log_length = 50000 # 50KB per log field if stdout and len(stdout) > max_log_length: stdout = stdout[:max_log_length] + f"\n\n[... truncated, total length: {len(stdout)} chars ...]" if stderr and len(stderr) > max_log_length: stderr = stderr[:max_log_length] + f"\n\n[... truncated, total length: {len(stderr)} chars ...]" # Get platform from source if available platform = None if download.source: platform = download.source.platform return { "id": download.id, "classification": classification, "assigned_error_type": download.error_type, "assigned_error_message": download.error_message, "platform": platform, "url": download.url, "status": download.status, # Already a string, not enum "file_count": download.file_count, "created_at": download.created_at.isoformat() if download.created_at else None, "duration_seconds": metadata.get("duration_seconds"), "logs": { "stdout": stdout, "stderr": stderr, }, "user_feedback": None, # User can fill this in: "correct", "false_positive", "false_negative", "wrong_type" "suggested_error_type": None, # User can suggest what it should be "notes": None, # User can add context } def generate_analysis_prompt() -> str: """Generate a prompt for Claude to analyze the logs.""" return """ ## Error Log Analysis Task I'm providing you with download logs from a gallery-dl based downloader. Each record includes: - The error_type that was assigned by the current classification logic - The actual stdout/stderr logs from gallery-dl - Platform and URL information Please analyze these logs and identify: 1. **False Positives**: Cases where an error was reported but the download actually succeeded - Look for: successful HTTP responses (200), "skipping" messages indicating archive deduplication worked 2. **False Negatives**: Cases classified as one error type that should be another - Example: classified as "not_found" but logs show authentication issues 3. **Pattern Improvements**: Suggest more specific patterns to detect each error type - Current patterns may be too broad (matching debug output) or too narrow (missing variations) 4. **New Error Types**: Any failure modes not currently categorized For each issue found, provide: - The download ID - Current classification vs suggested classification - The specific log lines that support your analysis - Suggested pattern improvements (if applicable) Focus on actionable improvements to the _categorize_error() function in gallery_dl.py. """ def main(): parser = argparse.ArgumentParser( description="Export failed download logs for error classification analysis" ) parser.add_argument( "--limit", type=int, default=50, help="Maximum number of failed records to export (default: 50)" ) parser.add_argument( "--days", type=int, default=30, help="Only include failures from last N days (default: 30)" ) parser.add_argument( "--error-type", type=str, default=None, help="Filter by specific error type (e.g., not_found, auth_error)" ) parser.add_argument( "--output", type=str, default="failed_logs_export.json", help="Output file path (default: failed_logs_export.json)" ) parser.add_argument( "--include-success", action="store_true", help="Also include some successful 'no new content' records for comparison" ) args = parser.parse_args() export_failed_logs( limit=args.limit, days=args.days, error_type=args.error_type, include_success=args.include_success, output_path=args.output, ) if __name__ == "__main__": main()