Bryan Van Deusen
233 lines
8.0 KiB
Python
233 lines
8.0 KiB
Python
#!/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()
|