From 575e20f58c58f2ab56d88a8b38d3e6a567cb6bb9 Mon Sep 17 00:00:00 2001 From: Bryan Van Deusen Date: Wed, 28 Jan 2026 09:32:47 -0500 Subject: [PATCH] ongoing polish in queue handling --- backend/app/api/downloads.py | 41 +++++ backend/app/services/gallery_dl.py | 113 ++++++++++--- backend/app/tasks/celery_app.py | 69 +++++++- backend/app/tasks/downloads.py | 82 +++++++++- backend/app/tasks/maintenance.py | 121 +++++++++++++- frontend/src/services/api.js | 1 + frontend/src/views/Dashboard.vue | 109 ++++++++---- frontend/src/views/Settings.vue | 16 +- summary.md | 255 +++++++++++++++++++++++++++++ 9 files changed, 738 insertions(+), 69 deletions(-) create mode 100644 summary.md diff --git a/backend/app/api/downloads.py b/backend/app/api/downloads.py index 98fd48b..6b2437d 100644 --- a/backend/app/api/downloads.py +++ b/backend/app/api/downloads.py @@ -168,6 +168,45 @@ async def recent_activity(): }) +@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("/stats", methods=["GET"]) async def get_stats(): """Get download statistics.""" @@ -208,4 +247,6 @@ async def get_stats(): "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), }) diff --git a/backend/app/services/gallery_dl.py b/backend/app/services/gallery_dl.py index 0f24b46..0171a71 100644 --- a/backend/app/services/gallery_dl.py +++ b/backend/app/services/gallery_dl.py @@ -173,8 +173,15 @@ class GalleryDLService: }, } - def __init__(self): + def __init__(self, rate_limit: Optional[float] = None): + """Initialize the GalleryDL service. + + Args: + rate_limit: Override for download rate limit (seconds between requests). + If None, uses environment default. + """ self.settings = get_settings() + self._rate_limit = rate_limit if rate_limit is not None else self.settings.download_rate_limit self._base_config = self._load_base_config() def _load_base_config(self) -> dict: @@ -198,8 +205,8 @@ class GalleryDLService: "base-directory": self.settings.download_path, "archive": str(Path(self.settings.config_path) / "archive.sqlite3"), "skip": True, - "sleep": self.settings.download_rate_limit, - "sleep-request": max(1.0, self.settings.download_rate_limit / 2), + "sleep": self._rate_limit, + "sleep-request": max(1.0, self._rate_limit / 2), "retries": 3, "timeout": 30.0, "verify": True, @@ -311,19 +318,23 @@ class GalleryDLService: """Categorize the error based on output and return code.""" combined = f"{stdout} {stderr}".lower() - # Check for "no new content" first - skipped files with return code 1 is normal - if return_code == 1: - # Check if output indicates content was processed (skipped = already downloaded) - if "skipping" in combined or "skip" in combined: - return ErrorType.NO_NEW_CONTENT, "No new content to download" - # Empty output with return code 1 often means nothing to download - if not stdout.strip() or "no results" in combined: + # Check for "no new content" FIRST for any non-zero return code + # Gallery-dl returns 1 when all files are skipped or no new content found + # This check must run before other error checks to avoid false positives + skip_indicators = ["skipping", "skip", "already exists", "archive"] + if any(indicator in combined for indicator in skip_indicators): + # If we see skip messages and no actual error messages, it's just "no new content" + actual_errors = ["error", "failed", "exception", "traceback"] + if not any(err in combined for err in actual_errors): return ErrorType.NO_NEW_CONTENT, "No new content to download" - # Auth errors - be more specific to avoid false positives from debug URLs - # Look for actual error messages, not just keywords that appear in URLs - auth_patterns = [ - "401", + # Empty output with low return codes often means nothing to download + if return_code in (0, 1) and not stdout.strip(): + return ErrorType.NO_NEW_CONTENT, "No new content to download" + + # Auth errors - look for error context, not just HTTP codes that appear in debug logs + # HTTP debug lines look like: '"GET /path HTTP/1.1" 401 None' - we need to detect actual errors + auth_error_patterns = [ "unauthorized", "login required", "please login", @@ -332,28 +343,84 @@ class GalleryDLService: "session expired", "invalid cookie", "cookies are expired", + "cookies have expired", + "not logged in", ] - if any(pattern in combined for pattern in auth_patterns): + # Check for 401 specifically as an HTTP error response (not in URLs) + # Look for patterns like "401" followed by error context or at end of HTTP log line + has_401_error = ( + '" 401 ' in combined or # HTTP response: "GET /path HTTP/1.1" 401 None + "401 unauthorized" in combined or + "error 401" in combined or + "status: 401" in combined or + "status code: 401" in combined + ) + if has_401_error or any(pattern in combined for pattern in auth_error_patterns): return ErrorType.AUTH_ERROR, "Authentication failed - cookies may be expired or invalid" - if "429" in combined or "rate limit" in combined or "too many requests" in combined: + # Rate limiting - look for actual rate limit errors + rate_limit_patterns = [ + '" 429 ', # HTTP response + "rate limit", + "too many requests", + "ratelimit", + "throttl", + ] + if any(pattern in combined for pattern in rate_limit_patterns): return ErrorType.RATE_LIMITED, "Rate limited by server - try increasing sleep time" - if "404" in combined or "not found" in combined: + # 404 Not Found - check for HTTP 404 responses, not just "404" anywhere + not_found_patterns = [ + '" 404 ', # HTTP response + "error 404", + "status: 404", + "not found", + "does not exist", + "no longer available", + ] + if any(pattern in combined for pattern in not_found_patterns): return ErrorType.NOT_FOUND, "URL not found - artist may have changed username or deleted content" - if "timeout" in combined or "connection" in combined or "network" in combined: + # Network errors + network_error_patterns = [ + "timeout", + "timed out", + "connection refused", + "connection reset", + "network unreachable", + "name resolution", + "dns", + "ssl error", + "certificate verify", + ] + if any(pattern in combined for pattern in network_error_patterns): return ErrorType.NETWORK_ERROR, "Network error - check internet connection" - if "403" in combined or "forbidden" in combined or "access denied" in combined: + # 403 Forbidden / Access denied + access_denied_patterns = [ + '" 403 ', # HTTP response + "error 403", + "forbidden", + "access denied", + "permission denied", + "tier required", + "pledge required", + ] + if any(pattern in combined for pattern in access_denied_patterns): return ErrorType.ACCESS_DENIED, "Access denied - may need higher subscription tier" - if "http error" in combined: + # Generic HTTP errors + if "http error" in combined or "httperror" in combined: return ErrorType.HTTP_ERROR, "HTTP error occurred" - if "no suitable" in combined or "unsupported" in combined: + # Unsupported URL + if "no suitable" in combined or "unsupported" in combined or "no extractor" in combined: return ErrorType.UNSUPPORTED_URL, "URL format not supported by gallery-dl" + # If we got here with return code 1 and saw some activity, it might just be no new content + if return_code == 1: + return ErrorType.NO_NEW_CONTENT, "No new content to download" + return ErrorType.UNKNOWN_ERROR, f"Unknown error (return code: {return_code})" def _count_downloaded_files(self, stdout: str) -> int: @@ -575,6 +642,6 @@ class GalleryDLService: "content_types": defaults.get("content_types", ["all"]), "directory_pattern": defaults.get("directory"), "filename_pattern": defaults.get("filename"), - "sleep": self.settings.download_rate_limit, - "sleep_request": max(1.0, self.settings.download_rate_limit / 2), + "sleep": self._rate_limit, + "sleep_request": max(1.0, self._rate_limit / 2), } diff --git a/backend/app/tasks/celery_app.py b/backend/app/tasks/celery_app.py index 8aa27d1..406dde1 100644 --- a/backend/app/tasks/celery_app.py +++ b/backend/app/tasks/celery_app.py @@ -1,12 +1,51 @@ """Celery application configuration.""" +import logging from celery import Celery from celery.schedules import crontab from app.config import get_settings +logger = logging.getLogger(__name__) settings = get_settings() + +def _get_parallel_limit_from_db(): + """Try to get parallel_limit from database, fall back to environment. + + This is called once at Celery startup. Changes require worker restart. + """ + try: + import asyncio + from sqlalchemy import select + from sqlalchemy.ext.asyncio import create_async_engine, AsyncSession, async_sessionmaker + + async def _fetch(): + engine = create_async_engine(settings.async_database_url, echo=False) + session_factory = async_sessionmaker(engine, class_=AsyncSession, expire_on_commit=False) + try: + async with session_factory() as session: + from app.models.setting import Setting, DEFAULT_SETTINGS + result = await session.execute( + select(Setting).where(Setting.key == "download.parallel_limit") + ) + setting = result.scalar() + if setting: + return setting.value + return DEFAULT_SETTINGS.get("download.parallel_limit", settings.download_parallel_limit) + finally: + await engine.dispose() + + return asyncio.run(_fetch()) + except Exception as e: + logger.warning(f"Could not read parallel_limit from database, using environment: {e}") + return settings.download_parallel_limit + + +# Get parallel limit (database takes precedence over environment) +parallel_limit = _get_parallel_limit_from_db() +logger.info(f"Celery worker concurrency set to: {parallel_limit}") + celery_app = Celery( "gallery_subscriber", broker=settings.redis_url, @@ -24,7 +63,7 @@ celery_app.conf.update( task_track_started=True, task_time_limit=3600, # 1 hour max per task worker_prefetch_multiplier=1, # Process one task at a time - worker_concurrency=settings.download_parallel_limit, + worker_concurrency=parallel_limit, ) # Celery Beat schedule for periodic tasks @@ -44,4 +83,32 @@ celery_app.conf.beat_schedule = { "task": "tasks.update_storage_stats", "schedule": crontab(minute="*/30"), # Every 30 minutes }, + "reset-orphaned-jobs": { + "task": "tasks.reset_orphaned_jobs", + "schedule": crontab(minute="*/15"), # Every 15 minutes + }, } + + +# Worker startup signal - reset any orphaned running jobs +@celery_app.on_after_configure.connect +def setup_worker_startup(sender, **kwargs): + """Called after Celery is configured but before workers start processing.""" + pass # Placeholder - actual reset happens in worker_ready signal + + +from celery.signals import worker_ready + +@worker_ready.connect +def on_worker_ready(sender, **kwargs): + """Called when a worker is ready to accept tasks. + + Resets any jobs stuck in 'running' status from previous runs. + This handles the case where workers crashed or were restarted. + """ + try: + from app.tasks.maintenance import reset_all_running_jobs_sync + result = reset_all_running_jobs_sync() + logger.info(f"Worker startup: {result}") + except Exception as e: + logger.error(f"Failed to reset orphaned jobs on worker startup: {e}") diff --git a/backend/app/tasks/downloads.py b/backend/app/tasks/downloads.py index 1d79348..6673ed7 100644 --- a/backend/app/tasks/downloads.py +++ b/backend/app/tasks/downloads.py @@ -15,6 +15,7 @@ from app.models.base import utcnow from app.models.source import Source from app.models.subscription import Subscription from app.models.download import Download, DownloadStatus, ErrorType as DBErrorType +from app.models.setting import Setting, DEFAULT_SETTINGS from app.services.gallery_dl import GalleryDLService, SourceConfig, DownloadResult from app.services.credential_manager import CredentialManager from app.events import publish_event @@ -41,6 +42,46 @@ async def _cleanup_engine(engine): await engine.dispose() +async def _get_db_setting(session: AsyncSession, key: str, default=None): + """Get a setting value from the database, falling back to defaults. + + Args: + session: Database session + key: Setting key (e.g., 'download.rate_limit') + default: Fallback if not in database or DEFAULT_SETTINGS + + Returns: + The setting value + """ + result = await session.execute( + select(Setting).where(Setting.key == key) + ) + setting = result.scalar() + + if setting: + return setting.value + + # Fall back to DEFAULT_SETTINGS, then to provided default + return DEFAULT_SETTINGS.get(key, default) + + +def _get_db_setting_sync(key: str, default=None): + """Synchronously get a setting value from the database. + + This is used for getting settings in Celery task retry logic. + Creates a fresh connection for each call. + """ + async def _fetch(): + session_factory, engine = get_async_session() + try: + async with session_factory() as session: + return await _get_db_setting(session, key, default) + finally: + await _cleanup_engine(engine) + + return asyncio.run(_fetch()) + + async def _download_source_async(source_id: int, download_id: int = None) -> dict: """Async implementation of source download. @@ -139,6 +180,9 @@ async def _download_source_async(source_id: int, download_id: int = None) -> dic if not cookies_path and source_platform in ["patreon", "subscribestar"]: logger.warning(f"No credentials for {source_platform}, download may fail") + # Get rate limit from database settings + db_rate_limit = await _get_db_setting(session, "download.rate_limit", settings.download_rate_limit) + # Session is now closed - DB connection released finally: await _cleanup_engine(engine) @@ -149,7 +193,7 @@ async def _download_source_async(source_id: int, download_id: int = None) -> dic dl_error = None try: - gdl_service = GalleryDLService() + gdl_service = GalleryDLService(rate_limit=db_rate_limit) dl_result = await gdl_service.download( url=source_url, subscription_name=subscription_name, @@ -271,9 +315,16 @@ async def _scheduled_check_async() -> dict: async with session_factory() as session: now = utcnow() + # Get global schedule interval from database settings + # This is the primary interval that controls how often sources are checked + global_interval = await _get_db_setting( + session, "download.schedule_interval", 3600 # Default 1 hour + ) + logger.debug(f"Using global schedule interval: {global_interval} seconds") + # Find sources due for checking: # - enabled = True - # - last_check is NULL OR (now - last_check) > check_interval + # - last_check is NULL OR (now - last_check) > global_interval # Order by subscription priority, then by when last checked query = ( select(Source) @@ -298,14 +349,15 @@ async def _scheduled_check_async() -> dict: result = await session.execute(query) sources = result.scalars().all() - # Filter by check_interval + # Filter by global schedule interval from Settings + # All sources use the same interval - no per-source overrides due_sources = [] for source in sources: if source.last_check is None: due_sources.append(source) else: time_since_check = (now - source.last_check).total_seconds() - if time_since_check >= source.check_interval: + if time_since_check >= global_interval: due_sources.append(source) logger.info(f"Found {len(due_sources)} sources due for checking") @@ -403,7 +455,7 @@ async def _cleanup_old_downloads_async() -> dict: # Celery tasks that wrap async functions -@celery_app.task(bind=True, max_retries=3) +@celery_app.task(bind=True) def download_source(self, source_id: int, download_id: int = None): """Download new content from a single source. @@ -415,8 +467,14 @@ def download_source(self, source_id: int, download_id: int = None): return asyncio.run(_download_source_async(source_id, download_id=download_id)) except Exception as e: logger.exception(f"Task failed for source {source_id}: {e}") - # Retry with exponential backoff - raise self.retry(exc=e, countdown=60 * (2 ** self.request.retries)) + # Get retry count from database settings + max_retries = _get_db_setting_sync("download.retry_count", 3) + if self.request.retries < max_retries: + # Retry with exponential backoff + raise self.retry(exc=e, countdown=60 * (2 ** self.request.retries), max_retries=max_retries) + else: + logger.error(f"Max retries ({max_retries}) exceeded for source {source_id}") + raise @celery_app.task @@ -437,7 +495,7 @@ def cleanup_old_downloads(): return asyncio.run(_cleanup_old_downloads_async()) -@celery_app.task(bind=True, max_retries=2) +@celery_app.task(bind=True) def process_download(self, download_id: int): """Process a single download record (for retries). @@ -469,4 +527,10 @@ def process_download(self, download_id: int): return asyncio.run(_process()) except Exception as e: logger.exception(f"Task failed for download {download_id}: {e}") - raise self.retry(exc=e, countdown=60 * (2 ** self.request.retries)) + # Get retry count from database settings + max_retries = _get_db_setting_sync("download.retry_count", 3) + if self.request.retries < max_retries: + raise self.retry(exc=e, countdown=60 * (2 ** self.request.retries), max_retries=max_retries) + else: + logger.error(f"Max retries ({max_retries}) exceeded for download {download_id}") + raise diff --git a/backend/app/tasks/maintenance.py b/backend/app/tasks/maintenance.py index 5934f1a..56286b4 100644 --- a/backend/app/tasks/maintenance.py +++ b/backend/app/tasks/maintenance.py @@ -3,20 +3,26 @@ import asyncio import logging import os -from datetime import datetime +from datetime import datetime, timedelta from pathlib import Path -from sqlalchemy import select +from sqlalchemy import select, update from sqlalchemy.ext.asyncio import create_async_engine, AsyncSession, async_sessionmaker from app.config import get_settings from app.tasks.celery_app import celery_app from app.models.setting import Setting, STORAGE_STATS_SETTING +from app.models.download import Download, DownloadStatus +from app.models.base import utcnow logger = logging.getLogger(__name__) settings = get_settings() +# Maximum time a job can be in "running" state before considered orphaned +# This should be longer than the task_time_limit (1 hour) to account for legitimate long runs +STALE_RUNNING_THRESHOLD_MINUTES = 90 + def get_async_session(): """Get async session factory for Celery tasks.""" @@ -125,3 +131,114 @@ def update_storage_stats() -> dict: result = asyncio.run(_update_storage_stats_async()) logger.info(f"Storage stats updated: {result}") return result + + +async def _reset_orphaned_running_jobs_async(threshold_minutes: int = None) -> dict: + """Reset jobs stuck in 'running' status back to 'queued'. + + Jobs can get stuck in 'running' status if the worker crashes or restarts + while processing them. This function detects and resets these orphaned jobs. + + Args: + threshold_minutes: Jobs running longer than this are considered orphaned. + If None, uses STALE_RUNNING_THRESHOLD_MINUTES. + + Returns: + Dict with count of reset jobs + """ + if threshold_minutes is None: + threshold_minutes = STALE_RUNNING_THRESHOLD_MINUTES + + session_factory, engine = get_async_session() + + try: + async with session_factory() as session: + now = utcnow() + cutoff = now - timedelta(minutes=threshold_minutes) + + # Find running jobs that started before the cutoff + # OR running jobs with no started_at (shouldn't happen but handle it) + query = select(Download).where( + Download.status == DownloadStatus.RUNNING, + (Download.started_at < cutoff) | (Download.started_at == None) + ) + + result = await session.execute(query) + orphaned_jobs = result.scalars().all() + + reset_count = 0 + for job in orphaned_jobs: + job.status = DownloadStatus.QUEUED + job.started_at = None + job.error_message = f"Reset from orphaned running state (was running since {job.started_at})" + reset_count += 1 + logger.info(f"Reset orphaned job {job.id} (URL: {job.url[:50]}...) to QUEUED") + + if reset_count > 0: + await session.commit() + logger.info(f"Reset {reset_count} orphaned running jobs to QUEUED") + else: + logger.debug("No orphaned running jobs found") + + return {"reset_count": reset_count, "threshold_minutes": threshold_minutes} + + finally: + await _cleanup_engine(engine) + + +async def _reset_all_running_jobs_async() -> dict: + """Reset ALL jobs in 'running' status back to 'queued'. + + This is useful on worker startup to clean up any jobs that were + interrupted by a previous shutdown/crash. + + Returns: + Dict with count of reset jobs + """ + session_factory, engine = get_async_session() + + try: + async with session_factory() as session: + # Find all running jobs + query = select(Download).where(Download.status == DownloadStatus.RUNNING) + result = await session.execute(query) + running_jobs = result.scalars().all() + + reset_count = 0 + for job in running_jobs: + job.status = DownloadStatus.QUEUED + job.started_at = None + job.error_message = "Reset on worker startup" + reset_count += 1 + logger.info(f"Reset running job {job.id} to QUEUED on startup") + + if reset_count > 0: + await session.commit() + logger.info(f"Reset {reset_count} running jobs to QUEUED on worker startup") + + return {"reset_count": reset_count} + + finally: + await _cleanup_engine(engine) + + +@celery_app.task(name="tasks.reset_orphaned_jobs") +def reset_orphaned_jobs(threshold_minutes: int = None) -> dict: + """Celery task to reset orphaned running jobs. + + Jobs that have been in 'running' status for longer than the threshold + are considered orphaned (worker probably crashed) and reset to 'queued'. + """ + logger.info("Checking for orphaned running jobs...") + result = asyncio.run(_reset_orphaned_running_jobs_async(threshold_minutes)) + return result + + +def reset_all_running_jobs_sync() -> dict: + """Synchronous function to reset all running jobs. + + Called on worker startup before Celery is fully initialized. + """ + logger.info("Resetting all running jobs on worker startup...") + result = asyncio.run(_reset_all_running_jobs_async()) + return result diff --git a/frontend/src/services/api.js b/frontend/src/services/api.js index d7750fe..5874318 100644 --- a/frontend/src/services/api.js +++ b/frontend/src/services/api.js @@ -37,6 +37,7 @@ export const downloadsApi = { retry: (id) => api.post(`/downloads/${id}/retry`), stats: (params = {}) => api.get('/downloads/stats', { params }), recentActivity: (params = {}) => api.get('/downloads/recent-activity', { params }), + resetOrphaned: (thresholdMinutes = 0) => api.post('/downloads/reset-orphaned', { threshold_minutes: thresholdMinutes }), } // Credentials API diff --git a/frontend/src/views/Dashboard.vue b/frontend/src/views/Dashboard.vue index 108a671..1bea648 100644 --- a/frontend/src/views/Dashboard.vue +++ b/frontend/src/views/Dashboard.vue @@ -51,8 +51,11 @@ mdi-clock-outline
-
{{ stats?.pending || 0 }}
-
Pending Downloads
+
{{ activeDownloadsCount }}
+
Active Downloads
+
+ {{ stats?.queued || 0 }} queued, {{ stats?.running || 0 }} running +
@@ -81,6 +84,16 @@ > Retry All Failed ({{ failedCount }}) + + Reset Stuck ({{ stuckRunningCount }}) +
mdi-harddisk @@ -112,41 +125,43 @@ - + - - {{ runningDownloads.length ? 'mdi-loading' : 'mdi-download' }} + + {{ activeDownloads.length ? 'mdi-loading' : 'mdi-download' }} - Running - - {{ runningDownloads.length }} + Active + + {{ activeDownloads.length }} mdi-refresh - - + + + {{ truncate(dl.url, 30) }} - + + {{ dl.status }} + @@ -276,7 +291,7 @@ import { ref, computed, onMounted, onUnmounted } from 'vue' import { useSourcesStore } from '../stores/sources' import { useDownloadsStore } from '../stores/downloads' import { useNotificationStore } from '../stores/notifications' -import { settingsApi } from '../services/api' +import { settingsApi, downloadsApi } from '../services/api' const sourcesStore = useSourcesStore() const downloadsStore = useDownloadsStore() @@ -285,8 +300,9 @@ const notifications = useNotificationStore() // State const checkingAll = ref(false) const retryingAll = ref(false) +const resettingOrphaned = ref(false) const storageStats = ref(null) -const runningDownloads = ref([]) +const activeDownloads = ref([]) const loadingStats = ref(true) const loadingActivity = ref(true) const loadingStorage = ref(true) @@ -300,6 +316,11 @@ const sourcesWithErrors = computed(() => sourcesStore.sources.filter(s => s.error_count > 0).slice(0, 5) ) const failedCount = computed(() => stats.value?.failed || 0) +const activeDownloadsCount = computed(() => + (stats.value?.queued || 0) + (stats.value?.running || 0) +) +// Running jobs that might be stuck (shows count for UI, actual stuck detection happens server-side) +const stuckRunningCount = computed(() => stats.value?.running || 0) onMounted(() => { // Load data in background - DON'T await, let UI render immediately @@ -325,18 +346,25 @@ function loadDashboardData() { .catch(e => console.error('Failed to fetch recent activity:', e)) .finally(() => loadingActivity.value = false) - fetchRunningDownloads() + fetchActiveDownloads() // Storage stats can be slow - load independently fetchStorageStats() } -async function fetchRunningDownloads() { +async function fetchActiveDownloads() { try { - const response = await downloadsStore.fetchDownloads({ status: 'running', per_page: 20 }) - runningDownloads.value = response.items.filter(d => d.status === 'running') + // Fetch both queued and running downloads + const [queuedResponse, runningResponse] = await Promise.all([ + downloadsStore.fetchDownloads({ status: 'queued', per_page: 20 }), + downloadsStore.fetchDownloads({ status: 'running', per_page: 20 }), + ]) + // Combine and sort: running first, then queued + const running = runningResponse.items.filter(d => d.status === 'running') + const queued = queuedResponse.items.filter(d => d.status === 'queued') + activeDownloads.value = [...running, ...queued] } catch (e) { - console.error('Failed to fetch running downloads:', e) + console.error('Failed to fetch active downloads:', e) } } @@ -368,10 +396,10 @@ async function refreshStorageStats() { } function startAutoRefresh() { - // Refresh running downloads every 5 seconds when there are active downloads + // Refresh active downloads every 5 seconds when there are active downloads refreshInterval = setInterval(async () => { - if (runningDownloads.value.length > 0 || stats.value?.pending > 0) { - await fetchRunningDownloads() + if (activeDownloads.value.length > 0 || activeDownloadsCount.value > 0) { + await fetchActiveDownloads() await downloadsStore.fetchStats() await downloadsStore.fetchRecentActivity({ limit: 10 }) } @@ -385,8 +413,8 @@ function stopAutoRefresh() { } } -async function refreshRunning() { - await fetchRunningDownloads() +async function refreshActive() { + await fetchActiveDownloads() } async function checkAllSources() { @@ -403,7 +431,8 @@ async function checkAllSources() { } } notifications.success(`Queued checks for ${queued} sources`) - await fetchRunningDownloads() + await fetchActiveDownloads() + await downloadsStore.fetchStats() } catch (error) { notifications.error(`Failed to check sources: ${error.message}`) } finally { @@ -427,7 +456,7 @@ async function retryAllFailed() { } notifications.success(`Queued retry for ${retried} downloads`) await downloadsStore.fetchStats() - await fetchRunningDownloads() + await fetchActiveDownloads() } catch (error) { notifications.error(`Failed to retry downloads: ${error.message}`) } finally { @@ -435,11 +464,26 @@ async function retryAllFailed() { } } +async function resetOrphanedJobs() { + resettingOrphaned.value = true + try { + const response = await downloadsApi.resetOrphaned(0) // 0 = reset all running + notifications.success(response.data.message || `Reset ${response.data.reset_count} stuck jobs`) + await downloadsStore.fetchStats() + await fetchActiveDownloads() + } catch (error) { + notifications.error(`Failed to reset stuck jobs: ${error.message}`) + } finally { + resettingOrphaned.value = false + } +} + function getStatusIcon(status) { const icons = { completed: 'mdi-check-circle', failed: 'mdi-alert-circle', running: 'mdi-loading mdi-spin', + queued: 'mdi-clock-outline', pending: 'mdi-clock-outline', skipped: 'mdi-skip-next', } @@ -451,6 +495,7 @@ function getStatusColor(status) { completed: 'success', failed: 'error', running: 'info', + queued: 'warning', pending: 'warning', skipped: 'grey', } diff --git a/frontend/src/views/Settings.vue b/frontend/src/views/Settings.vue index a45606a..eaef7d8 100644 --- a/frontend/src/views/Settings.vue +++ b/frontend/src/views/Settings.vue @@ -18,7 +18,7 @@ type="number" min="1" max="10" - hint="Maximum number of concurrent downloads (1-10)" + hint="Maximum concurrent downloads (1-10). Requires container restart to take effect." persistent-hint /> @@ -54,7 +54,7 @@ label="Schedule Interval (seconds)" type="number" min="300" - hint="How often to check for new content (minimum 5 minutes)" + :hint="scheduleIntervalHint" persistent-hint /> @@ -419,6 +419,18 @@ const sortedAllPackages = computed(() => { return sorted }) +const scheduleIntervalHint = computed(() => { + const seconds = settings.value['download.schedule_interval'] || 3600 + if (seconds < 60) return `Check every ${seconds} seconds` + if (seconds < 3600) return `Check every ${Math.round(seconds / 60)} minutes` + if (seconds < 86400) { + const hours = seconds / 3600 + return `Check every ${hours % 1 === 0 ? hours : hours.toFixed(1)} hours` + } + const days = seconds / 86400 + return `Check every ${days % 1 === 0 ? days : days.toFixed(1)} days` +}) + onMounted(() => { // Don't block UI - load data in background loadAll() diff --git a/summary.md b/summary.md new file mode 100644 index 0000000..2181d25 --- /dev/null +++ b/summary.md @@ -0,0 +1,255 @@ +# GallerySubscriber - Project Summary + +**Updated:** 2026-01-28 (settings integration, orphan cleanup, error categorization fixes) + +## Overview + +GallerySubscriber is a Docker-based web application that automates downloading content from subscription platforms (Patreon, SubscribeStar, Discord, Hentai Foundry). It combines a Vue 3 web dashboard, Firefox extension for credential export, and a Python backend with Celery for background task scheduling. + +**Core Purpose:** Automatically download and organize content from multiple subscription platforms using gallery-dl as the underlying download engine. + +## Architecture + +``` +┌─────────────────────────────────────────────────────────────────┐ +│ GallerySubscriber Stack │ +├─────────────────────────────────────────────────────────────────┤ +│ FRONTEND (Vue 3 + Vuetify) │ +│ - Served as static assets from backend (/static) │ +│ - SPA with client-side routing │ +├─────────────────────────────────────────────────────────────────┤ +│ BACKEND API (Quart + SQLAlchemy) │ +│ - Async Python web framework │ +│ - RESTful API + WebSocket for real-time updates │ +├─────────────────────────────────────────────────────────────────┤ +│ CELERY WORKER (Background Tasks) │ +│ - Scheduled downloads via Celery Beat │ +│ - gallery-dl subprocess execution │ +├─────────────────────────────────────────────────────────────────┤ +│ DATABASE (PostgreSQL 15) + CACHE (Redis 7) │ +└─────────────────────────────────────────────────────────────────┘ +``` + +## Technology Stack + +| Layer | Technologies | +|-------|-------------| +| Backend | Quart 0.19.4, SQLAlchemy 2.0.25, asyncpg, Celery 5.3.6, Hypercorn | +| Frontend | Vue 3.4.15, Vuetify 3.5.1, Pinia 2.1.7, Vite 5.0.11 | +| Extension | Firefox WebExtension (Manifest V2) | +| Infrastructure | Docker, PostgreSQL 15, Redis 7 | +| Download Engine | gallery-dl >=1.31.0 | + +## Project Structure + +``` +GallerySubscriber/ +├── backend/app/ +│ ├── main.py # Quart app entry, serves frontend +│ ├── config.py # Pydantic Settings from env +│ ├── events.py # Redis pub/sub for WebSocket +│ ├── api/ # API blueprints +│ │ ├── subscriptions.py # Artist/creator CRUD + check trigger +│ │ ├── sources.py # Platform URL management +│ │ ├── downloads.py # Download history +│ │ ├── credentials.py # Encrypted credential storage +│ │ ├── settings.py # App configuration +│ │ ├── platforms.py # Platform schemas +│ │ └── websocket.py # Real-time events +│ ├── models/ # SQLAlchemy models +│ │ ├── subscription.py # Artist/creator groups +│ │ ├── source.py # Platform URLs per subscription +│ │ ├── download.py # Job records +│ │ ├── credential.py # Encrypted auth data +│ │ ├── content.py # Downloaded file records +│ │ └── setting.py # Key-value config +│ ├── services/ +│ │ ├── gallery_dl.py # gallery-dl wrapper +│ │ └── credential_manager.py +│ ├── tasks/ +│ │ ├── celery_app.py # Celery configuration +│ │ ├── downloads.py # Download tasks + scheduler +│ │ └── maintenance.py # Cleanup jobs +│ └── utils/ +│ ├── encryption.py # Fernet crypto +│ └── cookies.py # Netscape format handling +├── frontend/src/ +│ ├── main.js # Vue app entry +│ ├── views/ # Page components (Dashboard, Subscriptions, etc.) +│ ├── stores/ # Pinia state management +│ ├── services/api.js # Axios HTTP client +│ └── composables/ # Reusable logic (useWebSocket) +├── extension/ +│ ├── manifest.json # WebExtension config +│ ├── background/ # Service worker for cookie/token capture +│ ├── popup/ # Extension popup UI +│ └── lib/ # Platform definitions, API calls +├── alembic/ # Database migrations +├── Dockerfile # Multi-stage build (Node + Python) +├── docker-compose.yml # Production stack +└── docker-compose.dev.yml # Development with hot-reload +``` + +## Key Entry Points + +| Component | Entry Point | Command | +|-----------|-------------|---------| +| Backend API | `backend/app/main.py` | `hypercorn app.main:app --bind 0.0.0.0:8080` | +| Celery Worker | `backend/app/tasks/celery_app.py` | `celery -A app.tasks.celery_app worker -B` | +| Frontend | `frontend/src/main.js` | Built with Vite to `/static` | + +## Database Schema + +**Core Entities:** + +1. **subscriptions** - Artist/creator groups + - `name` (unique), `enabled`, `priority`, `metadata` (JSONB) + +2. **sources** - Platform URLs per subscription + - `subscription_id` (FK), `platform`, `url` + - `last_check`, `last_success`, `error_count`, `metadata` (JSONB) + - Note: All sources use the global `schedule_interval` setting + +3. **downloads** - Job records + - `source_id` (FK), `status` (queued|pending|running|completed|failed|skipped) + - `file_count`, `total_size`, `error_type`, `error_message` + +4. **credentials** - Encrypted platform authentication + - `platform` (unique), `credential_type` (cookies|token), `data` (encrypted binary) + +5. **content_items** - Downloaded file records + - `source_id`, `download_id` (FKs), `external_id`, `file_path`, `file_size` + +6. **settings** - Key-value configuration store + +**Relationships:** +``` +Subscription (1) ──→ (many) Source +Source (1) ──→ (many) Download, ContentItem +Download (1) ──→ (many) ContentItem +``` + +## API Routes + +| Endpoint | Methods | Purpose | +|----------|---------|---------| +| `/api/subscriptions` | GET, POST | List/create subscriptions | +| `/api/subscriptions/` | GET, PUT, DELETE | Single subscription CRUD | +| `/api/subscriptions//check` | POST | Trigger download check | +| `/api/subscriptions//sources` | GET, POST | Sources for subscription | +| `/api/sources/` | GET, PUT, DELETE | Single source CRUD | +| `/api/downloads` | GET | Download history with pagination | +| `/api/downloads/stats` | GET | Download counts by status/platform | +| `/api/downloads/recent-activity` | GET | Recent failures and completions | +| `/api/downloads/reset-orphaned` | POST | Reset stuck running jobs | +| `/api/credentials` | GET, POST | List/upload credentials | +| `/api/credentials/` | DELETE | Remove credential | +| `/api/settings` | GET, PUT | App configuration | +| `/api/platforms` | GET | Platform definitions | +| `/ws` | WebSocket | Real-time events | + +## Supported Platforms + +| Platform | Auth Type | Content Types | +|----------|-----------|---------------| +| Patreon | Cookies | images, attachments, posts | +| SubscribeStar | Cookies | images, attachments | +| SubscribeStar Adult | Cookies | images, attachments | +| Hentai Foundry | Cookies | images | +| Discord | Token | messages, attachments | + +## Configuration + +**Environment Variables (.env):** +- `DB_PASSWORD` (required) - PostgreSQL password +- `SECRET_KEY` (required) - Encryption key (32+ chars) +- `DOWNLOAD_PARALLEL_LIMIT` (default: 3) - Concurrent downloads +- `DOWNLOAD_RATE_LIMIT` (default: 3.0) - Seconds between requests +- `LOG_LEVEL` (default: INFO) +- `TZ` (default: America/New_York) + +## Key Patterns + +1. **Async/Await** - Quart + asyncpg for non-blocking I/O +2. **Event-Driven** - Redis pub/sub → WebSocket for real-time updates +3. **Service Layer** - GalleryDLService wraps gallery-dl subprocess +4. **Encrypted Credentials** - Fernet symmetric encryption with PBKDF2 key derivation +5. **Global Schedule Interval** - All sources checked at the same interval (configured in Settings) +6. **Celery Beat** - Runs `scheduled_check` every 60s to trigger due downloads +7. **Database-Driven Settings** - Worker reads rate_limit, retry_count, parallel_limit from DB +8. **Orphan Job Cleanup** - Worker startup + periodic task resets stuck "running" jobs + +## Docker Services + +```yaml +app: Backend API + Frontend (port 8080) +celery: Background worker with scheduler +db: PostgreSQL 15-Alpine +redis: Redis 7-Alpine + +Volumes: + downloads: /data/downloads (downloaded files) + config: /data/config (gallery-dl archive) + postgres_data: Database persistence +``` + +## Common Operations + +**Run migrations:** +```bash +docker compose exec app alembic upgrade head +``` + +**Build and start:** +```bash +docker compose up -d --build +``` + +**View logs:** +```bash +docker compose logs -f app celery +``` + +**Development mode:** +```bash +docker compose -f docker-compose.dev.yml up +``` + +## Extension Workflow + +1. User installs Firefox extension +2. User configures API URL and key in extension options +3. Extension detects supported platforms when user visits them +4. User clicks "Export Cookies" in popup +5. Extension sends encrypted cookies to backend API +6. Backend stores credentials for gallery-dl authentication + +## Download Workflow + +1. User creates subscription and adds source URLs +2. User triggers check or waits for scheduled check +3. Celery task retrieves credentials and builds gallery-dl config +4. gallery-dl downloads content to `/data/downloads` +5. Download records and content items stored in database +6. Real-time progress broadcast via WebSocket + +## Scheduled Tasks (Celery Beat) + +| Task | Schedule | Purpose | +|------|----------|---------| +| `scheduled_check` | Every 60s | Check if sources are due (based on global schedule_interval) | +| `cleanup_old_downloads` | Daily 3 AM | Remove old download records (30d completed, 7d failed) | +| `update_storage_stats` | Every 30 min | Calculate downloads folder size | +| `reset_orphaned_jobs` | Every 15 min | Reset jobs stuck in "running" > 90 min | + +**Worker Startup Behavior:** When Celery worker starts, it resets ALL "running" jobs to "queued" (handles crash recovery). + +## Settings (Database-Driven) + +| Key | Default | Purpose | +|-----|---------|---------| +| `download.schedule_interval` | 3600 | Seconds between source checks (applies to all sources) | +| `download.rate_limit` | 3.0 | Seconds between gallery-dl requests | +| `download.parallel_limit` | 3 | Concurrent Celery workers (requires restart) | +| `download.retry_count` | 3 | Max retries for failed downloads | +| `download.crawl_depth` | 0 | gallery-dl post crawl depth |