"""Download history API endpoints.""" from datetime import datetime, timedelta from quart import Blueprint, request, jsonify, current_app, Response from sqlalchemy import select, func, and_, or_, desc from sqlalchemy.ext.asyncio import AsyncSession from sqlalchemy.orm import selectinload import json from app.models.base import utcnow from app.models.download import Download, DownloadStatus from app.models.source import Source from app.tasks.downloads import process_download bp = Blueprint("downloads", __name__) @bp.route("", methods=["GET"]) async def list_downloads(): """List download history with filtering and pagination.""" # Query parameters source_id = request.args.get("source_id", type=int) status = request.args.get("status") from_date = request.args.get("from_date") to_date = request.args.get("to_date") page = int(request.args.get("page", 1)) per_page = min(int(request.args.get("per_page", 50)), 100) async with current_app.db_engine.connect() as conn: session = AsyncSession(bind=conn) # Build query - eager load source and subscription for labeling query = select(Download).options( selectinload(Download.source).selectinload(Source.subscription) ) count_query = select(func.count()).select_from(Download) filters = [] if source_id: filters.append(Download.source_id == source_id) if status: filters.append(Download.status == status) if from_date: filters.append(Download.created_at >= datetime.fromisoformat(from_date)) if to_date: filters.append(Download.created_at <= datetime.fromisoformat(to_date)) if filters: query = query.where(and_(*filters)) count_query = count_query.where(and_(*filters)) # Get total count total_result = await session.execute(count_query) total = total_result.scalar() # Apply pagination and ordering query = query.order_by(Download.created_at.desc()) query = query.offset((page - 1) * per_page).limit(per_page) result = await session.execute(query) downloads = result.scalars().all() # Include source info with each download items = [] for d in downloads: item = d.to_dict() if d.source: item["subscription_name"] = d.source.subscription.name if d.source.subscription else None item["platform"] = d.source.platform else: item["subscription_name"] = None item["platform"] = None items.append(item) return jsonify({ "items": items, "total": total, "page": page, "per_page": per_page, "pages": (total + per_page - 1) // per_page, }) @bp.route("/", methods=["GET"]) async def get_download(download_id: int): """Get download details.""" async with current_app.db_engine.connect() as conn: session = AsyncSession(bind=conn) result = await session.execute(select(Download).where(Download.id == download_id)) download = result.scalar() if not download: return jsonify({"error": "Download not found"}), 404 # Include source info if available response = download.to_dict() if download.source_id: source_result = await session.execute( select(Source) .options(selectinload(Source.subscription)) .where(Source.id == download.source_id) ) source = source_result.scalar() if source: response["source"] = { "id": source.id, "subscription_name": source.subscription.name if source.subscription else None, "platform": source.platform, } return jsonify(response) @bp.route("//retry", methods=["POST"]) async def retry_download(download_id: int): """Retry a failed download.""" async with current_app.db_engine.connect() as conn: session = AsyncSession(bind=conn) result = await session.execute(select(Download).where(Download.id == download_id)) download = result.scalar() if not download: return jsonify({"error": "Download not found"}), 404 if download.status != DownloadStatus.FAILED: return jsonify({"error": "Can only retry failed downloads"}), 400 # Reset status to queued download.status = DownloadStatus.QUEUED download.error_type = None download.error_message = None await session.commit() # Queue Celery task to retry the download task = process_download.delay(download_id) current_app.logger.info(f"Queued retry for download: {download_id}") return jsonify({ "message": "Retry queued", "download_id": download_id, "task_id": task.id, }), 202 @bp.route("/recent-activity", methods=["GET"]) async def recent_activity(): """Get recent noteworthy downloads: failures and completions with new files. These are downloads that warrant user attention, not routine "no new content" runs. Results are limited to the last 7 days by default. """ limit = min(int(request.args.get("limit", 10)), 50) days = int(request.args.get("days", 7)) from datetime import timedelta cutoff = utcnow() - timedelta(days=days) async with current_app.db_engine.connect() as conn: session = AsyncSession(bind=conn) # Get failed downloads OR completed with files (noteworthy activity) # Eager load source and subscription for labeling from sqlalchemy import or_ query = select(Download).options( selectinload(Download.source).selectinload(Source.subscription) ).where( and_( Download.created_at >= cutoff, or_( Download.status == DownloadStatus.FAILED, and_( Download.status == DownloadStatus.COMPLETED, Download.file_count > 0 ) ) ) ).order_by(Download.created_at.desc()).limit(limit) result = await session.execute(query) downloads = result.scalars().all() # Include source info with each download items = [] for d in downloads: item = d.to_dict() if d.source: item["subscription_name"] = d.source.subscription.name if d.source.subscription else None item["platform"] = d.source.platform else: item["subscription_name"] = None item["platform"] = None items.append(item) return jsonify({ "items": items, "count": len(downloads), }) @bp.route("/reset-orphaned", methods=["POST"]) async def reset_orphaned_downloads(): """Reset orphaned running downloads back to queued. Jobs can get stuck in 'running' status if the worker crashes. This endpoint resets them so they can be retried. """ from app.tasks.maintenance import reset_orphaned_jobs # Allow specifying threshold, default to 0 to reset ALL running jobs data = await request.get_json() or {} threshold_minutes = data.get("threshold_minutes", 0) # 0 = reset all running try: if threshold_minutes == 0: # Reset all running jobs immediately (synchronous for immediate feedback) from app.tasks.maintenance import _reset_all_running_jobs_async import asyncio # Run in the current event loop result = await _reset_all_running_jobs_async() current_app.logger.info(f"Manual reset of running jobs: {result}") return jsonify({ "message": f"Reset {result['reset_count']} orphaned running jobs", "reset_count": result['reset_count'], }) else: # Queue the task for threshold-based reset task = reset_orphaned_jobs.delay(threshold_minutes) return jsonify({ "message": "Orphaned job reset queued", "task_id": task.id, "threshold_minutes": threshold_minutes, }), 202 except Exception as e: current_app.logger.error(f"Failed to reset orphaned jobs: {e}") return jsonify({"error": str(e)}), 500 @bp.route("/requeue-stale", methods=["POST"]) async def requeue_stale_downloads(): """Re-queue Celery tasks for downloads stuck in 'queued' status. Downloads can get stuck in 'queued' database status if the Celery task was lost (e.g., sent to wrong Redis DB, Redis restart, etc.). This endpoint re-sends the Celery tasks for those jobs. """ from app.tasks.maintenance import requeue_stale_jobs, _requeue_all_queued_jobs_async data = await request.get_json() or {} threshold_minutes = data.get("threshold_minutes", 0) # 0 = re-queue ALL queued jobs try: if threshold_minutes == 0: # Re-queue all queued jobs immediately (synchronous for immediate feedback) result = await _requeue_all_queued_jobs_async() current_app.logger.info(f"Manual re-queue of all queued jobs: {result}") return jsonify({ "message": f"Re-queued {result['requeued_count']} queued jobs", "requeued_count": result['requeued_count'], }) else: # Queue the task for threshold-based re-queue task = requeue_stale_jobs.delay(threshold_minutes) return jsonify({ "message": "Stale job re-queue queued", "task_id": task.id, "threshold_minutes": threshold_minutes, }), 202 except Exception as e: current_app.logger.error(f"Failed to re-queue stale jobs: {e}") return jsonify({"error": str(e)}), 500 @bp.route("/stats", methods=["GET"]) async def get_stats(): """Get download statistics.""" period = request.args.get("period", "week") # day, week, month async with current_app.db_engine.connect() as conn: session = AsyncSession(bind=conn) # Total counts by status status_query = select( Download.status, func.count(Download.id) ).group_by(Download.status) status_result = await session.execute(status_query) status_counts = {row[0]: row[1] for row in status_result} # Counts by platform (through source) platform_query = select( Source.platform, func.count(Download.id) ).join(Source, Download.source_id == Source.id).group_by(Source.platform) platform_result = await session.execute(platform_query) platform_counts = {row[0]: row[1] for row in platform_result} # Recent downloads count recent_query = select(func.count()).select_from(Download).where( Download.status == DownloadStatus.COMPLETED ) recent_result = await session.execute(recent_query) recent_completed = recent_result.scalar() return jsonify({ "total": sum(status_counts.values()), "by_status": status_counts, "by_platform": platform_counts, "completed": status_counts.get(DownloadStatus.COMPLETED, 0), "failed": status_counts.get(DownloadStatus.FAILED, 0), "pending": status_counts.get(DownloadStatus.PENDING, 0), "queued": status_counts.get(DownloadStatus.QUEUED, 0), "running": status_counts.get(DownloadStatus.RUNNING, 0), }) @bp.route("/export-failed-logs", methods=["GET"]) async def export_failed_logs(): """Export failed download logs for error classification analysis. Returns a JSON file that can be fed to Claude to improve error recognition patterns. Query params: limit: Maximum number of records (default 50, max 200) days: Only include failures from last N days (default 30) error_type: Filter by specific error type (e.g., not_found, auth_error) include_success: Also include some successful "no new content" for comparison """ limit = min(int(request.args.get("limit", 50)), 200) days = int(request.args.get("days", 30)) error_type = request.args.get("error_type") include_success = request.args.get("include_success", "").lower() in ("true", "1", "yes") cutoff_date = utcnow() - timedelta(days=days) async with current_app.db_engine.connect() as conn: session = AsyncSession(bind=conn) # Build query for failed downloads (eager load source for platform) query = select(Download).options(selectinload(Download.source)).where( and_( Download.status == DownloadStatus.FAILED, Download.created_at >= cutoff_date, ) ) if error_type: query = query.where(Download.error_type == error_type) # Order by most recent first query = query.order_by(desc(Download.created_at)).limit(limit) result = await session.execute(query) failed_downloads = result.scalars().all() # Optionally include some successful NO_NEW_CONTENT for comparison success_downloads = [] if include_success: success_query = select(Download).options(selectinload(Download.source)).where( and_( 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_result = await session.execute(success_query) success_downloads = success_result.scalars().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": 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, } # Return as downloadable JSON file response = Response( json.dumps(output, indent=2, default=str), mimetype="application/json", headers={ "Content-Disposition": f"attachment; filename=failed_logs_export_{datetime.now().strftime('%Y%m%d_%H%M%S')}.json" } ) return response 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. """