"""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") exclude_superseded = request.args.get("exclude_superseded", "").lower() in ("true", "1", "yes") 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: async with AsyncSession(bind=conn) as session: # 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)) # When filtering failed downloads, optionally exclude ones superseded by a later success if exclude_superseded and status == DownloadStatus.FAILED: filters.append(Download.superseded == False) 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: async with AsyncSession(bind=conn) as session: 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) def _plan_bulk_retry(sorted_failures, sources_by_id, recent_cutoff): """Decide which failed downloads to retry given per-source dedup + recent-success filter. Args: sorted_failures: Download rows sorted by (source_id ASC, created_at DESC) — the first occurrence of each source_id is the most recent failure. sources_by_id: dict mapping source_id -> Source (for `last_success` lookup). recent_cutoff: datetime — any source whose last_success is >= this is skipped. Returns (retry_ids, summary) where summary has skip counts per reason. """ seen_sources = set() retry_ids = [] skipped_dup = 0 skipped_recent = 0 skipped_no_source = 0 for failure in sorted_failures: if failure.source_id is None: skipped_no_source += 1 continue if failure.source_id in seen_sources: skipped_dup += 1 continue seen_sources.add(failure.source_id) source = sources_by_id.get(failure.source_id) if source and source.last_success and source.last_success >= recent_cutoff: skipped_recent += 1 continue retry_ids.append(failure.id) return retry_ids, { "retried": len(retry_ids), "skipped_duplicate_source": skipped_dup, "skipped_recent_success": skipped_recent, "skipped_no_source": skipped_no_source, } @bp.route("/retry-failed-bulk", methods=["POST"]) async def retry_failed_bulk(): """Bulk-retry non-superseded failed downloads, with smart dedup. Applies two filters that the old per-row Dashboard loop missed: 1. Per-source dedup — a source with N historical failures enqueues once. 2. Recent-success skip — a source whose `last_success` is within the recent window (default 24h, overridable via `recent_window_hours`) is considered healthy; we leave its stale failures alone rather than pile on redundant retry work. """ data = await request.get_json() or {} recent_window_hours = max(0, int(data.get("recent_window_hours", 24))) recent_cutoff = utcnow() - timedelta(hours=recent_window_hours) async with current_app.db_engine.connect() as conn: async with AsyncSession(bind=conn) as session: failures_result = await session.execute( select(Download) .where( Download.status == DownloadStatus.FAILED, Download.superseded == False, ) .order_by(Download.source_id.asc(), Download.created_at.desc()) ) failures = list(failures_result.scalars().all()) source_ids = {f.source_id for f in failures if f.source_id is not None} sources_by_id = {} if source_ids: sources_result = await session.execute( select(Source).where(Source.id.in_(source_ids)) ) sources_by_id = {s.id: s for s in sources_result.scalars().all()} retry_ids, summary = _plan_bulk_retry(failures, sources_by_id, recent_cutoff) for failure in failures: if failure.id in retry_ids: failure.status = DownloadStatus.QUEUED failure.error_type = None failure.error_message = None failure.started_at = None failure.completed_at = None await session.commit() for download_id in retry_ids: process_download.delay(download_id) current_app.logger.info( f"retry-failed-bulk: {summary} (recent_window_hours={recent_window_hours})" ) return jsonify({ "message": f"Queued {summary['retried']} retries", **summary, "total_failures_considered": len(failures), "recent_window_hours": recent_window_hours, }) @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: async with AsyncSession(bind=conn) as session: 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 and clear prior-run timestamps download.status = DownloadStatus.QUEUED download.error_type = None download.error_message = None download.started_at = None download.completed_at = 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: async with AsyncSession(bind=conn) as session: # Get failed downloads OR completed with files (noteworthy activity) # Exclude failed downloads that have been superseded by a later success # for the same source - those are stale failures the user doesn't need to see query = select(Download).options( selectinload(Download.source).selectinload(Source.subscription) ).where( and_( Download.created_at >= cutoff, or_( and_( Download.status == DownloadStatus.FAILED, Download.superseded == False, ), 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("/activity-timeline", methods=["GET"]) async def activity_timeline(): """Return per-day counts of completed, failed, and new-files over the last N days. Used by the Dashboard sparkline to show run throughput at a glance. Failed counts exclude superseded rows so the line reflects *meaningful* failures, matching the Recent Activity filter. """ days = max(1, min(int(request.args.get("days", 7)), 90)) cutoff = utcnow() - timedelta(days=days) async with current_app.db_engine.connect() as conn: async with AsyncSession(bind=conn) as session: # Bucket by date (UTC) using func.date(). Group by status so we can # separate completed/failed series. SUM(file_count) gives throughput. day = func.date(Download.created_at).label("day") query = ( select( day, Download.status, func.count(Download.id).label("count"), func.coalesce(func.sum(Download.file_count), 0).label("file_count"), ) .where( Download.created_at >= cutoff, or_( Download.status != DownloadStatus.FAILED, Download.superseded == False, ), ) .group_by(day, Download.status) .order_by(day) ) result = await session.execute(query) rows = result.all() # Pre-fill every day so the sparkline has a point for each bucket # even when nothing happened. today = utcnow().date() buckets = {} for i in range(days): d = (today - timedelta(days=days - 1 - i)).isoformat() buckets[d] = {"date": d, "completed": 0, "failed": 0, "files": 0} for row in rows: d = row.day.isoformat() if hasattr(row.day, "isoformat") else str(row.day) bucket = buckets.setdefault( d, {"date": d, "completed": 0, "failed": 0, "files": 0} ) if row.status == DownloadStatus.COMPLETED: bucket["completed"] += row.count bucket["files"] += int(row.file_count or 0) elif row.status == DownloadStatus.FAILED: bucket["failed"] += row.count series = [buckets[k] for k in sorted(buckets.keys())] totals = { "completed": sum(b["completed"] for b in series), "failed": sum(b["failed"] for b in series), "files": sum(b["files"] for b in series), } return jsonify({"days": days, "series": series, "totals": totals}) @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: async with AsyncSession(bind=conn) as session: # 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() # Active failed: failures not superseded by a later success for the same source active_failed_query = select(func.count()).select_from(Download).where( and_( Download.status == DownloadStatus.FAILED, Download.superseded == False, ) ) active_failed_result = await session.execute(active_failed_query) active_failed = active_failed_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), "active_failed": active_failed, "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: async with AsyncSession(bind=conn) as session: # 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. """