706 lines
28 KiB
Python
706 lines
28 KiB
Python
"""Download-related Celery tasks."""
|
|
|
|
import asyncio
|
|
import logging
|
|
import re
|
|
from dataclasses import dataclass
|
|
from datetime import datetime, timedelta
|
|
from typing import Optional
|
|
|
|
from sqlalchemy import select, and_, or_, update
|
|
from sqlalchemy.ext.asyncio import AsyncSession
|
|
from sqlalchemy.orm import selectinload
|
|
|
|
from app.tasks.db import get_async_session, cleanup_engine as _cleanup_engine
|
|
from app.tasks.celery_app import celery_app
|
|
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.services.patreon_resolver import resolve_campaign_id
|
|
from app.events import publish_event
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
# Settings for database connection
|
|
from app.config import get_settings
|
|
settings = get_settings()
|
|
|
|
|
|
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:
|
|
logger.debug(f"DB setting '{key}' found: {setting.value} (type: {type(setting.value).__name__})")
|
|
return setting.value
|
|
|
|
# Fall back to DEFAULT_SETTINGS, then to provided default
|
|
fallback = DEFAULT_SETTINGS.get(key, default)
|
|
logger.debug(f"DB setting '{key}' not found, using fallback: {fallback}")
|
|
return fallback
|
|
|
|
|
|
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())
|
|
|
|
|
|
# --- Patreon campaign-ID auto-resolution helpers -------------------------
|
|
# These support auto-healing gallery-dl's "Failed to extract campaign ID"
|
|
# failures on Patreon sources. See docs/superpowers/specs/2026-04-18-patreon-campaign-resolver-design.md.
|
|
|
|
_PATREON_VANITY_RE = re.compile(
|
|
r"^https?://(?:www\.)?patreon\.com/(?:c/)?(?!id:)([^/?#]+)",
|
|
re.IGNORECASE,
|
|
)
|
|
_CAMPAIGN_ID_FAILURE_PATTERN = "failed to extract campaign id"
|
|
|
|
|
|
def _effective_patreon_url(platform: str, source_url: str, metadata: Optional[dict]) -> str:
|
|
"""Return the URL to pass to gallery-dl for this source.
|
|
|
|
For Patreon sources with a cached campaign_id in metadata, returns
|
|
patreon.com/id:<id> to bypass the broken vanity-URL lookup path.
|
|
All other cases return source_url unchanged.
|
|
"""
|
|
if platform == "patreon" and metadata:
|
|
campaign_id = metadata.get("patreon_campaign_id")
|
|
if campaign_id:
|
|
return f"https://www.patreon.com/id:{campaign_id}"
|
|
return source_url
|
|
|
|
|
|
def _looks_like_campaign_id_failure(dl_result) -> bool:
|
|
"""True if the gallery-dl output contains the campaign-ID extraction failure marker."""
|
|
stdout = getattr(dl_result, "stdout", None) or ""
|
|
stderr = getattr(dl_result, "stderr", None) or ""
|
|
return _CAMPAIGN_ID_FAILURE_PATTERN in f"{stdout}\n{stderr}".lower()
|
|
|
|
|
|
def _extract_vanity_from_url(url: str) -> Optional[str]:
|
|
"""Return the vanity (creator slug) from a Patreon URL, or None if not extractable.
|
|
|
|
Handles patreon.com/<vanity>, patreon.com/c/<vanity>[/...], with or without www.
|
|
Returns None for id:<N> URLs, bare domain, and non-Patreon URLs.
|
|
"""
|
|
m = _PATREON_VANITY_RE.match(url)
|
|
return m.group(1) if m else None
|
|
|
|
|
|
@dataclass
|
|
class _DownloadAttempt:
|
|
"""Result of executing a download, possibly with a retry after campaign-ID resolution."""
|
|
dl_result: DownloadResult
|
|
resolved_campaign_id: Optional[str]
|
|
|
|
|
|
async def _execute_download_with_retry(
|
|
*,
|
|
gdl_service,
|
|
platform: str,
|
|
source_url: str,
|
|
source_metadata: dict,
|
|
subscription_name: str,
|
|
source_config,
|
|
cookies_path: Optional[str],
|
|
auth_token: Optional[str],
|
|
) -> _DownloadAttempt:
|
|
"""Run the download once, auto-heal Patreon campaign-ID failures with one retry.
|
|
|
|
Returns a _DownloadAttempt wrapping:
|
|
- the final DownloadResult (retry's result if we retried, else the first)
|
|
- resolved_campaign_id: the newly-resolved ID (if we resolved one this call), else None
|
|
|
|
The caller is responsible for persisting resolved_campaign_id into source.metadata_.
|
|
"""
|
|
effective_url = _effective_patreon_url(platform, source_url, source_metadata)
|
|
dl_result = await gdl_service.download(
|
|
url=effective_url,
|
|
subscription_name=subscription_name,
|
|
platform=platform,
|
|
source_config=source_config,
|
|
cookies_path=cookies_path,
|
|
auth_token=auth_token,
|
|
)
|
|
|
|
# Only retry if: patreon + campaign-ID failure + no cached id + extractable vanity
|
|
if (
|
|
platform == "patreon"
|
|
and not dl_result.success
|
|
and "patreon_campaign_id" not in (source_metadata or {})
|
|
and _looks_like_campaign_id_failure(dl_result)
|
|
):
|
|
vanity = _extract_vanity_from_url(source_url)
|
|
if vanity:
|
|
logger.info(
|
|
f"Attempting campaign-ID resolution for {subscription_name}/patreon ({vanity})"
|
|
)
|
|
campaign_id = await resolve_campaign_id(vanity, cookies_path)
|
|
if campaign_id:
|
|
retry_result = await gdl_service.download(
|
|
url=f"https://www.patreon.com/id:{campaign_id}",
|
|
subscription_name=subscription_name,
|
|
platform=platform,
|
|
source_config=source_config,
|
|
cookies_path=cookies_path,
|
|
auth_token=auth_token,
|
|
)
|
|
return _DownloadAttempt(
|
|
dl_result=retry_result, resolved_campaign_id=campaign_id
|
|
)
|
|
|
|
return _DownloadAttempt(dl_result=dl_result, resolved_campaign_id=None)
|
|
|
|
|
|
async def _download_source_async(source_id: int, download_id: int = None) -> dict:
|
|
"""Async implementation of source download.
|
|
|
|
Uses separate DB sessions for setup and result writing to avoid
|
|
connection timeouts during long-running gallery-dl downloads.
|
|
|
|
Args:
|
|
source_id: ID of the source to download
|
|
download_id: Optional ID of a pre-created Download record (QUEUED state)
|
|
|
|
Returns:
|
|
Dict with download results
|
|
"""
|
|
# --- Phase 1: Setup (read source info, mark as RUNNING, get credentials) ---
|
|
session_factory, engine = get_async_session()
|
|
|
|
try:
|
|
async with session_factory() as session:
|
|
# Get source with subscription
|
|
result = await session.execute(
|
|
select(Source)
|
|
.options(selectinload(Source.subscription))
|
|
.where(Source.id == source_id)
|
|
)
|
|
source = result.scalar()
|
|
|
|
if not source:
|
|
logger.error(f"Source {source_id} not found")
|
|
return {"source_id": source_id, "status": "error", "error": "Source not found"}
|
|
|
|
if not source.enabled:
|
|
logger.info(f"Source {source.subscription.name}/{source.platform} is disabled, skipping")
|
|
if download_id:
|
|
dl_result = await session.execute(
|
|
select(Download).where(Download.id == download_id)
|
|
)
|
|
dl = dl_result.scalar()
|
|
if dl:
|
|
dl.status = DownloadStatus.SKIPPED
|
|
await session.commit()
|
|
return {"source_id": source_id, "status": "skipped", "reason": "disabled"}
|
|
|
|
# Capture values we need after the session closes
|
|
subscription_name = source.subscription.name
|
|
source_url = source.url
|
|
source_platform = source.platform
|
|
source_metadata = dict(source.metadata_ or {})
|
|
logger.info(f"Starting download for {subscription_name}/{source_platform}")
|
|
|
|
# Use existing download record or create a new one
|
|
if download_id:
|
|
result = await session.execute(
|
|
select(Download).where(Download.id == download_id)
|
|
)
|
|
download = result.scalar()
|
|
if download:
|
|
# Guard against race conditions: only proceed if still QUEUED
|
|
if download.status != DownloadStatus.QUEUED:
|
|
logger.info(f"Download {download_id} is no longer QUEUED (status: {download.status}), skipping")
|
|
return {"source_id": source_id, "download_id": download_id, "status": "skipped", "reason": "already_processed"}
|
|
|
|
# Also check if source already has another RUNNING download
|
|
running_check = await session.execute(
|
|
select(Download.id).where(
|
|
Download.source_id == source_id,
|
|
Download.status == DownloadStatus.RUNNING,
|
|
Download.id != download_id
|
|
)
|
|
)
|
|
running_download_id = running_check.scalar()
|
|
if running_download_id:
|
|
# Mark this download as skipped since another is already running
|
|
download.status = DownloadStatus.SKIPPED
|
|
download.error_message = f"Skipped - another download ({running_download_id}) already running for this source"
|
|
await session.commit()
|
|
logger.info(f"Source {source_id} already has a running download, marked {download_id} as SKIPPED")
|
|
return {"source_id": source_id, "download_id": download_id, "status": "skipped", "reason": "source_already_running"}
|
|
|
|
download.status = DownloadStatus.RUNNING
|
|
download.started_at = utcnow()
|
|
else:
|
|
logger.warning(f"Download {download_id} not found, creating new record")
|
|
download = None
|
|
|
|
if not download_id or download is None:
|
|
# Check if source already has an active download before creating new one
|
|
active_check = await session.execute(
|
|
select(Download.id).where(
|
|
Download.source_id == source_id,
|
|
Download.status.in_([DownloadStatus.QUEUED, DownloadStatus.RUNNING])
|
|
)
|
|
)
|
|
if active_check.scalar():
|
|
logger.info(f"Source {source_id} already has an active download, skipping new download creation")
|
|
return {"source_id": source_id, "status": "skipped", "reason": "source_already_active"}
|
|
|
|
download = Download(
|
|
source_id=source_id,
|
|
url=source_url,
|
|
status=DownloadStatus.RUNNING,
|
|
started_at=utcnow(),
|
|
)
|
|
session.add(download)
|
|
|
|
await session.commit()
|
|
await session.refresh(download)
|
|
actual_download_id = download.id
|
|
|
|
# Broadcast download started event
|
|
publish_event("download.started", {
|
|
"download_id": actual_download_id,
|
|
"source_id": source_id,
|
|
"subscription_name": subscription_name,
|
|
"platform": source_platform,
|
|
"url": source_url,
|
|
})
|
|
|
|
# Get credentials before closing session
|
|
cred_manager = CredentialManager(session)
|
|
cookies_path = None
|
|
auth_token = None
|
|
|
|
if source_platform in ("discord", "pixiv"):
|
|
auth_token = await cred_manager.get_token(source_platform)
|
|
if not auth_token:
|
|
logger.warning(f"No token for {source_platform}, download may fail")
|
|
else:
|
|
cookies_path = await cred_manager.get_cookies_path(source_platform)
|
|
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)
|
|
|
|
# --- Phase 2: Execute download (no DB connection held) ---
|
|
source_config = SourceConfig.from_dict(source_metadata)
|
|
dl_result = None
|
|
dl_error = None
|
|
resolved_campaign_id: Optional[str] = None
|
|
|
|
try:
|
|
gdl_service = GalleryDLService(rate_limit=db_rate_limit)
|
|
attempt = await _execute_download_with_retry(
|
|
gdl_service=gdl_service,
|
|
platform=source_platform,
|
|
source_url=source_url,
|
|
source_metadata=source_metadata,
|
|
subscription_name=subscription_name,
|
|
source_config=source_config,
|
|
cookies_path=cookies_path,
|
|
auth_token=auth_token,
|
|
)
|
|
dl_result = attempt.dl_result
|
|
resolved_campaign_id = attempt.resolved_campaign_id
|
|
except Exception as e:
|
|
logger.exception(f"Unexpected error downloading {subscription_name}/{source_platform}: {e}")
|
|
dl_error = e
|
|
|
|
# --- Phase 3: Write results (fresh DB session) ---
|
|
session_factory2, engine2 = get_async_session()
|
|
|
|
try:
|
|
async with session_factory2() as session:
|
|
# Re-fetch download and source with fresh session
|
|
download = (await session.execute(
|
|
select(Download).where(Download.id == actual_download_id)
|
|
)).scalar()
|
|
source = (await session.execute(
|
|
select(Source).where(Source.id == source_id)
|
|
)).scalar()
|
|
|
|
if not download or not source:
|
|
logger.error(f"Could not re-fetch download {actual_download_id} or source {source_id} for result update")
|
|
return {"source_id": source_id, "status": "error", "error": "Lost DB records"}
|
|
|
|
# Persist resolved campaign ID (if the retry hook resolved one this run)
|
|
if resolved_campaign_id:
|
|
source.metadata_ = {
|
|
**(source.metadata_ or {}),
|
|
"patreon_campaign_id": resolved_campaign_id,
|
|
}
|
|
|
|
if dl_error:
|
|
download.status = DownloadStatus.FAILED
|
|
download.error_type = DBErrorType.UNKNOWN_ERROR
|
|
download.error_message = str(dl_error)
|
|
download.completed_at = utcnow()
|
|
source.error_count = (source.error_count or 0) + 1
|
|
source.last_check = utcnow()
|
|
await session.commit()
|
|
|
|
publish_event("download.failed", {
|
|
"download_id": actual_download_id,
|
|
"source_id": source_id,
|
|
"subscription_name": subscription_name,
|
|
"platform": source_platform,
|
|
"error_type": "unknown_error",
|
|
"error_message": str(dl_error),
|
|
})
|
|
|
|
return {
|
|
"source_id": source_id,
|
|
"download_id": actual_download_id,
|
|
"status": "error",
|
|
"error": str(dl_error),
|
|
}
|
|
|
|
# Update download record with results
|
|
download.completed_at = utcnow()
|
|
download.file_count = dl_result.files_downloaded
|
|
# Store full logs - cleanup task will age out old records
|
|
download.metadata_ = {
|
|
"stdout": dl_result.stdout if dl_result.stdout else None,
|
|
"stderr": dl_result.stderr if dl_result.stderr else None,
|
|
"duration_seconds": dl_result.duration_seconds,
|
|
}
|
|
|
|
if dl_result.success:
|
|
download.status = DownloadStatus.COMPLETED
|
|
source.last_success = utcnow()
|
|
source.error_count = 0
|
|
logger.info(f"Download completed for {subscription_name}/{source_platform}: {dl_result.files_downloaded} files")
|
|
|
|
# Mark any prior failures for this source as superseded
|
|
await session.execute(
|
|
update(Download)
|
|
.where(
|
|
Download.source_id == source_id,
|
|
Download.status == DownloadStatus.FAILED,
|
|
)
|
|
.values(superseded=True)
|
|
)
|
|
|
|
publish_event("download.completed", {
|
|
"download_id": actual_download_id,
|
|
"source_id": source_id,
|
|
"subscription_name": subscription_name,
|
|
"platform": source_platform,
|
|
"file_count": dl_result.files_downloaded,
|
|
"duration_seconds": dl_result.duration_seconds,
|
|
})
|
|
else:
|
|
download.status = DownloadStatus.FAILED
|
|
download.error_type = dl_result.error_type.value if dl_result.error_type else None
|
|
download.error_message = dl_result.error_message
|
|
source.error_count = (source.error_count or 0) + 1
|
|
logger.warning(f"Download failed for {subscription_name}/{source_platform}: {dl_result.error_message}")
|
|
|
|
publish_event("download.failed", {
|
|
"download_id": actual_download_id,
|
|
"source_id": source_id,
|
|
"subscription_name": subscription_name,
|
|
"platform": source_platform,
|
|
"error_type": dl_result.error_type.value if dl_result.error_type else None,
|
|
"error_message": dl_result.error_message,
|
|
})
|
|
|
|
source.last_check = utcnow()
|
|
await session.commit()
|
|
|
|
return {
|
|
"source_id": source_id,
|
|
"download_id": actual_download_id,
|
|
"status": "completed" if dl_result.success else "failed",
|
|
"files_downloaded": dl_result.files_downloaded,
|
|
"error_type": dl_result.error_type.value if dl_result.error_type else None,
|
|
"error_message": dl_result.error_message,
|
|
"duration_seconds": dl_result.duration_seconds,
|
|
}
|
|
finally:
|
|
await _cleanup_engine(engine2)
|
|
|
|
|
|
async def _scheduled_check_async() -> dict:
|
|
"""Async implementation of scheduled source check.
|
|
|
|
Returns:
|
|
Dict with check results
|
|
"""
|
|
session_factory, engine = get_async_session()
|
|
|
|
try:
|
|
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
|
|
raw_interval = await _get_db_setting(
|
|
session, "download.schedule_interval", 28800 # Default 8 hours
|
|
)
|
|
# Ensure we have an integer value (JSONB might return different types)
|
|
global_interval = int(raw_interval) if raw_interval is not None else 28800
|
|
logger.info(f"Schedule check: using interval {global_interval}s (raw value: {raw_interval}, type: {type(raw_interval).__name__})")
|
|
|
|
# Find sources due for checking:
|
|
# - enabled = True
|
|
# - last_check is NULL OR (now - last_check) > global_interval
|
|
# Order by subscription priority, then by when last checked
|
|
query = (
|
|
select(Source)
|
|
.options(selectinload(Source.subscription))
|
|
.join(Subscription)
|
|
.where(
|
|
and_(
|
|
Source.enabled == True,
|
|
Subscription.enabled == True, # Subscription must also be enabled
|
|
or_(
|
|
Source.last_check == None,
|
|
Source.last_check < now - timedelta(seconds=1) # Will be refined below
|
|
)
|
|
)
|
|
)
|
|
.order_by(
|
|
Subscription.priority.desc(), # Higher subscription priority first
|
|
Source.last_check.asc().nullsfirst(), # Never checked first
|
|
)
|
|
)
|
|
|
|
result = await session.execute(query)
|
|
sources = result.scalars().all()
|
|
|
|
# Filter by global schedule interval from Settings
|
|
# All sources use the same interval - no per-source overrides
|
|
due_sources = []
|
|
not_due_count = 0
|
|
for source in sources:
|
|
if source.last_check is None:
|
|
due_sources.append(source)
|
|
logger.debug(f"Source {source.id} ({source.platform}) - never checked, marking due")
|
|
else:
|
|
time_since_check = (now - source.last_check).total_seconds()
|
|
if time_since_check >= global_interval:
|
|
due_sources.append(source)
|
|
logger.debug(f"Source {source.id} ({source.platform}) - last check {time_since_check:.0f}s ago >= {global_interval}s, marking due")
|
|
else:
|
|
not_due_count += 1
|
|
logger.debug(f"Source {source.id} ({source.platform}) - last check {time_since_check:.0f}s ago < {global_interval}s, not due yet")
|
|
|
|
logger.info(f"Schedule check: {len(due_sources)} sources due, {not_due_count} sources not due (interval: {global_interval}s)")
|
|
|
|
# Get source IDs that already have a queued or running download
|
|
due_source_ids = [s.id for s in due_sources]
|
|
active_result = await session.execute(
|
|
select(Download.source_id).where(
|
|
Download.source_id.in_(due_source_ids),
|
|
Download.status.in_([DownloadStatus.QUEUED, DownloadStatus.RUNNING])
|
|
)
|
|
)
|
|
active_source_ids = set(active_result.scalars().all())
|
|
|
|
# Create download records and queue tasks for each eligible source
|
|
queued = 0
|
|
skipped = 0
|
|
for source in due_sources:
|
|
if source.id in active_source_ids:
|
|
skipped += 1
|
|
logger.debug(f"Skipping {source.subscription.name}/{source.platform} - already active")
|
|
continue
|
|
|
|
download = Download(
|
|
source_id=source.id,
|
|
url=source.url,
|
|
status=DownloadStatus.QUEUED,
|
|
)
|
|
session.add(download)
|
|
await session.commit()
|
|
await session.refresh(download)
|
|
|
|
download_source.delay(source.id, download_id=download.id)
|
|
queued += 1
|
|
logger.debug(f"Queued download for {source.subscription.name}/{source.platform}")
|
|
|
|
if skipped:
|
|
logger.info(f"Skipped {skipped} sources with active downloads")
|
|
|
|
return {
|
|
"checked": len(sources),
|
|
"queued": queued,
|
|
"timestamp": now.isoformat(),
|
|
}
|
|
finally:
|
|
await _cleanup_engine(engine)
|
|
|
|
|
|
async def _cleanup_old_downloads_async() -> dict:
|
|
"""Async implementation of download cleanup.
|
|
|
|
Returns:
|
|
Dict with cleanup results
|
|
"""
|
|
session_factory, engine = get_async_session()
|
|
|
|
try:
|
|
async with session_factory() as session:
|
|
now = utcnow()
|
|
|
|
# Delete completed downloads older than 30 days
|
|
completed_cutoff = now - timedelta(days=30)
|
|
completed_query = select(Download).where(
|
|
and_(
|
|
Download.status == DownloadStatus.COMPLETED,
|
|
Download.created_at < completed_cutoff
|
|
)
|
|
)
|
|
completed_result = await session.execute(completed_query)
|
|
completed_old = completed_result.scalars().all()
|
|
|
|
# Delete failed downloads older than 7 days
|
|
failed_cutoff = now - timedelta(days=7)
|
|
failed_query = select(Download).where(
|
|
and_(
|
|
Download.status == DownloadStatus.FAILED,
|
|
Download.created_at < failed_cutoff
|
|
)
|
|
)
|
|
failed_result = await session.execute(failed_query)
|
|
failed_old = failed_result.scalars().all()
|
|
|
|
deleted_count = 0
|
|
for download in completed_old + failed_old:
|
|
await session.delete(download)
|
|
deleted_count += 1
|
|
|
|
await session.commit()
|
|
|
|
logger.info(f"Cleaned up {deleted_count} old download records")
|
|
return {"deleted": deleted_count}
|
|
finally:
|
|
await _cleanup_engine(engine)
|
|
|
|
|
|
# Celery tasks that wrap async functions
|
|
|
|
@celery_app.task(bind=True)
|
|
def download_source(self, source_id: int, download_id: int = None):
|
|
"""Download new content from a single source.
|
|
|
|
Args:
|
|
source_id: ID of the source to download from
|
|
download_id: Optional ID of a pre-created Download record (QUEUED state)
|
|
"""
|
|
try:
|
|
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}")
|
|
# 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
|
|
def scheduled_check():
|
|
"""Check all enabled sources that are due for a check.
|
|
|
|
This is called periodically by Celery Beat.
|
|
"""
|
|
logger.info("=== SCHEDULED_CHECK TASK STARTING ===")
|
|
try:
|
|
result = asyncio.run(_scheduled_check_async())
|
|
logger.info(f"=== SCHEDULED_CHECK TASK COMPLETE: {result} ===")
|
|
return result
|
|
except Exception as e:
|
|
logger.exception(f"=== SCHEDULED_CHECK TASK FAILED: {e} ===")
|
|
raise
|
|
|
|
|
|
@celery_app.task
|
|
def cleanup_old_downloads():
|
|
"""Clean up old download records.
|
|
|
|
Keeps the database from growing indefinitely.
|
|
"""
|
|
return asyncio.run(_cleanup_old_downloads_async())
|
|
|
|
|
|
@celery_app.task(bind=True)
|
|
def process_download(self, download_id: int):
|
|
"""Process a single download record (for retries).
|
|
|
|
Args:
|
|
download_id: ID of the download to process
|
|
"""
|
|
async def _process():
|
|
session_factory, engine = get_async_session()
|
|
|
|
try:
|
|
async with session_factory() as session:
|
|
result = await session.execute(
|
|
select(Download).where(Download.id == download_id)
|
|
)
|
|
download = result.scalar()
|
|
|
|
if not download:
|
|
return {"download_id": download_id, "status": "error", "error": "Download not found"}
|
|
|
|
if not download.source_id:
|
|
return {"download_id": download_id, "status": "error", "error": "No source associated"}
|
|
|
|
# Delegate to download_source
|
|
return await _download_source_async(download.source_id, download_id=download_id)
|
|
finally:
|
|
await _cleanup_engine(engine)
|
|
|
|
try:
|
|
return asyncio.run(_process())
|
|
except Exception as e:
|
|
logger.exception(f"Task failed for download {download_id}: {e}")
|
|
# 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
|