From 4403305da6ee2360d8979361c818a490b213eddd Mon Sep 17 00:00:00 2001 From: NixReaper Date: Sun, 6 Sep 2026 05:18:02 -0500 Subject: [PATCH] Add trending movies/series, refreshed on a schedule Mirrors TMDb's own /trending/{movie|tv}/{day|week} ranking rather than Aurora computing trending itself from popularity deltas -- that would require re-syncing the entire catalog on a schedule, infeasible at any real scale, whereas TMDb already computes this well from signals (real view/search velocity) Aurora has no access to. New: TrendingEntry model + migration 0003, MetadataProvider.get_trending (concrete/optional, like aclose(), so existing test fakes don't all need updating), refresh_trending() which actually ingests each trending item (only ~20 per window, cheap enough to fully ingest rather than apply the never-fabricate-a-stub rule recommendations uses), a twice-daily arq cron job (WorkerSettings.cron_jobs, new to this codebase), a manual admin trigger for right after a deploy, and GET /movies/trending + GET /series/trending. Co-Authored-By: Claude Sonnet 5 --- app/api/v1/admin.py | 23 +++ app/api/v1/movies.py | 15 ++ app/api/v1/series.py | 15 ++ app/models/__init__.py | 2 + app/models/trending.py | 35 ++++ app/providers/base.py | 11 ++ app/providers/tmdb.py | 7 + app/repositories/trending_repository.py | 40 +++++ app/services/ingestion/trending_ingestion.py | 53 ++++++ app/services/movie_service.py | 26 ++- app/services/tv_service.py | 26 ++- app/workers/tasks.py | 17 ++ app/workers/worker.py | 6 + docs/api.md | 19 +++ migrations/versions/0003_trending.py | 55 +++++++ tests/integration/test_trending.py | 165 +++++++++++++++++++ 16 files changed, 513 insertions(+), 2 deletions(-) create mode 100644 app/models/trending.py create mode 100644 app/repositories/trending_repository.py create mode 100644 app/services/ingestion/trending_ingestion.py create mode 100644 migrations/versions/0003_trending.py create mode 100644 tests/integration/test_trending.py diff --git a/app/api/v1/admin.py b/app/api/v1/admin.py index b13aa9f..9afac0e 100644 --- a/app/api/v1/admin.py +++ b/app/api/v1/admin.py @@ -168,6 +168,29 @@ async def enqueue_collection_ingestion( ) +class TrendingRefreshEnqueuedOut(BaseModel): + arq_job_id: str + status: str = "enqueued" + + +@router.post("/refresh-trending", response_model=TrendingRefreshEnqueuedOut) +async def enqueue_trending_refresh( + queue: ArqRedis = Depends(get_queue), +) -> TrendingRefreshEnqueuedOut: + """Manually kick off the same refresh the twice-daily cron job runs + (see WorkerSettings.cron_jobs) -- useful right after a fresh deploy so + GET /movies/trending isn't empty until the next scheduled tick. Not + tracked via ingestion_jobs (see refresh_trending_task's docstring), so + there's no job_id to poll -- check GET /movies/trending directly.""" + + arq_job = await queue.enqueue_job("refresh_trending_task") + if arq_job is None: + raise HTTPException( + status.HTTP_503_SERVICE_UNAVAILABLE, "Could not enqueue trending refresh (queue busy)." + ) + return TrendingRefreshEnqueuedOut(arq_job_id=arq_job.job_id) + + @router.get("/ingest/jobs/{job_id}", response_model=IngestionJobOut) async def get_ingestion_job_status( job_id: uuid.UUID, session: AsyncSession = Depends(get_db) diff --git a/app/api/v1/movies.py b/app/api/v1/movies.py index aafbe97..c906fe4 100644 --- a/app/api/v1/movies.py +++ b/app/api/v1/movies.py @@ -75,6 +75,21 @@ async def upcoming_movies( return await movie_service.list_upcoming(session, region=region, page=page, page_size=page_size) +@router.get("/trending", response_model=MovieListOut) +async def trending_movies( + time_window: Literal["day", "week"] = Query("day", alias="window"), + page: int = Query(1, ge=1), + page_size: int = Query(20, ge=1, le=50), + session: AsyncSession = Depends(get_db), +) -> MovieListOut: + try: + return await movie_service.list_trending( + session, time_window=time_window, page=page, page_size=page_size + ) + except movie_service.InvalidMovieFilterError as exc: + raise HTTPException(status.HTTP_422_UNPROCESSABLE_CONTENT, str(exc)) from exc + + @router.get("/{aurora_id}", response_model=MovieDetailOut) async def get_movie( aurora_id: str, diff --git a/app/api/v1/series.py b/app/api/v1/series.py index 6b3e935..3ff0586 100644 --- a/app/api/v1/series.py +++ b/app/api/v1/series.py @@ -55,6 +55,21 @@ async def list_series( raise HTTPException(status.HTTP_422_UNPROCESSABLE_CONTENT, str(exc)) from exc +@router.get("/trending", response_model=SeriesListOut) +async def trending_series( + time_window: Literal["day", "week"] = Query("day", alias="window"), + page: int = Query(1, ge=1), + page_size: int = Query(20, ge=1, le=50), + session: AsyncSession = Depends(get_db), +) -> SeriesListOut: + try: + return await tv_service.list_trending( + session, time_window=time_window, page=page, page_size=page_size + ) + except tv_service.InvalidSeriesFilterError as exc: + raise HTTPException(status.HTTP_422_UNPROCESSABLE_CONTENT, str(exc)) from exc + + @router.get("/{aurora_id}", response_model=SeriesDetailOut) async def get_series( aurora_id: str, diff --git a/app/models/__init__.py b/app/models/__init__.py index db2a19e..99fc6bb 100644 --- a/app/models/__init__.py +++ b/app/models/__init__.py @@ -24,6 +24,7 @@ from app.models.movie import ( ) from app.models.person import Person from app.models.provider import IngestionJob, Provider, ProviderRecord +from app.models.trending import TrendingEntry from app.models.tv import ( Episode, Season, @@ -75,4 +76,5 @@ __all__ = [ "Video", "WatchProvider", "WatchProviderOffering", + "TrendingEntry", ] diff --git a/app/models/trending.py b/app/models/trending.py new file mode 100644 index 0000000..21bc296 --- /dev/null +++ b/app/models/trending.py @@ -0,0 +1,35 @@ +"""Aurora's mirror of TMDb's own trending lists (section 17, additive). + +TMDb already computes trending using signals Aurora has no access to +(real view/search velocity across their whole audience) — reinventing +that from Aurora's own popularity snapshots would require re-syncing the +entire catalog on a schedule, which isn't feasible at any real scale. +Instead Aurora asks TMDb what's trending right now and mirrors that +ranked list (see app/services/ingestion/trending_ingestion.py).""" + +from __future__ import annotations + +import uuid + +from sqlalchemy import Index, Integer, String +from sqlalchemy.orm import Mapped, mapped_column + +from app.models.base import Base, TimestampMixin, UUIDPrimaryKeyMixin + + +class TrendingEntry(Base, UUIDPrimaryKeyMixin, TimestampMixin): + """One row per (entity, time_window) currently on Aurora's trending + list -- replaced wholesale on each refresh, same as ReleaseDate/ + WatchProviderOffering. `updated_at` doubles as "when this window was + last refreshed".""" + + __tablename__ = "trending_entries" + __table_args__ = (Index("ix_trending_entries_window", "entity_type", "time_window"),) + + entity_type: Mapped[str] = mapped_column(String(30), nullable=False) # "movie" | "series" + entity_id: Mapped[uuid.UUID] = mapped_column(nullable=False, index=True) + time_window: Mapped[str] = mapped_column(String(10), nullable=False) # "day" | "week" + rank: Mapped[int] = mapped_column(Integer, nullable=False) + + +__all__ = ["TrendingEntry"] diff --git a/app/providers/base.py b/app/providers/base.py index e226355..a3dc87b 100644 --- a/app/providers/base.py +++ b/app/providers/base.py @@ -70,5 +70,16 @@ class MetadataProvider(ABC): """Search this provider's catalog for movies matching `query`. Used to seed ingestion, not exposed directly to API clients.""" + async def get_trending(self, media_type: str, time_window: str) -> list[dict[str, Any]]: + """Fetch a ranked list of raw items (movies or series, per + `media_type`) currently trending over `time_window` ("day" | + "week"). Not every provider has a trending concept, so this is + concrete rather than abstract (like `aclose` below) — a provider + that doesn't support it simply doesn't override this, and callers + get an empty list rather than every other provider implementation + having to stub out a method that means nothing for them.""" + + return [] + async def aclose(self) -> None: """Release any held resources (HTTP clients, ...). Optional.""" diff --git a/app/providers/tmdb.py b/app/providers/tmdb.py index ca2f515..8436568 100644 --- a/app/providers/tmdb.py +++ b/app/providers/tmdb.py @@ -124,6 +124,13 @@ class TMDbProvider(MetadataProvider): data = await self._get("/search/movie", {"query": query}) return data.get("results", []) + async def get_trending(self, media_type: str, time_window: str) -> list[dict[str, Any]]: + # TMDb's own endpoint shape: /trending/{movie|tv}/{day|week} -- page + # 1 only (~20 results), enough for a trending rail without pulling + # in TMDb's full multi-page trending list. + data = await self._get(f"/trending/{media_type}/{time_window}") + return data.get("results", []) + async def aclose(self) -> None: if self._owns_client: await self._client.aclose() diff --git a/app/repositories/trending_repository.py b/app/repositories/trending_repository.py new file mode 100644 index 0000000..81fc4f9 --- /dev/null +++ b/app/repositories/trending_repository.py @@ -0,0 +1,40 @@ +"""Data access for TrendingEntry.""" + +from __future__ import annotations + +import uuid + +from sqlalchemy import delete, select +from sqlalchemy.ext.asyncio import AsyncSession + +from app.models.trending import TrendingEntry + + +async def get_trending_entity_ids( + session: AsyncSession, *, entity_type: str, time_window: str +) -> list[uuid.UUID]: + stmt = ( + select(TrendingEntry.entity_id) + .where(TrendingEntry.entity_type == entity_type, TrendingEntry.time_window == time_window) + .order_by(TrendingEntry.rank.asc()) + ) + result = await session.execute(stmt) + return list(result.scalars().all()) + + +async def replace_trending( + session: AsyncSession, *, entity_type: str, time_window: str, entity_ids: list[uuid.UUID] +) -> None: + """Replace a window's trending snapshot wholesale, same delete-then- + recreate approach as ReleaseDate/WatchProviderOffering -- there is no + history to preserve, only the current ranking.""" + + await session.execute( + delete(TrendingEntry).where( + TrendingEntry.entity_type == entity_type, TrendingEntry.time_window == time_window + ) + ) + for rank, entity_id in enumerate(entity_ids): + session.add( + TrendingEntry(entity_type=entity_type, entity_id=entity_id, time_window=time_window, rank=rank) + ) diff --git a/app/services/ingestion/trending_ingestion.py b/app/services/ingestion/trending_ingestion.py new file mode 100644 index 0000000..bfbb69b --- /dev/null +++ b/app/services/ingestion/trending_ingestion.py @@ -0,0 +1,53 @@ +"""Refreshes Aurora's mirror of TMDb's own trending lists (section 17, +additive) -- see app/models/trending.py for why this mirrors TMDb's +ranking rather than Aurora computing its own from popularity deltas. + +Unlike recommendations' never-fabricate-a-stub rule, this actually +ingests/refreshes each trending item's own metadata as a side effect -- +there are only ~20 items per (media_type, time_window) window (TMDb's +first page), cheap enough to fully ingest rather than silently drop +anything Aurora hasn't seen yet. +""" + +from __future__ import annotations + +from app.core.database import session_scope +from app.core.logging import get_logger +from app.providers.base import MetadataProvider +from app.repositories.trending_repository import replace_trending +from app.services.ingestion.movie_ingestion import ingest_movie +from app.services.ingestion.series_ingestion import ingest_series + +logger = get_logger(__name__) + +# TMDb's own media_type values -> Aurora's entity_type values. +_MEDIA_TYPE_TO_ENTITY_TYPE = {"movie": "movie", "tv": "series"} + + +async def refresh_trending(provider: MetadataProvider, media_type: str, time_window: str) -> int: + """Fetch TMDb's current trending list for (media_type, time_window), + ingest/refresh each item, and replace Aurora's own ranked snapshot for + that window. Returns how many entries were written.""" + + entity_type = _MEDIA_TYPE_TO_ENTITY_TYPE[media_type] + raw_results = await provider.get_trending(media_type, time_window) + + entity_ids = [] + for item in raw_results: + external_id = str(item["id"]) + async with session_scope() as session: + if media_type == "movie": + entity_id = await ingest_movie(session, provider, external_id) + else: + entity_id = await ingest_series(session, provider, external_id) + entity_ids.append(entity_id) + + async with session_scope() as session: + await replace_trending( + session, entity_type=entity_type, time_window=time_window, entity_ids=entity_ids + ) + + logger.info( + "trending_refreshed", media_type=media_type, time_window=time_window, count=len(entity_ids) + ) + return len(entity_ids) diff --git a/app/services/movie_service.py b/app/services/movie_service.py index ce99d4a..cd2e86d 100644 --- a/app/services/movie_service.py +++ b/app/services/movie_service.py @@ -9,7 +9,7 @@ from sqlalchemy.ext.asyncio import AsyncSession from app.core.ids import aurora_id, parse_aurora_id from app.models.enums import ArtworkType, MovieStatus from app.models.movie import Movie -from app.repositories import movie_repository, polymorphic +from app.repositories import movie_repository, polymorphic, trending_repository from app.schemas.common import CollectionBriefOut, CompanyBriefOut, GenreOut, KeywordOut from app.schemas.movie import ( MovieCreditsOut, @@ -233,6 +233,30 @@ async def list_upcoming(session: AsyncSession, *, region: str, page: int, page_s ) +_TRENDING_WINDOWS = ("day", "week") + + +async def list_trending( + session: AsyncSession, *, time_window: str, page: int, page_size: int +) -> MovieListOut: + if time_window not in _TRENDING_WINDOWS: + raise InvalidMovieFilterError(f"unknown time_window: {time_window!r}") + + all_ids = await trending_repository.get_trending_entity_ids( + session, entity_type="movie", time_window=time_window + ) + total = len(all_ids) + start = (page - 1) * page_size + page_ids = all_ids[start : start + page_size] + movies = await movie_repository.get_movies_by_ids(session, page_ids) + results = [await _movie_summary_from_row(session, m) for m in movies] + total_pages = (total + page_size - 1) // page_size if total else 0 + + return MovieListOut( + page=page, page_size=page_size, total_results=total, total_pages=total_pages, results=results + ) + + async def list_movies( session: AsyncSession, *, diff --git a/app/services/tv_service.py b/app/services/tv_service.py index 8631a41..dbda0f4 100644 --- a/app/services/tv_service.py +++ b/app/services/tv_service.py @@ -10,7 +10,7 @@ from app.core.ids import aurora_id, parse_aurora_id from app.models.credit import Credit from app.models.enums import ArtworkType, CompanyType, SeriesStatus from app.models.tv import Series -from app.repositories import polymorphic, tv_repository +from app.repositories import polymorphic, trending_repository, tv_repository from app.schemas.common import CastCreditOut, CompanyBriefOut, CrewCreditOut, GenreOut, KeywordOut from app.schemas.tv import ( EpisodeBriefOut, @@ -316,6 +316,30 @@ async def build_series_summaries( return out +_TRENDING_WINDOWS = ("day", "week") + + +async def list_trending( + session: AsyncSession, *, time_window: str, page: int, page_size: int +) -> SeriesListOut: + if time_window not in _TRENDING_WINDOWS: + raise InvalidSeriesFilterError(f"unknown time_window: {time_window!r}") + + all_ids = await trending_repository.get_trending_entity_ids( + session, entity_type="series", time_window=time_window + ) + total = len(all_ids) + start = (page - 1) * page_size + page_ids = all_ids[start : start + page_size] + series_rows = await tv_repository.get_series_by_ids(session, page_ids) + results = [await _series_summary_from_row(session, s) for s in series_rows] + total_pages = (total + page_size - 1) // page_size if total else 0 + + return SeriesListOut( + page=page, page_size=page_size, total_results=total, total_pages=total_pages, results=results + ) + + async def list_series( session: AsyncSession, *, diff --git a/app/workers/tasks.py b/app/workers/tasks.py index 8854c21..b0dc6f6 100644 --- a/app/workers/tasks.py +++ b/app/workers/tasks.py @@ -19,6 +19,7 @@ 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 ( @@ -34,6 +35,7 @@ from app.services.ingestion.runner import ( run_season_ingestion, run_series_ingestion, ) +from app.services.ingestion.trending_ingestion import refresh_trending logger = get_logger(__name__) @@ -112,3 +114,18 @@ async def ingest_collection_task(ctx: dict[str, Any], job_id: str, external_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() diff --git a/app/workers/worker.py b/app/workers/worker.py index 488c8a1..061eff8 100644 --- a/app/workers/worker.py +++ b/app/workers/worker.py @@ -10,6 +10,7 @@ each job is picked up by exactly one worker. from __future__ import annotations from arq.connections import RedisSettings +from arq.cron import cron from app.core.config import get_settings from app.core.logging import configure_logging, get_logger @@ -20,6 +21,7 @@ from app.workers.tasks import ( ingest_person_task, ingest_season_task, ingest_series_task, + refresh_trending_task, ) logger = get_logger(__name__) @@ -43,6 +45,10 @@ class WorkerSettings: ingest_collection_task, download_artwork_task, ] + # Twice a day (UTC) rather than run_at_startup=True -- a worker + # container restart (common during deploys) must not hammer TMDb with + # a full trending refresh every time it comes up. + cron_jobs = [cron(refresh_trending_task, hour={3, 15}, minute=0)] on_startup = startup on_shutdown = shutdown redis_settings = RedisSettings.from_dsn(get_settings().REDIS_URL) diff --git a/docs/api.md b/docs/api.md index 70fc3d3..fe1883d 100644 --- a/docs/api.md +++ b/docs/api.md @@ -52,6 +52,7 @@ GET /api/v1/search?q=the%20matrix | GET | `/api/v1/movies/{aurora_id}/watch-providers` | | GET | `/api/v1/movies/now-playing` | | GET | `/api/v1/movies/upcoming` | +| GET | `/api/v1/movies/trending` | `GET /api/v1/movies` browses/discovers without a search query — the complement to `/search` (section 17, additive beyond the spec's minimum @@ -159,6 +160,22 @@ sourcing watch-provider data back to JustWatch specifically if you display it, not just crediting TMDb itself. Read TMDb's current terms for the exact required wording/logo before showing this in the frontend. +`GET /api/v1/movies/trending` mirrors TMDb's own `/trending/movie/{window}` +ranking rather than Aurora computing trending itself from popularity deltas +-- that would require re-syncing the entire catalog on a schedule, which +isn't feasible at any real scale, whereas TMDb already computes this well +from signals (real view/search velocity) Aurora has no access to. `window` +(`day` default | `week`), `page`, `page_size` — same response shape as +`GET /movies`. Refreshed by a worker cron job twice a day (00:00 and +12:00 UTC-ish — see `WorkerSettings.cron_jobs` in `app/workers/worker.py`); +`POST /api/v1/admin/refresh-trending` (admin-protected) triggers the same +refresh on demand, useful right after a fresh deploy so this isn't empty +until the next scheduled tick. Unlike recommendations' never-fabricate- +a-stub rule, a refresh actually ingests each trending item — there are only +~20 per window (TMDb's first page), cheap enough to fully ingest rather +than drop anything Aurora hasn't seen yet. `GET /series/trending` is the +same idea for series. + `GET /api/v1/movies/now-playing` and `GET /api/v1/movies/upcoming` use the `release_dates` data above rather than `release_date`/`year`: `region` (ISO 3166-1 alpha-2, default `US`), `page`, `page_size` — same @@ -181,6 +198,7 @@ now-playing, soonest for upcoming) rather than appearing twice. | GET | `/api/v1/series/{aurora_id}/images` | | GET | `/api/v1/series/{aurora_id}/videos` | | GET | `/api/v1/series/{aurora_id}/watch-providers` | +| GET | `/api/v1/series/trending` | | GET | `/api/v1/seasons/{aurora_id}` | | GET | `/api/v1/episodes/{aurora_id}` | @@ -290,6 +308,7 @@ want network-level isolation on top of the API key. | POST | `/api/v1/admin/ingest/season` | `{"tmdb_id": "1396", "season_number": 1}` | Ingests one season's full episode list. Requires the series to already be ingested (`ingest/series` first) — fails with a clear error otherwise. | | POST | `/api/v1/admin/ingest/collection` | `{"tmdb_id": "2344"}` | Ingests a franchise's own metadata (overview, poster/backdrop) and links its already-ingested member movies (see "Collections / Companies" above). Movie ingestion also lazily creates a bare name-only stub the first time it sees a `belongs_to_collection` reference — this is what actually fills it in. | | GET | `/api/v1/admin/ingest/jobs/{job_id}` | — | Poll job status: `pending` → `running` → `succeeded`/`failed`/`retrying`. | +| POST | `/api/v1/admin/refresh-trending` | — | Manually triggers the same refresh the twice-daily cron job runs (see "Trending" above). Not tracked via `ingest/jobs` — check `GET /movies/trending`/`GET /series/trending` directly instead of polling a job id. | ## Errors diff --git a/migrations/versions/0003_trending.py b/migrations/versions/0003_trending.py new file mode 100644 index 0000000..20ec826 --- /dev/null +++ b/migrations/versions/0003_trending.py @@ -0,0 +1,55 @@ +"""trending + +Revision ID: 0003_trending +Revises: 0002_watch_providers +Create Date: 2026-09-06 + +Adds trending_entries (section 17, additive): Aurora's mirror of TMDb's +own trending lists, refreshed on a schedule -- see +app/services/ingestion/trending_ingestion.py. +""" + +from __future__ import annotations + +from collections.abc import Sequence + +import sqlalchemy as sa +from alembic import op + +revision: str = "0003_trending" +down_revision: str | None = "0002_watch_providers" +branch_labels: str | Sequence[str] | None = None +depends_on: str | Sequence[str] | None = None + + +def upgrade() -> None: + op.create_table( + "trending_entries", + sa.Column("id", sa.Uuid(), primary_key=True), + sa.Column( + "created_at", + sa.DateTime(timezone=True), + server_default=sa.text("now()"), + nullable=False, + ), + sa.Column( + "updated_at", + sa.DateTime(timezone=True), + server_default=sa.text("now()"), + nullable=False, + ), + sa.Column("entity_type", sa.String(30), nullable=False), + sa.Column("entity_id", sa.Uuid(), nullable=False), + sa.Column("time_window", sa.String(10), nullable=False), + sa.Column("rank", sa.Integer(), nullable=False), + ) + op.create_index("ix_trending_entries_entity_id", "trending_entries", ["entity_id"]) + op.create_index( + "ix_trending_entries_window", "trending_entries", ["entity_type", "time_window"] + ) + + +def downgrade() -> None: + op.drop_index("ix_trending_entries_window", table_name="trending_entries") + op.drop_index("ix_trending_entries_entity_id", table_name="trending_entries") + op.drop_table("trending_entries") diff --git a/tests/integration/test_trending.py b/tests/integration/test_trending.py new file mode 100644 index 0000000..f304d35 --- /dev/null +++ b/tests/integration/test_trending.py @@ -0,0 +1,165 @@ +"""Integration tests for trending (section 17, additive): Aurora mirrors +TMDb's own /trending/{media_type}/{time_window} ranking rather than +computing its own from popularity deltas (see app/models/trending.py) -- +refresh_trending ingests/refreshes each item and replaces the window's +ranked snapshot. + +refresh_trending manages its own per-item session_scope() (so one failing +item doesn't roll back others already committed -- see its docstring), +so these tests read back through session_scope() too rather than the +db_session fixture, same convention as test_ingestion_retry.py: db_session +is accepted only to trigger the Postgres-reachability skip, real reads and +writes all go through session_scope() to stay on a connection that +actually sees committed data.""" + +from __future__ import annotations + +from typing import Any + +import pytest +from sqlalchemy.ext.asyncio import AsyncSession + +from app.core.database import session_scope +from app.providers.base import MetadataProvider +from app.repositories.trending_repository import get_trending_entity_ids, replace_trending +from app.services import movie_service, tv_service +from app.services.ingestion.trending_ingestion import refresh_trending +from tests.unit.test_normalizer import FAKE_MOVIE, FAKE_SERIES + + +class FakeTrendingProvider(MetadataProvider): + slug = "tmdb" + name = "The Movie Database (TMDb)" + + def __init__(self, movie_ids: list[int], series_ids: list[int]) -> None: + self._movie_ids = movie_ids + self._series_ids = series_ids + + async def get_movie(self, external_id: str) -> dict[str, Any]: + return {**FAKE_MOVIE, "id": int(external_id), "title": f"Movie {external_id}"} + + async def get_series(self, external_id: str) -> dict[str, Any]: + return {**FAKE_SERIES, "id": int(external_id), "name": f"Series {external_id}"} + + async def get_season(self, series_external_id: str, season_number: int) -> dict[str, Any]: + raise NotImplementedError + + async def get_person(self, external_id: str) -> dict[str, Any]: + raise NotImplementedError + + async def get_collection(self, external_id: str) -> dict[str, Any]: + raise NotImplementedError + + async def search_movie(self, query: str) -> list[dict[str, Any]]: + raise NotImplementedError + + async def get_trending(self, media_type: str, time_window: str) -> list[dict[str, Any]]: + ids = self._movie_ids if media_type == "movie" else self._series_ids + return [{"id": i} for i in ids] + + +@pytest.mark.asyncio +async def test_refresh_trending_ingests_and_ranks_movies(db_session: AsyncSession): + provider = FakeTrendingProvider(movie_ids=[901, 902, 903], series_ids=[]) + + count = await refresh_trending(provider, "movie", "day") + assert count == 3 + + async with session_scope() as session: + ids = await get_trending_entity_ids(session, entity_type="movie", time_window="day") + assert len(ids) == 3 + + +@pytest.mark.asyncio +async def test_refresh_trending_preserves_tmdb_rank_order(db_session: AsyncSession): + provider = FakeTrendingProvider(movie_ids=[901, 902, 903], series_ids=[]) + await refresh_trending(provider, "movie", "day") + + async with session_scope() as session: + out = await movie_service.list_trending(session, time_window="day", page=1, page_size=20) + + assert [m.title for m in out.results] == ["Movie 901", "Movie 902", "Movie 903"] + + +@pytest.mark.asyncio +async def test_refresh_trending_is_scoped_per_window(db_session: AsyncSession): + """A "week" refresh must not appear in "day" results, and vice versa.""" + + day_provider = FakeTrendingProvider(movie_ids=[901], series_ids=[]) + week_provider = FakeTrendingProvider(movie_ids=[902], series_ids=[]) + await refresh_trending(day_provider, "movie", "day") + await refresh_trending(week_provider, "movie", "week") + + async with session_scope() as session: + day_out = await movie_service.list_trending(session, time_window="day", page=1, page_size=20) + week_out = await movie_service.list_trending(session, time_window="week", page=1, page_size=20) + + assert [m.title for m in day_out.results] == ["Movie 901"] + assert [m.title for m in week_out.results] == ["Movie 902"] + + +@pytest.mark.asyncio +async def test_refresh_trending_replaces_stale_entries(db_session: AsyncSession): + """A movie that falls off trending must not linger from a prior + refresh -- this is a wholesale replace, not an accumulate.""" + + provider = FakeTrendingProvider(movie_ids=[901, 902], series_ids=[]) + await refresh_trending(provider, "movie", "day") + + provider_v2 = FakeTrendingProvider(movie_ids=[903], series_ids=[]) + await refresh_trending(provider_v2, "movie", "day") + + async with session_scope() as session: + out = await movie_service.list_trending(session, time_window="day", page=1, page_size=20) + assert [m.title for m in out.results] == ["Movie 903"] + + +@pytest.mark.asyncio +async def test_refresh_trending_ingests_series(db_session: AsyncSession): + provider = FakeTrendingProvider(movie_ids=[], series_ids=[1901, 1902]) + count = await refresh_trending(provider, "tv", "day") + assert count == 2 + + async with session_scope() as session: + out = await tv_service.list_trending(session, time_window="day", page=1, page_size=20) + assert [s.title for s in out.results] == ["Series 1901", "Series 1902"] + + +@pytest.mark.asyncio +async def test_list_trending_paginates(db_session: AsyncSession): + provider = FakeTrendingProvider(movie_ids=[901, 902, 903, 904], series_ids=[]) + await refresh_trending(provider, "movie", "day") + + async with session_scope() as session: + page1 = await movie_service.list_trending(session, time_window="day", page=1, page_size=2) + page2 = await movie_service.list_trending(session, time_window="day", page=2, page_size=2) + + assert page1.total_results == 4 + assert page1.total_pages == 2 + assert [m.title for m in page1.results] == ["Movie 901", "Movie 902"] + assert [m.title for m in page2.results] == ["Movie 903", "Movie 904"] + + +@pytest.mark.asyncio +async def test_list_trending_rejects_unknown_window(db_session: AsyncSession): + async with session_scope() as session: + with pytest.raises(movie_service.InvalidMovieFilterError): + await movie_service.list_trending(session, time_window="month", page=1, page_size=20) + + +@pytest.mark.asyncio +async def test_list_trending_returns_empty_when_never_refreshed(db_session: AsyncSession): + # Unlike the other tests here, this one's precondition is "no trending + # data exists yet" -- every other test calls refresh_trending, which + # always wipes its own (entity_type, time_window) scope before writing, + # making the prior state irrelevant. This one has no such call, so it + # must clear the scope itself rather than assume a clean slate (these + # tests commit for real via session_scope(), unlike db_session's + # per-test rollback, so a prior test's rows would otherwise leak in). + async with session_scope() as session: + await replace_trending(session, entity_type="movie", time_window="day", entity_ids=[]) + + async with session_scope() as session: + out = await movie_service.list_trending(session, time_window="day", page=1, page_size=20) + assert out.results == [] + assert out.total_results == 0