Aurora-Core/app/workers/tasks.py

"""arq task functions. Thin wrappers around the ingestion runner so retry/
concurrency/observability behavior lives in arq's worker loop, not here.

Each ingest_* task receives the `IngestionJob` row's id (created
synchronously by the enqueuing endpoint via `get_or_create_ingestion_job`,
*not* generated here) so status polling has a real, immediately-valid id to
look up from the moment the HTTP response comes back. `download_artwork_task`
is the exception — it's an internal side effect of ingestion (see
`_cascade_artwork_downloads`), not something a client triggers or polls, so
it isn't tracked in `ingestion_jobs` at all; arq's own per-function retries
are enough for it (section 10: artwork self-hosting is best-effort caching,
not a user-facing operation with a status to report).
"""

from __future__ import annotations

import uuid
from typing import Any

from app.core.database import session_scope
from app.core.logging import get_logger
from app.providers.tmdb import TMDbProvider
from app.services.artwork.download import download_artwork
from app.services.artwork.storage import get_artwork_storage
from app.services.ingestion.runner import (
    get_movie_company_ids,
    get_or_create_ingestion_job,
    get_primary_artwork_for_download,
    get_series_company_ids,
    get_series_season_ids,
    get_series_season_numbers,
    run_collection_ingestion,
    run_movie_ingestion,
    run_person_ingestion,
    run_season_ingestion,
    run_series_ingestion,
)
from app.services.ingestion.trending_ingestion import refresh_trending

logger = get_logger(__name__)


async def _cascade_artwork_downloads(
    ctx: dict[str, Any], entity_type: str, entity_ids: list[uuid.UUID]
) -> None:
    artwork_ids = await get_primary_artwork_for_download(entity_type, entity_ids)
    if not artwork_ids:
        return
    redis = ctx["redis"]
    for artwork_id in artwork_ids:
        await redis.enqueue_job("download_artwork_task", str(artwork_id))
    logger.info(
        "artwork_downloads_cascaded", entity_type=entity_type, count=len(artwork_ids)
    )


async def ingest_movie_task(ctx: dict[str, Any], job_id: str, external_id: str) -> str:
    movie_id = await run_movie_ingestion(uuid.UUID(job_id), external_id)
    await _cascade_artwork_downloads(ctx, "movie", [movie_id])
    await _cascade_artwork_downloads(ctx, "company", await get_movie_company_ids(movie_id))
    return str(movie_id)


async def ingest_person_task(ctx: dict[str, Any], job_id: str, external_id: str) -> str:
    person_id = await run_person_ingestion(uuid.UUID(job_id), external_id)
    await _cascade_artwork_downloads(ctx, "person", [person_id])
    return str(person_id)


async def ingest_series_task(
    ctx: dict[str, Any], job_id: str, external_id: str, include_seasons: bool = False
) -> str:
    series_id = await run_series_ingestion(uuid.UUID(job_id), external_id)
    await _cascade_artwork_downloads(ctx, "series", [series_id])
    # Season posters come from the season *stubs* ingest_series already
    # creates, independent of include_seasons (which only controls whether
    # full per-season episode data gets fetched) -- so this always runs.
    await _cascade_artwork_downloads(ctx, "season", await get_series_season_ids(series_id))
    await _cascade_artwork_downloads(ctx, "company", await get_series_company_ids(series_id))

    if include_seasons:
        # Fan out one ingest_season job per season this series has, using
        # the same arq connection this task is running on (arq hands every
        # job's ctx a "redis" entry precisely for chaining/fan-out like
        # this) — not a second, unrelated connection.
        redis = ctx["redis"]
        for season_number in await get_series_season_numbers(series_id):
            season_job_id = await get_or_create_ingestion_job(
                "season_sync", "season", f"{external_id}:{season_number}"
            )
            await redis.enqueue_job(
                "ingest_season_task", str(season_job_id), external_id, season_number
            )
        logger.info(
            "series_seasons_cascaded", series_id=str(series_id), external_id=external_id
        )

    return str(series_id)


async def ingest_season_task(
    ctx: dict[str, Any], job_id: str, series_external_id: str, season_number: int
) -> str:
    season_id = await run_season_ingestion(uuid.UUID(job_id), series_external_id, season_number)
    return str(season_id)


async def ingest_collection_task(ctx: dict[str, Any], job_id: str, external_id: str) -> str:
    collection_id = await run_collection_ingestion(uuid.UUID(job_id), external_id)
    await _cascade_artwork_downloads(ctx, "collection", [collection_id])
    return str(collection_id)


async def download_artwork_task(ctx: dict[str, Any], artwork_id: str) -> bool:
    async with session_scope() as session:
        return await download_artwork(session, get_artwork_storage(), uuid.UUID(artwork_id))


async def refresh_trending_task(ctx: dict[str, Any]) -> None:
    """Runs on a schedule (see WorkerSettings.cron_jobs in
    app/workers/worker.py) -- not tracked in ingestion_jobs, same as
    download_artwork_task, since there's no single external_id a client
    would poll status for."""

    provider = TMDbProvider()
    try:
        for media_type in ("movie", "tv"):
            for time_window in ("day", "week"):
                await refresh_trending(provider, media_type, time_window)
    finally:
        await provider.aclose()