diff --git a/.env.example b/.env.example index 7cc0a87..5174557 100644 --- a/.env.example +++ b/.env.example @@ -23,3 +23,11 @@ SUPPORTED_PROVIDERS=netflix,disney_plus,prime_video,apple_tv_plus NEW_RELEASE_DAYS=90 RECENTLY_ADDED_DAYS=30 LEAVING_SOON_DAYS=30 + +# Bounded TMDB enrichment policy. Backfills can override max titles explicitly. +TMDB_ENRICHMENT_MAX_TITLES_PER_RUN=250 +TMDB_ENRICHMENT_UPCOMING_REFRESH_DAYS=7 +TMDB_ENRICHMENT_RECENT_REFRESH_DAYS=30 +TMDB_ENRICHMENT_SERIES_REFRESH_DAYS=90 +TMDB_ENRICHMENT_MOVIE_REFRESH_DAYS=180 +TMDB_ENRICHMENT_PROVIDER_REFRESH_DAYS=60 diff --git a/README.md b/README.md index f1304df..8af1ba0 100644 --- a/README.md +++ b/README.md @@ -175,11 +175,23 @@ make frontend-dev ``` Open `http://127.0.0.1:5173`. The v0.5 frontend reads API health, catalog -freshness, reference filters, Top 10, New Releases, and Recently Added from -FastAPI, proving the complete local serving path. The browser never calls either -upstream data provider. See the +freshness, reference filters, Top 10, New Releases, Recently Added, and Upcoming +from FastAPI, proving the complete local serving path. The browser never calls +either upstream data provider. See the [frontend guide](frontend/README.md) for configuration and quality commands. +Lifecycle-only titles can be enriched incrementally with canonical TMDB +metadata and current watch-provider evidence without repeating broad discovery: + +```bash +python -m ingestion.enrich_streaming_metadata --event-type upcoming --country GR +make dbt-publish +``` + +The command independently skips title IDs with retained metadata or provider +responses and reports its bounded +TMDB request count, and makes no Streaming Availability requests. + ## Dependency management `pyproject.toml` is the canonical project configuration. It contains package @@ -300,6 +312,32 @@ only be used for a bounded set while the incremental enrichment queue is built: python -m ingestion.run --max-pages 1 --enrich ``` +Plan the dedicated catalog enrichment locally before making any TMDB requests: + +```bash +python -m ingestion.enrich_catalog --mode incremental --dry-run +``` + +The initial broad backfill selects titles that have never retained metadata or +provider payloads. Set an explicit cap appropriate for the catalog size: + +```bash +python -m ingestion.enrich_catalog --mode backfill --max-titles 10000 --dry-run +python -m ingestion.enrich_catalog --mode backfill --max-titles 10000 +``` + +Afterward, routine incremental runs deduplicate titles, prioritize product +value, and refresh only new or stale payloads: + +```bash +python -m ingestion.enrich_catalog --mode incremental +``` + +Add `--metadata-only` to skip watch-provider payloads. Add +`--plan-output data/enrichment-plan.json` to save the complete local plan for +inspection. Planning never calls TMDB; only execution does. Completed batches +are retained as immutable Parquet files, allowing later plans to resume. + To override the configured regions for one run, repeat `--country`: ```bash diff --git a/api/tests/test_discovery_integration.py b/api/tests/test_discovery_integration.py index 76c44e1..61d9735 100644 --- a/api/tests/test_discovery_integration.py +++ b/api/tests/test_discovery_integration.py @@ -74,6 +74,8 @@ def _catalog_row( "tmdb_discovery" if is_available else "streaming_availability", datetime.now(UTC), "https://www.netflix.com/title/example" if provider_key == "netflix" else None, + None, + None, ) @@ -96,7 +98,7 @@ def _create_catalog(path: Path) -> None: is_available boolean, is_upcoming boolean, poster_path varchar, backdrop_path varchar, metadata_source varchar, availability_source varchar, last_updated_at timestamptz, - watch_url varchar + watch_url varchar, episode_count integer, season_count integer ) """ ) @@ -251,6 +253,35 @@ def test_region_and_provider_never_leak_across_http_routes(catalog_app: FastAPI) assert [item["tmdb_id"] for item in greek_disney.json()["items"]] == [5] +def test_title_search_uses_local_scoped_catalog(catalog_app: FastAPI) -> None: + response = _get( + catalog_app, + "/api/v1/discovery/search", + [("query", "Title 1"), ("region", "GR"), ("providers", "netflix")], + ) + other_region = _get( + catalog_app, + "/api/v1/discovery/search", + [("query", "Title 1"), ("region", "US"), ("providers", "netflix")], + ) + + assert response.status_code == 200 + assert response.json()["query"] == "Title 1" + assert [item["tmdb_id"] for item in response.json()["items"]] == [1] + assert other_region.json()["items"] == [] + + +def test_title_search_treats_sql_wildcards_as_text(catalog_app: FastAPI) -> None: + response = _get( + catalog_app, + "/api/v1/discovery/search", + [("query", "%_"), ("region", "GR"), ("providers", "netflix")], + ) + + assert response.status_code == 200 + assert response.json()["items"] == [] + + def test_api_exposes_verified_watch_urls_with_scoped_availability( catalog_app: FastAPI, ) -> None: @@ -322,6 +353,7 @@ def test_openapi_contains_the_complete_v04_contract(catalog_app: FastAPI) -> Non "/api/v1/catalog/genres", "/api/v1/catalog/filter-options", "/api/v1/discovery/top-10", + "/api/v1/discovery/search", "/api/v1/discovery/new-releases", "/api/v1/discovery/recently-added", "/api/v1/discovery/upcoming", diff --git a/api/tests/test_query_engine.py b/api/tests/test_query_engine.py index 8a402cd..446cc07 100644 --- a/api/tests/test_query_engine.py +++ b/api/tests/test_query_engine.py @@ -50,7 +50,9 @@ def _create_catalog(path: Path) -> None: metadata_source varchar, availability_source varchar, last_updated_at timestamp, - watch_url varchar + watch_url varchar, + episode_count integer, + season_count integer ) """ ) @@ -201,7 +203,7 @@ def _create_catalog(path: Path) -> None: datetime(2026, 8, 22), ), ] - rows = [(*row, None) for row in rows] + rows = [(*row, None, None, None) for row in rows] placeholders = ", ".join("?" for _ in rows[0]) connection.executemany( f"insert into main_marts.catalog_availability values ({placeholders})", diff --git a/docs/architecture.md b/docs/architecture.md index 68355d5..4840dd7 100644 --- a/docs/architecture.md +++ b/docs/architecture.md @@ -257,6 +257,12 @@ filters are applied. It must never claim to be an official provider Top 10. - Broad TMDB catalog backfills are discovery-only and reuse the metadata already present in discovery responses. - Incremental jobs enrich only new, changed, or prioritized TMDB titles. +- Catalog enrichment has two explicit modes. One-time `backfill` selects titles + with no retained metadata/provider payload. Routine `incremental` mode uses + configurable freshness windows and a per-run title cap, deduplicates + `(tmdb_id, content_type)` across regions and providers, and prioritizes + upcoming, recently added, recently released, Top 10, incomplete, and popular + titles. Planning reads only WatchPulse-owned data and supports dry runs. - Periodic reconciliation corrects missed upstream lifecycle changes. - HTTP clients use timeouts, conservative throttling, capped retries, and exponential backoff for transport errors, rate limits, and server errors. diff --git a/docs/decisions.md b/docs/decisions.md index aca5707..b3b1f92 100644 --- a/docs/decisions.md +++ b/docs/decisions.md @@ -335,6 +335,24 @@ frontend renders an external link only for validated HTTPS values, opens it in a separate tab with safe relationship attributes, and leaves other provider badges informational. +## ADR-020: Incrementally enrich lifecycle-only titles from TMDB + +Status: Accepted (2026-08-24) + +Streaming lifecycle events can introduce upcoming or newly added TMDB IDs that +are absent from the current provider discovery snapshot. Embedded lifecycle +metadata may omit descriptions and does not provide TMDB poster paths. + +WatchPulse extracts the stable TMDB ID from retained lifecycle data and fetches +metadata plus current watch-provider evidence in a separate bounded ingestion +command. Raw responses are stored locally. `stg_tmdb_metadata` takes precedence +over broad discovery and streaming metadata fallbacks, while targeted provider +evidence reconciles current availability with lifecycle announcements. An +already-available series may therefore be both current and upcoming; the UI +labels this as a new season without claiming a season number absent upstream. +Browser requests never trigger enrichment, and each retained response type is +skipped independently on subsequent runs. + ## Open decisions These remain unresolved and should receive new ADRs when decided: diff --git a/docs/roadmap.md b/docs/roadmap.md index 2671eac..2b788ce 100644 --- a/docs/roadmap.md +++ b/docs/roadmap.md @@ -176,6 +176,12 @@ independent loading, empty, failure, and retry behavior. The sixth increment adds Recently Added based on provider addition time rather than content release date, and extracts a reusable lifecycle rail for consistent loading, empty, failure, retry, and card behavior across the remaining sections. +The seventh increment adds Upcoming from future provider arrival evidence and +surfaces expected arrival dates directly on cards without browser-side upstream +requests. Targeted TMDB reconciliation lets an available series also carry an +upcoming-season state without guessing a season number. +It also consolidates discovery into a content-first workspace with a sticky +desktop filter sidebar, denser rails, and a responsive stacked mobile layout. Verified provider deep links are retained from historical lifecycle payloads and exposed as optional safe card actions; TMDB-only rows remain unlinked rather than receiving guessed destinations. diff --git a/frontend/README.md b/frontend/README.md index 0e8768e..9047701 100644 --- a/frontend/README.md +++ b/frontend/README.md @@ -44,6 +44,12 @@ state and the same query-parameter contract as every FastAPI discovery section. The Top 10, New Releases, and Recently Added rails query only FastAPI and render reusable poster cards; ranking badges remain specific to Top 10. Recently Added uses provider addition time and remains distinct from content release recency. +Upcoming uses future provider arrival evidence and displays the expected date. +An already-available series may also appear there as a new-season announcement. +On desktop, region, provider, and global controls remain in a sticky sidebar +while discovery rails use the wider content workspace. The layout stacks for +touch and narrow screens, and every control still updates local API queries +immediately without an Apply action. Poster artwork is loaded from TMDB's public image CDN using the ingested `poster_path`; this is an image asset request, not a TMDB API lookup, and it uses no API key or discovery quota. Missing or failed images have a local diff --git a/frontend/public/tmdb-logo.svg b/frontend/public/tmdb-logo.svg new file mode 100644 index 0000000..ca6c64d --- /dev/null +++ b/frontend/public/tmdb-logo.svg @@ -0,0 +1 @@ +The Movie Database (TMDB) diff --git a/frontend/src/App.test.tsx b/frontend/src/App.test.tsx index 7579efc..6ec8804 100644 --- a/frontend/src/App.test.tsx +++ b/frontend/src/App.test.tsx @@ -23,8 +23,15 @@ test("shows catalog readiness returned by the local API", async () => { render(); expect(screen.getByText("Connecting to WatchPulse…")).toBeInTheDocument(); - expect(await screen.findByText(/Updated/)).toBeInTheDocument(); + const timeZone = new Intl.DateTimeFormat().resolvedOptions().timeZone || "UTC"; + expect(await screen.findByText(/Updated/)).toHaveTextContent(`(${timeZone})`); expect(await screen.findByRole("button", { name: /Netflix/ })).toBeInTheDocument(); + expect(screen.getByRole("img", { name: "The Movie Database (TMDB)" })).toBeInTheDocument(); + expect(screen.getByText(/uses TMDB and the TMDB APIs/)).toBeInTheDocument(); + expect(screen.getByRole("link", { name: "JustWatch" })).toHaveAttribute( + "href", + "https://www.justwatch.com", + ); }); test("shows an honest error when the catalog is unavailable", async () => { diff --git a/frontend/src/App.tsx b/frontend/src/App.tsx index 241f286..5bc0e87 100644 --- a/frontend/src/App.tsx +++ b/frontend/src/App.tsx @@ -5,6 +5,8 @@ import { GlobalFilters } from "./components/GlobalFilters"; import { NewReleasesRail } from "./components/NewReleasesRail"; import { RecentlyAddedRail } from "./components/RecentlyAddedRail"; import { TopTenRail } from "./components/TopTenRail"; +import { TitleSearch } from "./components/TitleSearch"; +import { UpcomingRail } from "./components/UpcomingRail"; import { type CatalogScope, type GlobalFilters as FilterValue } from "./discovery"; import { loadPreferences } from "./preferences"; import "./styles.css"; @@ -14,10 +16,14 @@ type LoadState = | { kind: "ready"; status: CatalogStatus } | { kind: "error"; message: string }; -const formatRefresh = (value: string) => - new Intl.DateTimeFormat(undefined, { dateStyle: "medium", timeStyle: "short" }).format( - new Date(value), - ); +const formatRefresh = (value: string) => { + const formatter = new Intl.DateTimeFormat(undefined, { + dateStyle: "medium", + timeStyle: "short", + }); + const timeZone = formatter.resolvedOptions().timeZone || "UTC"; + return `${formatter.format(new Date(value))} (${timeZone})`; +}; export default function App() { const [state, setState] = useState({ kind: "loading" }); @@ -58,26 +64,52 @@ export default function App() { Updated {formatRefresh(state.status.freshness.latest_source_updated_at)} ) : v0.5 preview} -
-

YOUR STREAMS. ONE DECISION.

-

Find your next great watch.

-

WatchPulse searches the catalog available on your services and in your region—without the endless scroll.

-
- {state.kind === "loading" &&

Connecting to WatchPulse…

} - {state.kind === "error" &&

{state.message}

} - {state.kind === "ready" && } - {scope && scope.providers.length > 0 && ( - <> - - - - - - )} - {scope && scope.providers.length === 0 && ( -

Select at least one streaming service to refine your catalog.

- )} -

Upcoming and Leaving Soon arrive next.

+
+ + +
+
+

YOUR STREAM. ONE DECISION.

+

Find your next great watch.

+

WatchPulse searches the catalog available on your services and in your region—without the endless scroll.

+
+ {scope && scope.providers.length > 0 && ( + <> + + + + + + + )} + {scope && scope.providers.length === 0 && ( +

Select at least one streaming service to start discovering titles.

+ )} +

Leaving Soon and title details arrive next.

+
+
+
+ + The Movie Database (TMDB) + +
+

This website uses TMDB and the TMDB APIs but is not endorsed, certified, or otherwise approved by TMDB.

+

Streaming availability data provided in part by JustWatch.

+
+
); } diff --git a/frontend/src/api/catalog.ts b/frontend/src/api/catalog.ts index b9ce4be..aad9c8f 100644 --- a/frontend/src/api/catalog.ts +++ b/frontend/src/api/catalog.ts @@ -52,6 +52,8 @@ export type CatalogItem = { release_date: string | null; release_year: number | null; runtime_minutes: number | null; + episode_count?: number | null; + season_count?: number | null; original_language: string | null; genre_ids: number[]; genre_names: string[]; @@ -161,3 +163,30 @@ export async function getRecentlyAdded( ); return response.items; } + +export async function getUpcoming( + scope: CatalogScope, + filters: GlobalFilters, + signal?: AbortSignal, +): Promise { + const response = await getJson<{ items: CatalogItem[] }>( + `/api/v1/discovery/upcoming?${discoveryParams(scope, filters)}`, + signal, + ); + return response.items; +} + +export async function searchTitles( + query: string, + scope: CatalogScope, + filters: GlobalFilters, + signal?: AbortSignal, +): Promise { + const params = discoveryParams(scope, filters); + params.set("query", query); + const response = await getJson<{ items: CatalogItem[] }>( + `/api/v1/discovery/search?${params}`, + signal, + ); + return response.items; +} diff --git a/frontend/src/components/CatalogRail.tsx b/frontend/src/components/CatalogRail.tsx index 97073a4..ca923bf 100644 --- a/frontend/src/components/CatalogRail.tsx +++ b/frontend/src/components/CatalogRail.tsx @@ -12,7 +12,6 @@ type CatalogLoader = ( type CatalogRailProps = { id: string; title: string; - subtitle: string; loadingLabel: string; emptyTitle: string; emptyHint: string; @@ -20,6 +19,7 @@ type CatalogRailProps = { scope: CatalogScope; filters: GlobalFilters; load: CatalogLoader; + lifecycleDate?: "available_since" | "available_from"; }; type RailState = @@ -30,7 +30,6 @@ type RailState = export function CatalogRail({ id, title, - subtitle, loadingLabel, emptyTitle, emptyHint, @@ -38,6 +37,7 @@ export function CatalogRail({ scope, filters, load, + lifecycleDate, }: CatalogRailProps) { const [state, setState] = useState({ kind: "loading" }); const [retry, setRetry] = useState(0); @@ -61,10 +61,7 @@ export function CatalogRail({ return (
-
-

{subtitle}

-

{title}

-
+

{title}

{state.kind === "ready" && state.items.length > 0 && {state.items.length} titles}
@@ -91,7 +88,11 @@ export function CatalogRail({ {state.kind === "ready" && state.items.length > 0 && (
{state.items.map((item) => ( - + ))}
)} diff --git a/frontend/src/components/DiscoverySetup.test.tsx b/frontend/src/components/DiscoverySetup.test.tsx index 1b88e95..2ac121b 100644 --- a/frontend/src/components/DiscoverySetup.test.tsx +++ b/frontend/src/components/DiscoverySetup.test.tsx @@ -49,3 +49,22 @@ test("changing region reloads providers and removes an invalid selection", async localStorage.getItem("watchpulse.discovery-preferences.v1") ?? "{}", )).toMatchObject({ region: "US", providers: [] })); }); + +test("selects and clears every region-supported provider", async () => { + vi.spyOn(globalThis, "fetch") + .mockImplementationOnce(() => jsonResponse({ regions: [{ code: "GR" }] })) + .mockImplementationOnce(() => jsonResponse({ region: "GR", providers: [ + { key: "netflix", name: "Netflix" }, + { key: "disney_plus", name: "Disney+" }, + ] })); + + render(); + await screen.findByRole("button", { name: /Netflix/ }); + const selectAll = screen.getByRole("button", { name: "Select all" }); + fireEvent.click(selectAll); + + expect(screen.getByRole("button", { name: /Netflix/ })).toHaveAttribute("aria-pressed", "true"); + expect(screen.getByRole("button", { name: /Disney\+/ })).toHaveAttribute("aria-pressed", "true"); + fireEvent.click(screen.getByRole("button", { name: "Clear all" })); + expect(screen.getByRole("button", { name: /Netflix/ })).toHaveAttribute("aria-pressed", "false"); +}); diff --git a/frontend/src/components/DiscoverySetup.tsx b/frontend/src/components/DiscoverySetup.tsx index d66cbd6..1e67875 100644 --- a/frontend/src/components/DiscoverySetup.tsx +++ b/frontend/src/components/DiscoverySetup.tsx @@ -75,6 +75,11 @@ export function DiscoverySetup({ onScopeChange }: DiscoverySetupProps) { ); }; + const allSelected = providers.length > 0 && selectedProviders.length === providers.length; + const toggleAllProviders = () => setSelectedProviders( + allSelected ? [] : providers.map(({ key }) => key), + ); + return (
@@ -101,7 +106,12 @@ export function DiscoverySetup({ onScopeChange }: DiscoverySetupProps) {
- Streaming services + + Streaming services + +
{providers.map((provider) => { const selected = selectedProviders.includes(provider.key); diff --git a/frontend/src/components/NewReleasesRail.test.tsx b/frontend/src/components/NewReleasesRail.test.tsx index 2aecb26..a559530 100644 --- a/frontend/src/components/NewReleasesRail.test.tsx +++ b/frontend/src/components/NewReleasesRail.test.tsx @@ -55,8 +55,8 @@ test("renders unranked new releases and sends the shared filters", async () => { expect(await screen.findByRole("article", { name: "A New Release" })).toBeInTheDocument(); expect(screen.queryByText("1")).not.toBeInTheDocument(); expect(screen.getByLabelText("Disney+; direct link unavailable")).toHaveAttribute( - "title", - "Direct link unavailable", + "data-tooltip", + "Available on this provider, but a direct title link is unavailable.", ); expect(String(fetchMock.mock.calls[0][0])).toContain( "new-releases?region=GR&providers=disney_plus&content_type=movie&genre_ids=18", diff --git a/frontend/src/components/NewReleasesRail.tsx b/frontend/src/components/NewReleasesRail.tsx index a18a23e..7a9b38c 100644 --- a/frontend/src/components/NewReleasesRail.tsx +++ b/frontend/src/components/NewReleasesRail.tsx @@ -12,7 +12,6 @@ export function NewReleasesRail({ scope, filters }: NewReleasesRailProps) { ); } diff --git a/frontend/src/components/TitleCard.test.tsx b/frontend/src/components/TitleCard.test.tsx new file mode 100644 index 0000000..97ef05c --- /dev/null +++ b/frontend/src/components/TitleCard.test.tsx @@ -0,0 +1,76 @@ +import { render, screen } from "@testing-library/react"; +import { expect, test } from "vitest"; +import type { CatalogItem } from "../api/catalog"; +import { TitleCard } from "./TitleCard"; + +const series: CatalogItem = { + tmdb_id: 42, + content_type: "tv", + title: "Current Series", + original_title: "Current Series", + overview: null, + release_date: "2024-01-01", + release_year: 2024, + runtime_minutes: 50, + episode_count: 9, + season_count: 1, + original_language: "en", + genre_ids: [18], + genre_names: ["Drama"], + tmdb_rating: 7.5, + vote_count: 100, + popularity_score: 10, + poster_path: null, + backdrop_path: null, + metadata_source: "tmdb", + last_updated_at: "2026-08-25T00:00:00Z", + availabilities: [{ + provider_key: "netflix", + provider_name: "Netflix", + monetization_type: "subscription", + available_since: null, + available_from: null, + expires_on: null, + is_available: true, + is_upcoming: false, + source: "tmdb", + watch_url: null, + }, { + provider_key: "disney_plus", + provider_name: "Disney+", + monetization_type: "subscription", + available_since: null, + available_from: null, + expires_on: null, + is_available: true, + is_upcoming: false, + source: "tmdb", + watch_url: null, + }, { + provider_key: "prime_video", + provider_name: "Prime Video", + monetization_type: "subscription", + available_since: null, + available_from: null, + expires_on: null, + is_available: true, + is_upcoming: false, + source: "tmdb", + watch_url: null, + }], +}; + +test("shows total episodes for a currently available series", () => { + render(); + + expect(screen.getByText("1 season · 9 episodes", { exact: false })).toBeInTheDocument(); + expect(screen.getByText("★ 7.5")).toBeInTheDocument(); + expect(screen.queryByText("50 min", { exact: false })).not.toBeInTheDocument(); + expect(screen.getByLabelText("Netflix; direct link unavailable")).toHaveAttribute( + "data-tooltip", + "Available on this provider, but a direct title link is unavailable.", + ); + expect(screen.getByLabelText("Disney+; direct link unavailable")).toBeInTheDocument(); + expect(screen.getByLabelText("Prime Video; direct link unavailable")).toBeInTheDocument(); + expect(screen.queryByText("+1")).not.toBeInTheDocument(); +}); diff --git a/frontend/src/components/TitleCard.tsx b/frontend/src/components/TitleCard.tsx index 16050e0..a3cdd88 100644 --- a/frontend/src/components/TitleCard.tsx +++ b/frontend/src/components/TitleCard.tsx @@ -9,6 +9,7 @@ const IMAGE_BASE_URL = ( type TitleCardProps = { item: CatalogItem; rank?: number; + lifecycleDate?: "available_since" | "available_from"; }; function safeWatchUrl(value: string | null): string | null { @@ -21,7 +22,13 @@ function safeWatchUrl(value: string | null): string | null { } } -export function TitleCard({ item, rank }: TitleCardProps) { +const formatLifecycleDate = (value: string) => new Intl.DateTimeFormat(undefined, { + month: "short", + day: "numeric", + year: "numeric", +}).format(new Date(value)); + +export function TitleCard({ item, rank, lifecycleDate }: TitleCardProps) { const [imageFailed, setImageFailed] = useState(false); const posterUrl = item.poster_path && !imageFailed ? `${IMAGE_BASE_URL}/${item.poster_path.replace(/^\//, "")}` @@ -31,6 +38,27 @@ export function TitleCard({ item, rank }: TitleCardProps) { ).values()]; const tmdbUrl = `https://www.themoviedb.org/${item.content_type === "tv" ? "tv" : "movie"}/${item.tmdb_id}`; const genreLabel = item.genre_names.slice(0, 2).join(" · "); + const hasRating = item.tmdb_rating !== null && item.tmdb_rating > 0; + const lifecycleDates = lifecycleDate + ? item.availabilities + .map((availability) => availability[lifecycleDate]) + .filter((value): value is string => value !== null) + .sort() + : []; + const lifecycleValue = lifecycleDate === "available_since" + ? lifecycleDates.at(-1) + : lifecycleDates.at(0); + const hasCurrentAvailability = item.availabilities.some((availability) => availability.is_available); + const lifecycleLabel = lifecycleDate === "available_from" + ? item.content_type === "tv" && hasCurrentAvailability ? "New season coming" : "Coming" + : "Added"; + const seriesDetails = item.content_type === "tv" && lifecycleDate !== "available_from" + ? [ + item.season_count ? `${item.season_count} ${item.season_count === 1 ? "season" : "seasons"}` : null, + item.episode_count ? `${item.episode_count} ${item.episode_count === 1 ? "episode" : "episodes"}` : null, + ].filter(Boolean).join(" · ") + : null; + const formatDetail = seriesDetails || (item.runtime_minutes ? `${item.runtime_minutes} min` : null); return (
@@ -54,25 +82,28 @@ export function TitleCard({ item, rank }: TitleCardProps) { {item.overview} )} - {item.tmdb_rating !== null && ★ {item.tmdb_rating.toFixed(1)}} + + {hasRating ? `★ ${item.tmdb_rating!.toFixed(1)}` : "N/A"} +

{item.title}

- {[item.release_year, item.content_type === "tv" ? "Series" : "Movie", item.runtime_minutes ? `${item.runtime_minutes} min` : null] + {[item.release_year, item.content_type === "tv" ? "Series" : "Movie", formatDetail] .filter(Boolean) .join(" · ")}

{genreLabel &&

{genreLabel}

} + {lifecycleValue && ( +

+ {lifecycleLabel} {formatLifecycleDate(lifecycleValue)} +

+ )}
provider_name).join(", ")}`}> - {providers.slice(0, 2).map((availability) => { + {providers.map((availability) => { const watchUrl = safeWatchUrl(availability.watch_url); const contents = ( - <> - - {availability.provider_name} - {watchUrl && } - + ); return watchUrl ? ( {contents} @@ -88,15 +120,15 @@ export function TitleCard({ item, rank }: TitleCardProps) { ) : ( {contents} ); })} - {providers.length > 2 && +{providers.length - 2}}
diff --git a/frontend/src/components/TitleSearch.test.tsx b/frontend/src/components/TitleSearch.test.tsx new file mode 100644 index 0000000..a4c7bee --- /dev/null +++ b/frontend/src/components/TitleSearch.test.tsx @@ -0,0 +1,83 @@ +import { fireEvent, render, screen, waitFor } from "@testing-library/react"; +import { afterEach, expect, test, vi } from "vitest"; +import { TitleSearch } from "./TitleSearch"; +import { EMPTY_FILTERS } from "../discovery"; + +afterEach(() => vi.restoreAllMocks()); + +test("searches the selected local catalog after two characters", async () => { + const fetchMock = vi.spyOn(globalThis, "fetch").mockResolvedValue( + new Response(JSON.stringify({ + items: [{ + tmdb_id: 42, + content_type: "movie", + title: "Monsters", + original_title: "Monsters", + overview: null, + release_date: "2024-01-01", + release_year: 2024, + runtime_minutes: 100, + original_language: "en", + genre_ids: [18], + genre_names: ["Drama"], + tmdb_rating: 7.2, + vote_count: 100, + popularity_score: 20, + poster_path: "/poster.jpg", + backdrop_path: null, + metadata_source: "tmdb", + last_updated_at: "2026-08-25T10:00:00Z", + availabilities: [{ provider_name: "Netflix" }], + }], + }), { status: 200 }), + ); + render( + , + ); + + const searchbox = screen.getByRole("searchbox"); + fireEvent.focus(searchbox); + fireEvent.change(searchbox, { target: { value: "Mon" } }); + expect(await screen.findByText("Monsters", {}, { timeout: 1500 })).toBeInTheDocument(); + expect(screen.getByText("Netflix")).toBeInTheDocument(); + expect(screen.getByText("7.2")).toHaveClass("search-rating"); + await waitFor(() => expect(fetchMock).toHaveBeenCalledTimes(1)); + const url = String(fetchMock.mock.calls[0][0]); + expect(url).toContain("query=Mon"); + expect(url).toContain("region=GR"); + expect(url).toContain("providers=netflix"); +}); + +test("does not request a one-character query", () => { + const fetchMock = vi.spyOn(globalThis, "fetch"); + render( + , + ); + fireEvent.change(screen.getByRole("searchbox"), { target: { value: "M" } }); + expect(fetchMock).not.toHaveBeenCalled(); +}); + +test("makes the active search filters visible", () => { + render( + , + ); + + expect(screen.getByText(/Searching within:/).closest("p")).toHaveTextContent( + "GR · 2 services · Series · Up to 120 min · Rating 7+ · EN", + ); +}); diff --git a/frontend/src/components/TitleSearch.tsx b/frontend/src/components/TitleSearch.tsx new file mode 100644 index 0000000..24d0966 --- /dev/null +++ b/frontend/src/components/TitleSearch.tsx @@ -0,0 +1,120 @@ +import { useEffect, useId, useRef, useState } from "react"; +import { searchTitles, type CatalogItem } from "../api/catalog"; +import { type CatalogScope, type GlobalFilters } from "../discovery"; + +type Props = { scope: CatalogScope; filters: GlobalFilters }; + +const posterUrl = (path: string | null) => + path ? `https://image.tmdb.org/t/p/w92${path}` : null; + +export function TitleSearch({ scope, filters }: Props) { + const [query, setQuery] = useState(""); + const [results, setResults] = useState([]); + const [status, setStatus] = useState<"idle" | "loading" | "ready" | "error">("idle"); + const [focused, setFocused] = useState(false); + const blurTimer = useRef(undefined); + const listboxId = useId(); + const normalized = query.trim(); + const activeScope = [ + scope.region, + `${scope.providers.length} ${scope.providers.length === 1 ? "service" : "services"}`, + filters.contentType === "movie" ? "Movies" : filters.contentType === "tv" ? "Series" : "Movies & series", + filters.genreIds.length ? `${filters.genreIds.length} ${filters.genreIds.length === 1 ? "genre" : "genres"}` : null, + filters.runtimeMax !== null ? `Up to ${filters.runtimeMax} min` : null, + filters.releaseYearFrom !== null || filters.releaseYearTo !== null + ? `${filters.releaseYearFrom ?? "Any"}–${filters.releaseYearTo ?? "Now"}` + : null, + filters.ratingMin !== null ? `Rating ${filters.ratingMin}+` : null, + filters.language ? filters.language.toUpperCase() : null, + ].filter((value): value is string => value !== null); + + useEffect(() => { + if (normalized.length < 2) { + setResults([]); + setStatus("idle"); + return; + } + const controller = new AbortController(); + const timer = window.setTimeout(() => { + setStatus("loading"); + searchTitles(normalized, scope, filters, controller.signal) + .then((items) => { + setResults(items); + setStatus("ready"); + }) + .catch(() => { + if (!controller.signal.aborted) setStatus("error"); + }); + }, 300); + return () => { + window.clearTimeout(timer); + controller.abort(); + }; + }, [normalized, scope, filters]); + + const open = focused && normalized.length >= 2; + return ( +
+ +
+ + setQuery(event.target.value)} + onFocus={() => { + window.clearTimeout(blurTimer.current); + setFocused(true); + }} + onBlur={() => { + blurTimer.current = window.setTimeout(() => setFocused(false), 120); + }} + onKeyDown={(event) => { + if (event.key === "Escape") setFocused(false); + }} + /> + {status === "loading" && } +
+

+ Searching within: {activeScope.join(" · ")} +

+ {open && ( + + )} +
+ ); +} diff --git a/frontend/src/components/TopTenRail.test.tsx b/frontend/src/components/TopTenRail.test.tsx index e045937..7f8834a 100644 --- a/frontend/src/components/TopTenRail.test.tsx +++ b/frontend/src/components/TopTenRail.test.tsx @@ -52,7 +52,7 @@ test("renders ranked local catalog results and sends every active filter", async expect(screen.getByLabelText("Loading Top 10")).toBeInTheDocument(); expect(await screen.findByRole("article", { name: /Number 1.*Spider-Man/ })).toBeInTheDocument(); - expect(screen.getByText("Netflix")).toBeInTheDocument(); + expect(screen.getByLabelText("Available on Netflix")).toBeInTheDocument(); expect(screen.getByText("2021 · Movie · 148 min")).toBeInTheDocument(); expect(screen.getByText("Action · Adventure")).toBeInTheDocument(); expect(screen.getByText("A multiverse adventure.")).toBeInTheDocument(); @@ -64,6 +64,10 @@ test("renders ranked local catalog results and sends every active filter", async "href", "https://www.netflix.com/title/81727396", ); + expect(screen.getByRole("link", { name: "Open Spider-Man: No Way Home on Netflix" })).toHaveAttribute( + "data-tooltip", + "Watch on Netflix", + ); expect(String(fetchMock.mock.calls[0][0])).toContain( "region=GR&providers=netflix&content_type=movie&genre_ids=28&rating_min=7", ); diff --git a/frontend/src/components/TopTenRail.tsx b/frontend/src/components/TopTenRail.tsx index 81f514a..d70de9e 100644 --- a/frontend/src/components/TopTenRail.tsx +++ b/frontend/src/components/TopTenRail.tsx @@ -33,10 +33,7 @@ export function TopTenRail({ scope, filters }: TopTenRailProps) { return (
-
-

Popular across your services

-

Top 10 for you

-
+

Top 10

{state.kind === "ready" && state.items.length > 0 && {state.items.length} titles}
diff --git a/frontend/src/components/UpcomingRail.test.tsx b/frontend/src/components/UpcomingRail.test.tsx new file mode 100644 index 0000000..3af1e7d --- /dev/null +++ b/frontend/src/components/UpcomingRail.test.tsx @@ -0,0 +1,102 @@ +import { render, screen } from "@testing-library/react"; +import { afterEach, expect, test, vi } from "vitest"; +import { EMPTY_FILTERS } from "../discovery"; +import { UpcomingRail } from "./UpcomingRail"; + +afterEach(() => vi.restoreAllMocks()); + +const item = { + tmdb_id: 999, + content_type: "tv", + title: "Coming Soon", + original_title: "Coming Soon", + overview: "A series announced for a future streaming date.", + release_date: "2026-09-01", + release_year: 2026, + runtime_minutes: 50, + episode_count: 9, + season_count: 1, + original_language: "en", + genre_ids: [18], + genre_names: ["Drama"], + tmdb_rating: null, + vote_count: 0, + popularity_score: 20, + poster_path: "/poster.jpg", + backdrop_path: null, + metadata_source: "streaming_availability", + last_updated_at: "2026-08-23T10:00:00", + availabilities: [{ + provider_key: "prime_video", + provider_name: "Prime Video", + monetization_type: "flatrate", + available_since: null, + available_from: "2026-09-03T00:00:00Z", + expires_on: null, + is_available: false, + is_upcoming: true, + source: "streaming_availability", + watch_url: null, + }], +}; + +test("renders future provider arrivals with their expected date", async () => { + const fetchMock = vi.spyOn(globalThis, "fetch").mockResolvedValue( + new Response(JSON.stringify({ items: [item] }), { status: 200 }), + ); + render( + , + ); + + expect(screen.getByLabelText("Loading Upcoming")).toBeInTheDocument(); + expect(await screen.findByRole("article", { name: "Coming Soon" })).toBeInTheDocument(); + expect(screen.getByText(/Coming .*2026/)).toBeInTheDocument(); + expect(screen.queryByText("9 episodes")).not.toBeInTheDocument(); + expect(screen.queryByText("1 season", { exact: false })).not.toBeInTheDocument(); + expect(screen.getByText("N/A")).toBeInTheDocument(); + expect(String(fetchMock.mock.calls[0][0])).toContain( + "upcoming?region=GR&providers=prime_video&content_type=tv", + ); +}); + +test("shows an honest empty state when no arrivals are announced", async () => { + vi.spyOn(globalThis, "fetch").mockResolvedValue( + new Response(JSON.stringify({ items: [] }), { status: 200 }), + ); + render( + , + ); + + expect(await screen.findByText("No upcoming titles match these filters.")).toBeInTheDocument(); + expect(screen.getByText(/depends on announcements/)).toBeInTheDocument(); +}); + +test("labels an upcoming TV event as a new season when the series is already available", async () => { + const availableSeries = { + ...item, + availabilities: [{ + ...item.availabilities[0], + is_available: true, + available_since: "2024-09-19T00:00:00Z", + }], + }; + vi.spyOn(globalThis, "fetch").mockResolvedValue( + new Response(JSON.stringify({ items: [availableSeries] }), { status: 200 }), + ); + + render( + , + ); + + expect(await screen.findByText(/New season coming .*2026/)).toBeInTheDocument(); + expect(screen.queryByText("Available now")).not.toBeInTheDocument(); +}); diff --git a/frontend/src/components/UpcomingRail.tsx b/frontend/src/components/UpcomingRail.tsx new file mode 100644 index 0000000..4926644 --- /dev/null +++ b/frontend/src/components/UpcomingRail.tsx @@ -0,0 +1,25 @@ +import { getUpcoming } from "../api/catalog"; +import type { CatalogScope, GlobalFilters } from "../discovery"; +import { CatalogRail } from "./CatalogRail"; + +type UpcomingRailProps = { + scope: CatalogScope; + filters: GlobalFilters; +}; + +export function UpcomingRail({ scope, filters }: UpcomingRailProps) { + return ( + + ); +} diff --git a/frontend/src/styles.css b/frontend/src/styles.css index d7525c4..980a5ab 100644 --- a/frontend/src/styles.css +++ b/frontend/src/styles.css @@ -4,7 +4,7 @@ * { box-sizing: border-box; } body { margin: 0; min-width: 320px; min-height: 100vh; } body::before { position: fixed; inset: 0; z-index: -1; content: ""; background: radial-gradient(circle at 75% 10%, rgb(118 47 231 / 18%), transparent 34rem), radial-gradient(circle at 15% 80%, rgb(225 44 106 / 10%), transparent 30rem), #090b10; } -main { width: min(1120px, calc(100% - 40px)); margin: 0 auto; } +main { width: min(1480px, calc(100% - 40px)); margin: 0 auto; } .nav { display: flex; align-items: center; justify-content: space-between; padding: 32px 0; } .brand { display: flex; gap: 10px; align-items: center; color: inherit; font: 800 21px Manrope, sans-serif; text-decoration: none; } .pulse { width: 17px; height: 17px; border: 4px solid #a971ff; border-radius: 50%; box-shadow: 0 0 22px #8c50e9; } @@ -72,6 +72,7 @@ h2 { margin: 0; font: 700 22px Manrope, sans-serif; } .rail-heading > span { color: #696b74; font-size: 12px; } .title-row { display: flex; min-height: 395px; align-items: flex-start; gap: 18px; padding: 5px 4px 22px; overflow-x: auto; overscroll-behavior-inline: contain; scrollbar-color: #39323f transparent; scroll-snap-type: inline proximity; } .title-card { position: relative; isolation: isolate; flex: 0 0 190px; min-width: 0; scroll-snap-align: start; } +.title-card:hover, .title-card:focus-within { z-index: 20; } .rank { position: absolute; z-index: 3; top: 10px; left: 10px; display: grid; min-width: 46px; height: 46px; padding: 0 8px; place-items: center; border: 1px solid rgb(255 255 255 / 24%); border-radius: 12px; color: #fff; background: linear-gradient(145deg, rgb(17 12 24 / 94%), rgb(75 38 119 / 92%)); box-shadow: 0 8px 22px rgb(0 0 0 / 45%); font: 800 25px/1 Manrope, sans-serif; text-shadow: 0 2px 8px #000; backdrop-filter: blur(8px); } .poster-frame { position: relative; z-index: 1; aspect-ratio: 2 / 3; overflow: hidden; border-radius: 12px; background: #171921; box-shadow: 0 18px 40px rgb(0 0 0 / 42%); } .poster-frame:focus-visible { outline: 2px solid #a66ee5; outline-offset: 3px; } @@ -82,19 +83,23 @@ h2 { margin: 0; font: 700 22px Manrope, sans-serif; } .poster-frame:hover .poster-description, .poster-frame:focus-visible .poster-description { opacity: 1; transform: translateY(0); } .poster-fallback { display: grid; place-items: center; color: #cdb1f5; background: radial-gradient(circle at 70% 20%, #604085, #21192d 55%, #13151b); } .poster-fallback span { font: 800 72px Manrope, sans-serif; opacity: .7; } -.rating { position: absolute; z-index: 3; right: 8px; bottom: 8px; padding: 5px 7px; border-radius: 7px; color: #fff; background: rgb(7 8 12 / 78%); font-size: 11px; font-weight: 700; backdrop-filter: blur(8px); transition: opacity 120ms ease; } +.rating { position: absolute; z-index: 3; right: 8px; bottom: 8px; padding: 5px 7px; border: 1px solid rgb(244 201 93 / 28%); border-radius: 7px; color: #f4c95d; background: rgb(7 8 12 / 82%); font-size: 11px; font-weight: 800; backdrop-filter: blur(8px); transition: opacity 120ms ease; } +.rating.unrated { border-color: rgb(255 255 255 / 12%); color: #a3a5ad; font-weight: 700; } .poster-frame:hover .rating, .poster-frame:focus-visible .rating { opacity: 0; } .card-copy { position: relative; z-index: 1; padding: 13px 2px 0; } .card-copy h3 { margin: 0; overflow: hidden; color: #f0f0ef; font: 700 15px/1.35 Manrope, sans-serif; text-overflow: ellipsis; white-space: nowrap; } .card-copy p { margin: 5px 0 0; color: #7f818b; font-size: 11px; } .card-copy .card-genres { overflow: hidden; color: #aaa0b7; text-overflow: ellipsis; white-space: nowrap; } -.provider-badges { display: flex; align-items: center; gap: 5px; margin-top: 7px; overflow: hidden; } -.provider-badge { display: flex; min-width: 0; align-items: center; gap: 4px; padding: 3px 6px 3px 3px; border: 1px solid #2b2d35; border-radius: 6px; overflow: hidden; color: #a8a9b0; background: #111319; font-size: 9px; text-overflow: ellipsis; white-space: nowrap; } +.card-copy .lifecycle-date { color: #c794ff; font-weight: 700; } +.provider-badges { display: flex; align-items: center; gap: 6px; margin-top: 8px; overflow: visible; flex-wrap: wrap; } +.provider-badge { position: relative; display: grid; width: 36px; height: 36px; flex: 0 0 36px; padding: 5px; place-items: center; border: 1px solid #2b2d35; border-radius: 8px; overflow: visible; color: #a8a9b0; background: #111319; } .provider-badge.linked { border-color: #46345d; color: #d8c5ef; text-decoration: none; cursor: pointer; } .provider-badge.linked:hover, .provider-badge.linked:focus-visible { border-color: #9664d2; color: #fff; outline: none; background: #21172c; } -.provider-badge.unavailable { opacity: .7; cursor: help; } -.provider-badge .provider-logo-image, .provider-badge .provider-logo-fallback { width: 18px; height: 18px; flex: 0 0 18px; border-radius: 4px; } -.provider-more { flex: 0 0 auto; color: #a481d3; font-size: 10px; font-weight: 700; } +.provider-badge.unavailable { opacity: .75; cursor: help; } +.provider-badge::before { position: absolute; z-index: 12; bottom: calc(100% + 9px); left: 0; width: max-content; max-width: 220px; padding: 8px 10px; border: 1px solid #57406f; border-radius: 8px; color: #eee5f8; background: #17111f; box-shadow: 0 10px 28px rgb(0 0 0 / 50%); content: attr(data-tooltip); font-size: 10px; font-weight: 500; line-height: 1.4; white-space: normal; opacity: 0; transform: translateY(4px); transition: opacity 140ms ease, transform 140ms ease; pointer-events: none; } +.provider-badge:hover::before, .provider-badge:focus-visible::before { opacity: 1; transform: translateY(0); } +.provider-badge.unavailable:focus-visible { border-color: #9664d2; outline: 2px solid rgb(150 100 210 / 35%); outline-offset: 2px; } +.provider-badge .provider-logo-image, .provider-badge .provider-logo-fallback { width: 24px; height: 24px; flex: 0 0 24px; border-radius: 5px; } .skeleton-row { overflow: hidden; } .card-skeleton { flex: 0 0 190px; height: 285px; border-radius: 12px; background: linear-gradient(105deg, #15171e 35%, #232630 48%, #15171e 62%); background-size: 220% 100%; animation: shimmer 1.4s infinite linear; } .rail-message { display: flex; min-height: 180px; flex-direction: column; align-items: flex-start; justify-content: center; padding: 30px; border: 1px dashed #30323d; border-radius: 14px; color: #a0a2aa; background: #101219; } @@ -102,9 +107,108 @@ h2 { margin: 0; font: 700 22px Manrope, sans-serif; } .rail-message span { color: #777983; font-size: 13px; } .rail-message button { margin-top: 12px; padding: 8px 13px; border: 1px solid #68478e; border-radius: 8px; color: #e7d9f8; background: #271a37; cursor: pointer; } .coming-next { margin: 18px 4px 60px; color: #60626c; font-size: 13px; } +.credits { display: flex; max-width: 760px; align-items: center; gap: 18px; margin: 22px auto 34px; padding: 20px 22px; border-top: 1px solid #22242c; color: #747680; font-size: 10px; line-height: 1.55; } +.credits p { margin: 0; } +.credits p + p { margin-top: 4px; } +.credits a { color: #9d83bf; } +.credits a:hover, .credits a:focus-visible { color: #c5a2f1; } +.tmdb-credit { display: block; flex: 0 0 54px; } +.tmdb-credit img { display: block; width: 54px; height: auto; } @keyframes breathe { to { opacity: .3; } } @keyframes shimmer { to { background-position-x: -220%; } } @media (max-width: 900px) { .filter-grid { grid-template-columns: repeat(2, 1fr); } } @media (max-width: 640px) { main { width: min(100% - 28px, 1120px); } .nav { padding: 24px 0; } .freshness { max-width: 150px; text-align: right; } .hero { padding: 5vh 0 38px; } .discovery-setup { padding: 20px 0; } .global-filters { padding: 20px; } .region-field, .filter-grid { grid-template-columns: 1fr; gap: 8px; } .provider-button { width: 100%; } .year-filter { grid-column: auto; } .title-row { min-height: 355px; } .title-card { flex-basis: 165px; } .card-skeleton { flex-basis: 165px; height: 248px; } .rank { min-width: 40px; height: 40px; font-size: 22px; } } @media (hover: none) { .poster-description { min-height: 32%; padding: 38px 12px 12px; opacity: 1; transform: none; } .poster-description > span { -webkit-line-clamp: 2; } .rating { bottom: auto; top: 8px; } } @media (prefers-reduced-motion: reduce) { .status-dot.loading { animation: none; } } + +/* v0.5 content-first workspace */ +.app-layout { display: grid; grid-template-columns: 250px minmax(0, 1fr); align-items: start; gap: 48px; } +.discovery-sidebar { position: sticky; top: 20px; max-height: calc(100vh - 40px); overflow-y: auto; border: 1px solid #30323c; border-radius: 14px; background: rgb(12 14 20 / 88%); scrollbar-width: thin; scrollbar-color: #44344f transparent; } +.content-workspace { min-width: 0; } +.content-workspace .hero { max-width: 760px; padding: 3vh 0 20px; } +.content-workspace .hero h1 { font-size: clamp(48px, 6vw, 76px); } +.content-workspace .intro { max-width: 580px; margin-top: 20px; font-size: 16px; } +.title-search { position: relative; z-index: 30; max-width: 620px; margin: 12px 0 28px; } +.title-search > label { display: block; margin-bottom: 8px; color: #a7a0b1; font-size: 11px; font-weight: 700; letter-spacing: .08em; text-transform: uppercase; } +.search-input-wrap { position: relative; display: flex; align-items: center; } +.search-input-wrap > span:first-child { position: absolute; left: 15px; color: #ad70f4; font-size: 24px; transform: rotate(-20deg); pointer-events: none; } +.search-input-wrap input { width: 100%; min-height: 52px; padding: 12px 48px; border: 1px solid #3a3045; border-radius: 12px; outline: none; color: #f4f1f6; background: rgb(16 17 24 / 92%); box-shadow: 0 14px 38px rgb(0 0 0 / 18%); font: 500 15px "DM Sans", sans-serif; } +.search-input-wrap input:focus { border-color: #9360cf; box-shadow: 0 0 0 3px rgb(147 96 207 / 14%); } +.search-input-wrap input::placeholder { color: #6f7079; } +.search-scope { margin: 8px 2px 0; overflow: hidden; color: #747680; font-size: 10px; line-height: 1.4; text-overflow: ellipsis; white-space: nowrap; } +.search-scope strong { color: #a889c7; font-weight: 700; } +.search-spinner { position: absolute; right: 16px; width: 16px; height: 16px; border: 2px solid #4b3d59; border-top-color: #b778ff; border-radius: 50%; animation: search-spin .7s linear infinite; } +.search-results { position: absolute; top: calc(100% + 8px); right: 0; left: 0; overflow: hidden; border: 1px solid #3b3047; border-radius: 12px; background: #111219; box-shadow: 0 22px 55px rgb(0 0 0 / 58%); } +.search-results > p { margin: 0; padding: 18px; color: #93949d; font-size: 13px; } +.search-results > a { display: grid; grid-template-columns: 42px 1fr auto; min-height: 64px; align-items: center; gap: 12px; padding: 8px 14px; border-bottom: 1px solid #24252d; color: #dedde1; text-decoration: none; } +.search-results > a:last-child { border-bottom: 0; } +.search-results > a:hover, .search-results > a:focus-visible { outline: none; background: #21172d; } +.search-results img, .search-poster-fallback { display: grid; width: 38px; height: 52px; object-fit: cover; place-items: center; border-radius: 5px; color: #d9bcf7; background: #352347; font-weight: 800; } +.search-result-copy { display: grid; min-width: 0; gap: 2px; } +.search-result-copy strong { overflow: hidden; font: 700 13px Manrope, sans-serif; text-overflow: ellipsis; white-space: nowrap; } +.search-result-copy small { overflow: hidden; color: #83858e; font-size: 10px; text-overflow: ellipsis; white-space: nowrap; } +.search-rating { display: inline-flex; min-width: 48px; align-items: center; justify-content: flex-end; gap: 4px; color: #f4c95d; font-size: 12px; font-weight: 800; } +.search-rating > span { font-size: 13px; } +.search-rating.unrated { color: #8b8d96; font-weight: 700; } +@keyframes search-spin { to { transform: rotate(360deg); } } +.discovery-sidebar .discovery-setup { margin: 0; padding: 18px 14px; border: 0; } +.discovery-sidebar .setup-heading { display: block; margin-bottom: 18px; } +.discovery-sidebar .setup-heading h2 { font-size: 13px; letter-spacing: .08em; text-transform: uppercase; } +.discovery-sidebar .status-label { margin-bottom: 5px; font-size: 9px; } +.discovery-sidebar .selection-count { display: inline-block; margin-top: 8px; padding: 4px 7px; font-size: 10px; } +.discovery-sidebar .region-field { display: block; margin-bottom: 20px; font-size: 11px; } +.discovery-sidebar .region-field span { display: block; margin-bottom: 8px; } +.discovery-sidebar .region-field select { min-height: 40px; padding: 9px 10px; font-size: 12px; } +.discovery-sidebar .provider-fieldset legend { display: flex; width: 100%; align-items: center; justify-content: space-between; margin-bottom: 9px; font-size: 11px; } +.discovery-sidebar .provider-fieldset legend button { border: 0; color: #bb82ff; background: transparent; font: 600 10px "DM Sans", sans-serif; cursor: pointer; } +.discovery-sidebar .provider-grid { display: grid; gap: 6px; } +.discovery-sidebar .provider-button { width: 100%; min-width: 0; min-height: 39px; padding: 4px 7px; border-radius: 8px; font-size: 11px; } +.discovery-sidebar .provider-mark { width: 27px; height: 27px; border-radius: 6px; } +.discovery-sidebar .global-filters { margin: 0; padding: 18px 14px; border-width: 1px 0 0; border-radius: 0; background: transparent; } +.discovery-sidebar .global-filters .setup-heading { display: flex; align-items: flex-start; margin-bottom: 18px; } +.discovery-sidebar .clear-filters { padding: 2px; color: #b77cff; font-size: 10px; } +.discovery-sidebar .filter-grid { display: grid; grid-template-columns: 1fr; gap: 16px; } +.discovery-sidebar .filter-group { font-size: 11px; } +.discovery-sidebar .filter-group > span, .discovery-sidebar .filter-group > legend { margin-bottom: 7px; } +.discovery-sidebar .filter-group select, .discovery-sidebar .filter-group input { min-height: 39px; padding: 8px 9px; border-radius: 8px; font-size: 12px; } +.discovery-sidebar .segmented-control button { min-height: 31px; font-size: 11px; } +.discovery-sidebar .year-filter { grid-column: auto; } +.discovery-sidebar .genre-filter { margin-top: 18px; } +.discovery-sidebar .genre-filter legend { font-size: 11px; } +.discovery-sidebar .genre-list { gap: 5px; } +.discovery-sidebar .genre-list button { padding: 6px 8px; font-size: 9px; } +.content-workspace .discovery-rail { margin-top: 36px; } +.content-workspace .rail-heading { margin-bottom: 14px; } +.content-workspace .rail-heading h2 { font-size: 24px; } +.content-workspace .rail-heading .status-label { margin-bottom: 4px; font-size: 9px; } +.content-workspace .title-row { min-height: 390px; align-items: stretch; gap: 12px; } +.content-workspace .title-card { display: flex; flex-basis: 158px; flex-direction: column; } +.content-workspace .poster-frame { flex: 0 0 auto; } +.content-workspace .rank { top: 6px; left: 6px; min-width: 38px; height: 38px; border-radius: 8px; font-size: 20px; } +.content-workspace .card-copy { display: flex; height: 100%; padding-top: 9px; flex-direction: column; } +.content-workspace .card-copy h3 { display: -webkit-box; min-height: 35px; overflow: hidden; font-size: 13px; line-height: 1.35; white-space: normal; -webkit-box-orient: vertical; -webkit-line-clamp: 2; } +.content-workspace .provider-badges { margin-top: auto; padding-top: 9px; } +.content-workspace .card-skeleton { flex-basis: 158px; height: 237px; } + +@media (max-width: 980px) { + .app-layout { grid-template-columns: 220px minmax(0, 1fr); gap: 28px; } + .content-workspace .title-card { flex-basis: 150px; } +} + +@media (max-width: 760px) { + main { width: min(100% - 28px, 1480px); } + .app-layout { display: block; } + .discovery-sidebar { position: static; max-height: none; overflow: visible; } + .content-workspace .hero { padding-top: 44px; } + .content-workspace .hero h1 { font-size: clamp(44px, 14vw, 66px); } + .discovery-sidebar .provider-grid { grid-template-columns: repeat(2, minmax(0, 1fr)); } + .discovery-sidebar .filter-grid { grid-template-columns: repeat(2, minmax(0, 1fr)); } + .discovery-sidebar .year-filter { grid-column: span 2; } +} + +@media (max-width: 480px) { + .discovery-sidebar .provider-grid, .discovery-sidebar .filter-grid { grid-template-columns: 1fr; } + .discovery-sidebar .year-filter { grid-column: auto; } + .credits { align-items: flex-start; padding-inline: 4px; } + .tmdb-credit, .tmdb-credit img { width: 44px; flex-basis: 44px; } +} diff --git a/ingestion/enrich_catalog.py b/ingestion/enrich_catalog.py new file mode 100644 index 0000000..71926ca --- /dev/null +++ b/ingestion/enrich_catalog.py @@ -0,0 +1,393 @@ +"""Plan and execute bounded TMDB enrichment for the local serving catalog. + +Modes: + +* ``backfill`` selects catalog titles that have never retained the requested + enrichment payloads. It is intended for the initial broad enrichment. +* ``incremental`` selects only new or stale titles, ordered by product value. + +Planning is entirely local. External requests happen only after the plan has +been built and only when ``--dry-run`` is not supplied. +""" + +from __future__ import annotations + +import argparse +import json +import logging +import time +import uuid +from dataclasses import asdict, dataclass +from datetime import UTC, datetime, timedelta +from pathlib import Path +from typing import Literal + +import duckdb +from dotenv import load_dotenv + +from ingestion.core.lake import RawRecord, write_raw_batch +from ingestion.sources.tmdb.client import TMDBSource +from watchpulse.config import Settings +from watchpulse.pipeline_runs import PipelineRunRepository, sanitize_error + +logger = logging.getLogger("ingestion.enrich_catalog") +EnrichmentMode = Literal["backfill", "incremental"] +BATCH_SIZE = 25 + + +@dataclass(frozen=True) +class EnrichmentCandidate: + content_type: str + tmdb_id: int + title: str + priority: float + reasons: tuple[str, ...] + metadata_due: bool + providers_due: bool + + +def _latest_observations(lake_root: Path, endpoint: str) -> dict[tuple[str, int], datetime]: + files = sorted( + lake_root.glob( + f"raw/source=tmdb/endpoint={endpoint}/entity_type=*/country=ALL/date=*/*.parquet" + ) + ) + if not files: + return {} + with duckdb.connect() as connection: + rows = connection.execute( + """ + select + json_extract_string(cast(request_params as json), '$.entity_type'), + try_cast(json_extract_string(cast(request_params as json), '$.tmdb_id') as bigint), + max(cast(fetched_at as timestamptz)) + from read_parquet(?, union_by_name = true) + group by 1, 2 + """, + [[str(path) for path in files]], + ).fetchall() + return { + (str(content_type), int(tmdb_id)): fetched_at + for content_type, tmdb_id, fetched_at in rows + if content_type in {"movie", "tv"} and tmdb_id is not None + } + + +def _catalog_rows(database_path: Path) -> list[tuple]: + if not database_path.is_file(): + raise FileNotFoundError(f"Serving catalog does not exist: {database_path}") + with duckdb.connect(str(database_path), read_only=True) as connection: + return connection.execute( + """ + with ranked as ( + select + *, + row_number() over ( + partition by region, provider_key, content_type + order by popularity_score desc nulls last, tmdb_id + ) as provider_rank + from main_marts.catalog_availability + where is_available or is_upcoming + ) + select + content_type, + tmdb_id, + max(title) as title, + max(release_date) as release_date, + max(popularity_score) as popularity_score, + bool_or(is_upcoming) as is_upcoming, + bool_or(is_available and provider_rank <= 10) as is_top_ten, + max(available_since) as available_since, + bool_or(poster_path is null or overview is null or release_date is null) + as missing_core_metadata, + bool_or(runtime_minutes is null) as missing_runtime, + bool_or( + content_type = 'tv' + and (season_count is null or episode_count is null) + ) as missing_series_totals + from ranked + group by content_type, tmdb_id + """ + ).fetchall() + + +def _is_due(last_seen: datetime | None, *, as_of: datetime, days: int) -> bool: + if last_seen is None: + return True + if last_seen.tzinfo is None: + last_seen = last_seen.replace(tzinfo=UTC) + return last_seen <= as_of - timedelta(days=days) + + +def plan_enrichment( + *, + database_path: Path, + lake_root: Path, + mode: EnrichmentMode = "incremental", + max_titles: int = 250, + include_providers: bool = True, + as_of: datetime | None = None, + upcoming_refresh_days: int = 7, + recent_refresh_days: int = 30, + series_refresh_days: int = 90, + movie_refresh_days: int = 180, + provider_refresh_days: int = 60, +) -> tuple[EnrichmentCandidate, ...]: + """Return a capped, deterministic enrichment plan without external calls.""" + if mode not in {"backfill", "incremental"}: + raise ValueError("mode must be backfill or incremental") + if max_titles <= 0: + raise ValueError("max_titles must be greater than zero") + now = as_of or datetime.now(UTC) + metadata_seen = _latest_observations(lake_root, "metadata") + providers_seen = _latest_observations(lake_root, "watch_providers") + candidates: list[EnrichmentCandidate] = [] + + for row in _catalog_rows(database_path): + ( + content_type, + tmdb_id, + title, + release_date, + popularity, + is_upcoming, + is_top_ten, + available_since, + missing_core, + missing_runtime, + missing_series_totals, + ) = row + key = (str(content_type), int(tmdb_id)) + recent_release = release_date is not None and release_date >= now.date() - timedelta( + days=365 + ) + recently_added = available_since is not None and available_since >= now - timedelta(days=30) + metadata_days = ( + upcoming_refresh_days + if is_upcoming + else recent_refresh_days + if recent_release + else series_refresh_days + if content_type == "tv" + else movie_refresh_days + ) + provider_days = upcoming_refresh_days if is_upcoming else provider_refresh_days + if mode == "backfill": + metadata_due = key not in metadata_seen + providers_due = include_providers and key not in providers_seen + else: + metadata_due = _is_due(metadata_seen.get(key), as_of=now, days=metadata_days) + providers_due = include_providers and _is_due( + providers_seen.get(key), as_of=now, days=provider_days + ) + if not metadata_due and not providers_due: + continue + + reasons: list[str] = [] + priority = min(float(popularity or 0), 200.0) + if is_upcoming: + priority += 1000 + reasons.append("upcoming") + if recently_added: + priority += 900 + reasons.append("recently_added") + if recent_release: + priority += 800 + reasons.append("recent_release") + if is_top_ten: + priority += 700 + reasons.append("top_10") + if missing_core: + priority += 500 + reasons.append("missing_core_metadata") + if missing_runtime or missing_series_totals: + priority += 300 + reasons.append("missing_details") + if not reasons: + reasons.append("new_or_stale") + candidates.append( + EnrichmentCandidate( + content_type=str(content_type), + tmdb_id=int(tmdb_id), + title=str(title), + priority=round(priority, 4), + reasons=tuple(reasons), + metadata_due=metadata_due, + providers_due=providers_due, + ) + ) + + candidates.sort(key=lambda item: (-item.priority, item.content_type, item.tmdb_id)) + return tuple(candidates[:max_titles]) + + +def _write_batch( + records: list[RawRecord], + *, + lake_root: Path, + endpoint: str, + run_id: str, +) -> int: + if not records: + return 0 + record_count = len(records) + write_raw_batch( + records, + lake_root=lake_root, + source="tmdb", + endpoint=endpoint, + entity_type="mixed", + country="ALL", + run_id=run_id, + ) + records.clear() + return record_count + + +def execute_enrichment( + plan: tuple[EnrichmentCandidate, ...], + *, + lake_root: Path, + api_key: str, + mode: EnrichmentMode, + tmdb_base_url: str | None = None, + database_path: Path | None = None, +) -> dict[str, object]: + """Execute a previously built plan and retain completed batches for resumption.""" + run_id = uuid.uuid4().hex[:12] + started_at = time.monotonic() + options = {"base_url": tmdb_base_url} if tmdb_base_url else {} + source = TMDBSource(api_key, **options) + runs = PipelineRunRepository(database_path) if database_path else None + metadata_records: list[RawRecord] = [] + provider_records: list[RawRecord] = [] + metadata_written = 0 + providers_written = 0 + if runs: + runs.start(run_id=run_id, job_name="tmdb_catalog_enrichment", source="tmdb") + try: + with source: + for candidate in plan: + if candidate.metadata_due: + metadata_records.append( + source.fetch_metadata( + entity_type=candidate.content_type, + source_title_id=candidate.tmdb_id, + ) + ) + if len(metadata_records) >= BATCH_SIZE: + metadata_written += _write_batch( + metadata_records, + lake_root=lake_root, + endpoint="metadata", + run_id=run_id, + ) + if candidate.providers_due: + provider_records.append( + source.fetch_availability( + entity_type=candidate.content_type, + source_title_id=candidate.tmdb_id, + ) + ) + if len(provider_records) >= BATCH_SIZE: + providers_written += _write_batch( + provider_records, + lake_root=lake_root, + endpoint="watch_providers", + run_id=run_id, + ) + metadata_written += _write_batch( + metadata_records, lake_root=lake_root, endpoint="metadata", run_id=run_id + ) + providers_written += _write_batch( + provider_records, + lake_root=lake_root, + endpoint="watch_providers", + run_id=run_id, + ) + summary: dict[str, object] = { + "run_id": run_id, + "mode": mode, + "titles_planned": len(plan), + "metadata_records_written": metadata_written, + "provider_records_written": providers_written, + "api_request_count": source.request_count, + "duration_seconds": round(time.monotonic() - started_at, 1), + } + if runs: + runs.succeed( + run_id=run_id, + api_request_count=source.request_count, + rows_fetched=metadata_written + providers_written, + rows_inserted=metadata_written + providers_written, + details=summary, + ) + return summary + except Exception as error: + if runs: + runs.fail( + run_id=run_id, + api_request_count=source.request_count, + error_message=sanitize_error(error, secrets=(api_key,)), + ) + raise + + +def main() -> None: + logging.basicConfig(level=logging.INFO, format="%(asctime)s %(levelname)s %(message)s") + logging.getLogger("httpx").setLevel(logging.WARNING) + logging.getLogger("httpcore").setLevel(logging.WARNING) + load_dotenv() + settings = Settings.from_env() + parser = argparse.ArgumentParser(description=__doc__) + parser.add_argument("--mode", choices=("backfill", "incremental"), default="incremental") + parser.add_argument( + "--max-titles", type=int, default=settings.tmdb_enrichment_max_titles_per_run + ) + parser.add_argument("--metadata-only", action="store_true") + parser.add_argument("--dry-run", action="store_true") + parser.add_argument("--plan-output", type=Path) + parser.add_argument("--lake-root", type=Path, default=settings.lake_root) + parser.add_argument("--serving-db", type=Path, default=settings.serving_database_path) + args = parser.parse_args() + plan = plan_enrichment( + database_path=args.serving_db, + lake_root=args.lake_root, + mode=args.mode, + max_titles=args.max_titles, + include_providers=not args.metadata_only, + upcoming_refresh_days=settings.tmdb_enrichment_upcoming_refresh_days, + recent_refresh_days=settings.tmdb_enrichment_recent_refresh_days, + series_refresh_days=settings.tmdb_enrichment_series_refresh_days, + movie_refresh_days=settings.tmdb_enrichment_movie_refresh_days, + provider_refresh_days=settings.tmdb_enrichment_provider_refresh_days, + ) + plan_payload = [asdict(candidate) for candidate in plan] + if args.plan_output: + args.plan_output.write_text(json.dumps(plan_payload, indent=2) + "\n", encoding="utf-8") + logger.info( + "planned titles=%d metadata_requests=%d provider_requests=%d mode=%s", + len(plan), + sum(item.metadata_due for item in plan), + sum(item.providers_due for item in plan), + args.mode, + ) + if args.dry_run: + for candidate in plan[:20]: + logger.info("candidate: %s", asdict(candidate)) + return + if not settings.tmdb_api_key: + raise SystemExit("TMDB_API_KEY is not set") + summary = execute_enrichment( + plan, + lake_root=args.lake_root, + api_key=settings.tmdb_api_key, + mode=args.mode, + tmdb_base_url=settings.tmdb_base_url, + database_path=settings.database_path, + ) + logger.info("enrichment complete: %s", summary) + + +if __name__ == "__main__": + main() diff --git a/ingestion/enrich_streaming_metadata.py b/ingestion/enrich_streaming_metadata.py new file mode 100644 index 0000000..4153187 --- /dev/null +++ b/ingestion/enrich_streaming_metadata.py @@ -0,0 +1,197 @@ +"""Incrementally enrich lifecycle titles with canonical TMDB metadata.""" + +from __future__ import annotations + +import argparse +import logging +import uuid +from pathlib import Path + +import duckdb +from dotenv import load_dotenv + +from ingestion.core.lake import write_raw_batch +from ingestion.sources.tmdb.client import TMDBSource +from watchpulse.config import Settings + +logger = logging.getLogger("ingestion.enrich_streaming_metadata") +BATCH_SIZE = 10 + + +def lifecycle_title_refs( + lake_root: Path, + *, + event_type: str = "upcoming", + country: str | None = None, +) -> tuple[tuple[str, int], ...]: + country_pattern = country.upper() if country else "*" + event_files = sorted( + lake_root.glob( + "raw/source=streaming_availability/" + f"endpoint=changes_{event_type}/entity_type=show/country={country_pattern}/" + "date=*/*.parquet" + ) + ) + if not event_files: + return () + + with duckdb.connect() as connection: + lifecycle_rows = connection.execute( + """ + with raw as ( + select cast(payload as json) as payload + from read_parquet(?, union_by_name = true) + ) + select distinct + split_part(json_extract_string(show_row.value, '$.tmdbId'), '/', 1) as content_type, + try_cast( + split_part(json_extract_string(show_row.value, '$.tmdbId'), '/', 2) + as bigint + ) as tmdb_id + from raw, json_each(payload, '$.shows') as show_row + """, + [[str(path) for path in event_files]], + ).fetchall() + return tuple( + sorted( + (str(content_type), int(tmdb_id)) + for content_type, tmdb_id in lifecycle_rows + if content_type in {"movie", "tv"} and tmdb_id is not None + ) + ) + + +def _retained_refs(lake_root: Path, endpoint: str) -> set[tuple[str, int]]: + files = sorted( + lake_root.glob( + f"raw/source=tmdb/endpoint={endpoint}/entity_type=*/country=ALL/date=*/*.parquet" + ) + ) + if not files: + return set() + with duckdb.connect() as connection: + existing_rows = connection.execute( + """ + select distinct + json_extract_string(cast(request_params as json), '$.entity_type'), + cast(json_extract_string(cast(request_params as json), '$.tmdb_id') as bigint) + from read_parquet(?, union_by_name = true) + """, + [[str(path) for path in files]], + ).fetchall() + return {(str(content_type), int(tmdb_id)) for content_type, tmdb_id in existing_rows} + + +def pending_title_refs( + lake_root: Path, + *, + event_type: str = "upcoming", + country: str | None = None, + endpoint: str = "metadata", +) -> tuple[tuple[str, int], ...]: + refs = set(lifecycle_title_refs(lake_root, event_type=event_type, country=country)) + return tuple(sorted(refs - _retained_refs(lake_root, endpoint))) + + +def run( + *, + lake_root: Path, + api_key: str, + event_type: str = "upcoming", + country: str | None = None, + max_titles: int | None = None, + tmdb_base_url: str | None = None, +) -> dict[str, object]: + metadata_refs = pending_title_refs( + lake_root, event_type=event_type, country=country, endpoint="metadata" + ) + availability_refs = pending_title_refs( + lake_root, event_type=event_type, country=country, endpoint="watch_providers" + ) + if max_titles is not None: + allowed = set( + lifecycle_title_refs(lake_root, event_type=event_type, country=country)[:max_titles] + ) + metadata_refs = tuple(ref for ref in metadata_refs if ref in allowed) + availability_refs = tuple(ref for ref in availability_refs if ref in allowed) + source_options = {"base_url": tmdb_base_url} if tmdb_base_url else {} + source = TMDBSource(api_key, **source_options) + run_id = uuid.uuid4().hex[:12] + metadata_records = [] + availability_records = [] + metadata_written = 0 + availability_written = 0 + with source: + for content_type, tmdb_id in metadata_refs: + metadata_records.append( + source.fetch_metadata(entity_type=content_type, source_title_id=tmdb_id) + ) + for content_type, tmdb_id in availability_refs: + availability_records.append( + source.fetch_availability(entity_type=content_type, source_title_id=tmdb_id) + ) + for start in range(0, len(metadata_records), BATCH_SIZE): + batch = metadata_records[start : start + BATCH_SIZE] + write_raw_batch( + batch, + lake_root=lake_root, + source="tmdb", + endpoint="metadata", + entity_type="mixed", + country="ALL", + run_id=run_id, + ) + metadata_written += len(batch) + for start in range(0, len(availability_records), BATCH_SIZE): + batch = availability_records[start : start + BATCH_SIZE] + write_raw_batch( + batch, + lake_root=lake_root, + source="tmdb", + endpoint="watch_providers", + entity_type="mixed", + country="ALL", + run_id=run_id, + ) + availability_written += len(batch) + return { + "run_id": run_id, + "event_type": event_type, + "country": country.upper() if country else None, + "metadata_titles_requested": len(metadata_refs), + "availability_titles_requested": len(availability_refs), + "metadata_records_written": metadata_written, + "availability_records_written": availability_written, + "api_request_count": source.request_count, + } + + +def main() -> None: + logging.basicConfig(level=logging.INFO, format="%(asctime)s %(levelname)s %(message)s") + logging.getLogger("httpx").setLevel(logging.WARNING) + logging.getLogger("httpcore").setLevel(logging.WARNING) + load_dotenv() + settings = Settings.from_env() + parser = argparse.ArgumentParser(description=__doc__) + parser.add_argument("--event-type", choices=("new", "upcoming"), default="upcoming") + parser.add_argument("--country") + parser.add_argument("--max-titles", type=int) + parser.add_argument("--lake-root", default=str(settings.lake_root)) + args = parser.parse_args() + if args.max_titles is not None and args.max_titles <= 0: + raise SystemExit("--max-titles must be greater than zero") + if not settings.tmdb_api_key: + raise SystemExit("TMDB_API_KEY is not set") + summary = run( + lake_root=Path(args.lake_root), + api_key=settings.tmdb_api_key, + event_type=args.event_type, + country=args.country, + max_titles=args.max_titles, + tmdb_base_url=settings.tmdb_base_url, + ) + logger.info("enrichment complete: %s", summary) + + +if __name__ == "__main__": + main() diff --git a/ingestion/run.py b/ingestion/run.py index 1cbc191..46d6a50 100644 --- a/ingestion/run.py +++ b/ingestion/run.py @@ -217,6 +217,8 @@ def run( def main() -> None: logging.basicConfig(level=logging.INFO, format="%(asctime)s %(levelname)s %(message)s") + logging.getLogger("httpx").setLevel(logging.WARNING) + logging.getLogger("httpcore").setLevel(logging.WARNING) load_dotenv() settings = Settings.from_env() diff --git a/ingestion/tests/test_catalog_enrichment.py b/ingestion/tests/test_catalog_enrichment.py new file mode 100644 index 0000000..9f266ab --- /dev/null +++ b/ingestion/tests/test_catalog_enrichment.py @@ -0,0 +1,193 @@ +from datetime import UTC, date, datetime, timedelta +from pathlib import Path + +import duckdb + +from ingestion.core.lake import RawRecord, write_raw_batch +from ingestion.enrich_catalog import execute_enrichment, plan_enrichment + + +def _catalog(path: Path, now: datetime) -> None: + with duckdb.connect(str(path)) as connection: + connection.execute("create schema main_marts") + connection.execute( + """ + create table main_marts.catalog_availability ( + tmdb_id bigint, content_type varchar, title varchar, + release_date date, popularity_score double, region varchar, + provider_key varchar, is_available boolean, is_upcoming boolean, + available_since timestamptz, poster_path varchar, overview varchar, + runtime_minutes integer, season_count integer, episode_count integer + ) + """ + ) + connection.executemany( + """ + insert into main_marts.catalog_availability + values (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) + """, + [ + ( + 1, + "tv", + "Upcoming", + now.date(), + 10.0, + "GR", + "netflix", + False, + True, + None, + "/1.jpg", + "One", + None, + None, + None, + ), + ( + 2, + "movie", + "Popular", + date(2000, 1, 1), + 100.0, + "GR", + "netflix", + True, + False, + None, + "/2.jpg", + "Two", + None, + None, + None, + ), + ( + 3, + "movie", + "Retained", + date(2000, 1, 1), + 1.0, + "GR", + "netflix", + True, + False, + None, + "/3.jpg", + "Three", + 90, + None, + None, + ), + ], + ) + + +def _retain(lake_root: Path, endpoint: str, tmdb_id: int, fetched_at: datetime) -> None: + write_raw_batch( + [ + RawRecord( + request_params={"entity_type": "movie", "tmdb_id": tmdb_id}, + payload={"id": tmdb_id}, + fetched_at=fetched_at, + ) + ], + lake_root=lake_root, + source="tmdb", + endpoint=endpoint, + entity_type="mixed", + country="ALL", + run_id=f"{endpoint}-{tmdb_id}", + run_date=fetched_at.date(), + ) + + +def test_backfill_selects_only_titles_without_retained_payloads(tmp_path: Path) -> None: + now = datetime(2026, 8, 25, tzinfo=UTC) + database = tmp_path / "serving.duckdb" + _catalog(database, now) + _retain(tmp_path, "metadata", 3, now) + _retain(tmp_path, "watch_providers", 3, now) + + plan = plan_enrichment( + database_path=database, + lake_root=tmp_path, + mode="backfill", + max_titles=10, + as_of=now, + ) + + assert [candidate.tmdb_id for candidate in plan] == [1, 2] + assert plan[0].reasons[0] == "upcoming" + assert all(candidate.metadata_due and candidate.providers_due for candidate in plan) + + +def test_incremental_prioritizes_product_value_and_respects_freshness(tmp_path: Path) -> None: + now = datetime(2026, 8, 25, tzinfo=UTC) + database = tmp_path / "serving.duckdb" + _catalog(database, now) + _retain(tmp_path, "metadata", 3, now - timedelta(days=10)) + _retain(tmp_path, "watch_providers", 3, now - timedelta(days=10)) + + plan = plan_enrichment( + database_path=database, + lake_root=tmp_path, + mode="incremental", + max_titles=2, + as_of=now, + ) + + assert [candidate.tmdb_id for candidate in plan] == [1, 2] + assert plan[0].priority > plan[1].priority + + +def test_execute_retains_batches_and_counts_requests(tmp_path: Path, monkeypatch) -> None: + now = datetime(2026, 8, 25, tzinfo=UTC) + database = tmp_path / "serving.duckdb" + _catalog(database, now) + plan = plan_enrichment( + database_path=database, + lake_root=tmp_path, + mode="backfill", + max_titles=1, + as_of=now, + ) + + class FakeTMDBSource: + request_count = 0 + + def __init__(self, api_key: str, **_options: str) -> None: + assert api_key == "" + + def __enter__(self): + return self + + def __exit__(self, *_exc_info: object) -> None: + return None + + def fetch_metadata(self, *, entity_type: str, source_title_id: int) -> RawRecord: + self.request_count += 1 + return RawRecord( + request_params={"entity_type": entity_type, "tmdb_id": source_title_id}, + payload={"id": source_title_id}, + ) + + def fetch_availability(self, *, entity_type: str, source_title_id: int) -> RawRecord: + self.request_count += 1 + return RawRecord( + request_params={"entity_type": entity_type, "tmdb_id": source_title_id}, + payload={"id": source_title_id, "results": {}}, + ) + + monkeypatch.setattr("ingestion.enrich_catalog.TMDBSource", FakeTMDBSource) + summary = execute_enrichment( + plan, + lake_root=tmp_path, + api_key="", + mode="backfill", + ) + + assert summary["api_request_count"] == 2 + assert summary["metadata_records_written"] == 1 + assert summary["provider_records_written"] == 1 + assert len(list(tmp_path.glob("raw/source=tmdb/endpoint=metadata/**/*.parquet"))) == 1 + assert len(list(tmp_path.glob("raw/source=tmdb/endpoint=watch_providers/**/*.parquet"))) == 1 diff --git a/ingestion/tests/test_config.py b/ingestion/tests/test_config.py index 4b5ae22..9118906 100644 --- a/ingestion/tests/test_config.py +++ b/ingestion/tests/test_config.py @@ -13,6 +13,8 @@ def test_settings_load_product_defaults() -> None: assert settings.lake_root == Path("data/lake") assert settings.new_release_days == 90 assert settings.streaming_availability_max_requests_per_run == 2 + assert settings.tmdb_enrichment_max_titles_per_run == 250 + assert settings.tmdb_enrichment_movie_refresh_days == 180 def test_default_region_must_be_supported() -> None: diff --git a/ingestion/tests/test_streaming_metadata_enrichment.py b/ingestion/tests/test_streaming_metadata_enrichment.py new file mode 100644 index 0000000..bb181e6 --- /dev/null +++ b/ingestion/tests/test_streaming_metadata_enrichment.py @@ -0,0 +1,109 @@ +from pathlib import Path + +from ingestion.core.lake import RawRecord, write_raw_batch +from ingestion.enrich_streaming_metadata import pending_title_refs, run + + +def _streaming_event(lake_root: Path) -> None: + write_raw_batch( + [ + RawRecord( + request_params={"country": "GR"}, + payload={ + "shows": { + "one": {"tmdbId": "movie/101", "title": "One"}, + "two": {"tmdbId": "tv/202", "title": "Two"}, + } + }, + ) + ], + lake_root=lake_root, + source="streaming_availability", + endpoint="changes_upcoming", + entity_type="show", + country="GR", + run_id="events", + ) + + +def test_pending_title_refs_excludes_retained_tmdb_metadata(tmp_path: Path) -> None: + _streaming_event(tmp_path) + write_raw_batch( + [ + RawRecord( + request_params={"entity_type": "movie", "tmdb_id": 101}, + payload={"id": 101, "title": "One"}, + ) + ], + lake_root=tmp_path, + source="tmdb", + endpoint="metadata", + entity_type="movie", + country="ALL", + run_id="metadata", + ) + + assert pending_title_refs(tmp_path, country="GR") == (("tv", 202),) + + +def test_run_fetches_only_pending_lifecycle_metadata(tmp_path: Path, monkeypatch) -> None: + _streaming_event(tmp_path) + + class FakeTMDBSource: + request_count = 0 + + def __init__(self, api_key: str, **options: str) -> None: + assert api_key == "" + + def __enter__(self): + return self + + def __exit__(self, *exc_info: object) -> None: + return None + + def fetch_metadata(self, *, entity_type: str, source_title_id: int) -> RawRecord: + self.request_count += 1 + return RawRecord( + request_params={"entity_type": entity_type, "tmdb_id": source_title_id}, + payload={"id": source_title_id, "title": str(source_title_id)}, + ) + + def fetch_availability(self, *, entity_type: str, source_title_id: int) -> RawRecord: + self.request_count += 1 + return RawRecord( + request_params={"entity_type": entity_type, "tmdb_id": source_title_id}, + payload={"id": source_title_id, "results": {}}, + ) + + monkeypatch.setattr( + "ingestion.enrich_streaming_metadata.TMDBSource", + FakeTMDBSource, + ) + + summary = run(lake_root=tmp_path, api_key="", country="GR", max_titles=1) + + assert summary["metadata_titles_requested"] == 1 + assert summary["availability_titles_requested"] == 1 + assert summary["metadata_records_written"] == 1 + assert summary["availability_records_written"] == 1 + assert summary["api_request_count"] == 2 + assert ( + len( + list( + tmp_path.glob( + "raw/source=tmdb/endpoint=metadata/entity_type=mixed/country=ALL/date=*/*.parquet" + ) + ) + ) + == 1 + ) + assert ( + len( + list( + tmp_path.glob( + "raw/source=tmdb/endpoint=watch_providers/entity_type=mixed/country=ALL/date=*/*.parquet" + ) + ) + ) + == 1 + ) diff --git a/tests/test_warehouse_publish.py b/tests/test_warehouse_publish.py index 64887ec..ede5173 100644 --- a/tests/test_warehouse_publish.py +++ b/tests/test_warehouse_publish.py @@ -62,6 +62,15 @@ def runner(command: tuple[str, ...], cwd: Path, environment: dict[str, str]) -> ).fetchone() == (2,) +def test_validation_allows_current_titles_with_upcoming_seasons(tmp_path: Path) -> None: + candidate_path = tmp_path / "dual-state.duckdb" + _create_candidate(candidate_path) + with duckdb.connect(str(candidate_path)) as connection: + connection.execute("update main_marts.catalog_freshness set upcoming_row_count = 1") + + validate_candidate(candidate_path) + + def test_failed_build_preserves_previous_output(tmp_path: Path) -> None: project_dir = tmp_path / "warehouse" project_dir.mkdir() diff --git a/warehouse/docs/staging/columns.md b/warehouse/docs/staging/columns.md index dd61d5e..fd823c1 100644 --- a/warehouse/docs/staging/columns.md +++ b/warehouse/docs/staging/columns.md @@ -2,6 +2,16 @@ Canonical TMDB content identifier. {% enddocs %} +{% docs col_episode_count %} +Total number of aired episodes reported by TMDB for a TV series. This does not +describe the episode count of an announced future season. +{% enddocs %} + +{% docs col_season_count %} +Total number of seasons reported by TMDB for a TV series. This is current title +metadata and must not be interpreted as the number of an announced new season. +{% enddocs %} + {% docs col_content_type %} Internal content type: `movie` or `tv`. {% enddocs %} diff --git a/warehouse/docs/staging/models.md b/warehouse/docs/staging/models.md index 3cb9a57..8559eb7 100644 --- a/warehouse/docs/staging/models.md +++ b/warehouse/docs/staging/models.md @@ -58,3 +58,11 @@ source. Source ratings and signed image URLs are intentionally excluded. Grain: one row per `tmdb_id + content_type`. {% enddocs %} +{% docs stg_tmdb_metadata %} +Canonical per-title TMDB metadata retained by targeted enrichment. This model +supplies poster artwork, descriptions, runtime, genres, ratings, and release +details for lifecycle titles that may not appear in provider discovery pages. +{% enddocs %} +{% docs stg_tmdb_watch_providers %} +Canonical subscription availability extracted from targeted TMDB watch-provider responses for lifecycle titles. It provides current-state evidence without placing TMDB in the user request path. +{% enddocs %} diff --git a/warehouse/models/intermediate/_intermediate.yml b/warehouse/models/intermediate/_intermediate.yml index 5f7b3c4..4fb1af1 100644 --- a/warehouse/models/intermediate/_intermediate.yml +++ b/warehouse/models/intermediate/_intermediate.yml @@ -53,6 +53,20 @@ models: arguments: min_value: 1 max_value: 1440 + - name: episode_count + description: "{{ doc('col_episode_count') }}" + data_type: integer + tests: + - dbt_expectations.expect_column_values_to_be_between: + arguments: + min_value: 1 + - name: season_count + description: "{{ doc('col_season_count') }}" + data_type: integer + tests: + - dbt_expectations.expect_column_values_to_be_between: + arguments: + min_value: 1 - name: genre_ids description: "{{ doc('col_genre_ids') }}" data_type: json @@ -186,7 +200,7 @@ models: [tmdb_id, content_type, region, provider_key, monetization_type] - dbt_utils.expression_is_true: arguments: - expression: "is_available and not is_upcoming" + expression: "is_available" columns: &availability_columns - name: tmdb_id description: "{{ doc('col_tmdb_id') }}" @@ -250,5 +264,5 @@ models: [tmdb_id, content_type, region, provider_key, monetization_type] - dbt_utils.expression_is_true: arguments: - expression: "not is_available and is_upcoming" + expression: "is_upcoming" columns: *availability_columns diff --git a/warehouse/models/intermediate/int_content.sql b/warehouse/models/intermediate/int_content.sql index 50c0a84..92ffc1d 100644 --- a/warehouse/models/intermediate/int_content.sql +++ b/warehouse/models/intermediate/int_content.sql @@ -1,4 +1,4 @@ -with tmdb_candidates as ( +with tmdb_metadata_candidates as ( select tmdb_id, content_type, @@ -7,7 +7,9 @@ with tmdb_candidates as ( overview, release_date, year(release_date)::integer as release_year, - cast(null as integer) as runtime_minutes, + runtime_minutes, + episode_count, + season_count, genre_ids, original_language, tmdb_rating, @@ -19,6 +21,32 @@ with tmdb_candidates as ( source_file, 'tmdb' as metadata_source, 1 as source_priority + from {{ ref('stg_tmdb_metadata') }} +), + +tmdb_candidates as ( + select + tmdb_id, + content_type, + title, + original_title, + overview, + release_date, + year(release_date)::integer as release_year, + cast(null as integer) as runtime_minutes, + cast(null as integer) as episode_count, + cast(null as integer) as season_count, + genre_ids, + original_language, + tmdb_rating, + vote_count, + tmdb_popularity, + poster_path, + backdrop_path, + source_updated_at, + source_file, + 'tmdb' as metadata_source, + 2 as source_priority from {{ ref('stg_tmdb_discovery') }} ), @@ -32,6 +60,8 @@ streaming_candidates as ( cast(null as date) as release_date, release_year, runtime_minutes, + cast(null as integer) as episode_count, + cast(null as integer) as season_count, cast(null as json) as genre_ids, cast(null as varchar) as original_language, cast(null as double) as tmdb_rating, @@ -42,11 +72,13 @@ streaming_candidates as ( source_updated_at, source_file, 'streaming_availability' as metadata_source, - 2 as source_priority + 3 as source_priority from {{ ref('stg_streaming_shows') }} ), candidates as ( + select * from tmdb_metadata_candidates + union all select * from tmdb_candidates union all select * from streaming_candidates @@ -77,6 +109,8 @@ select release_date, release_year, runtime_minutes, + episode_count, + season_count, genre_ids, original_language, tmdb_rating, diff --git a/warehouse/models/intermediate/int_current_availability.sql b/warehouse/models/intermediate/int_current_availability.sql index 5961d93..05f8e11 100644 --- a/warehouse/models/intermediate/int_current_availability.sql +++ b/warehouse/models/intermediate/int_current_availability.sql @@ -11,29 +11,64 @@ with additions as ( from {{ ref('int_streaming_events') }} where event_type = 'new' group by 1, 2, 3, 4, 5 +), + +current_sources as ( + select + tmdb_id, + content_type, + region, + provider_key, + 'subscription' as monetization_type, + source_updated_at, + 'tmdb_discovery' as source + from {{ ref('stg_tmdb_discovery') }} + + union all + + select + tmdb_id, + content_type, + region, + provider_key, + monetization_type, + source_updated_at, + 'tmdb_watch_providers' as source + from {{ ref('stg_tmdb_watch_providers') }} +), + +current_availability as ( + select + *, + row_number() over ( + partition by tmdb_id, content_type, region, provider_key, monetization_type + order by source_updated_at desc, source + ) as source_rank + from current_sources ) select - discovery.tmdb_id, - discovery.content_type, - discovery.region, - discovery.provider_key, - 'subscription' as monetization_type, + current.tmdb_id, + current.content_type, + current.region, + current.provider_key, + current.monetization_type, additions.available_since, cast(null as timestamptz) as available_from, cast(null as timestamptz) as expires_on, additions.watch_url, true as is_available, false as is_upcoming, - 'tmdb_discovery' as source, + current.source, greatest( - discovery.source_updated_at, - coalesce(additions.event_ingested_at, discovery.source_updated_at) + current.source_updated_at, + coalesce(additions.event_ingested_at, current.source_updated_at) ) as last_updated_at -from {{ ref('stg_tmdb_discovery') }} as discovery +from current_availability as current left join additions - on discovery.tmdb_id = additions.tmdb_id - and discovery.content_type = additions.content_type - and discovery.region = additions.region - and discovery.provider_key = additions.provider_key - and additions.monetization_type = 'subscription' + on current.tmdb_id = additions.tmdb_id + and current.content_type = additions.content_type + and current.region = additions.region + and current.provider_key = additions.provider_key + and current.monetization_type = additions.monetization_type +where current.source_rank = 1 diff --git a/warehouse/models/intermediate/int_upcoming_availability.sql b/warehouse/models/intermediate/int_upcoming_availability.sql index 9d65394..04b63f1 100644 --- a/warehouse/models/intermediate/int_upcoming_availability.sql +++ b/warehouse/models/intermediate/int_upcoming_availability.sql @@ -20,11 +20,11 @@ select upcoming.region, upcoming.provider_key, upcoming.monetization_type, - cast(null as timestamptz) as available_since, + current.available_since, upcoming.event_effective_at as available_from, cast(null as timestamptz) as expires_on, upcoming.watch_url, - false as is_available, + current.tmdb_id is not null as is_available, true as is_upcoming, upcoming.source, upcoming.ingested_at as last_updated_at @@ -36,7 +36,6 @@ left join current_availability as current and upcoming.provider_key = current.provider_key and upcoming.monetization_type = current.monetization_type where upcoming.announcement_rank = 1 - and current.tmdb_id is null and ( upcoming.event_effective_at is null or upcoming.event_effective_at > current_timestamp diff --git a/warehouse/models/marts/_marts.yml b/warehouse/models/marts/_marts.yml index 73a15e7..064bd92 100644 --- a/warehouse/models/marts/_marts.yml +++ b/warehouse/models/marts/_marts.yml @@ -32,7 +32,7 @@ models: [tmdb_id, content_type, region, provider_key, monetization_type] - dbt_utils.expression_is_true: arguments: - expression: "is_available <> is_upcoming" + expression: "is_available or is_upcoming" columns: - name: tmdb_id description: "{{ doc('col_tmdb_id') }}" @@ -65,6 +65,12 @@ models: - name: runtime_minutes description: "{{ doc('col_runtime_minutes') }}" data_type: integer + - name: episode_count + description: "{{ doc('col_episode_count') }}" + data_type: integer + - name: season_count + description: "{{ doc('col_season_count') }}" + data_type: integer - name: original_language description: "{{ doc('col_original_language') }}" data_type: varchar @@ -148,7 +154,7 @@ models: tests: - dbt_utils.expression_is_true: arguments: - expression: "catalog_row_count = current_row_count + upcoming_row_count" + expression: "catalog_row_count <= current_row_count + upcoming_row_count and catalog_row_count >= greatest(current_row_count, upcoming_row_count)" - dbt_utils.expression_is_true: arguments: expression: "catalog_row_count > 0" @@ -207,6 +213,8 @@ models: - {name: release_date, description: "{{ doc('col_release_date') }}", data_type: date} - {name: release_year, description: "{{ doc('col_release_year') }}", data_type: integer} - {name: runtime_minutes, description: "{{ doc('col_runtime_minutes') }}", data_type: integer} + - {name: episode_count, description: "{{ doc('col_episode_count') }}", data_type: integer} + - {name: season_count, description: "{{ doc('col_season_count') }}", data_type: integer} - {name: genre_ids, description: "{{ doc('col_genre_ids') }}", data_type: json} - {name: original_language, description: "{{ doc('col_original_language') }}", data_type: varchar} - {name: tmdb_rating, description: "{{ doc('col_tmdb_rating') }}", data_type: double} @@ -276,7 +284,7 @@ models: arguments: combination_of_columns: [tmdb_id, content_type, region, provider_key, monetization_type] - dbt_utils.expression_is_true: - arguments: {expression: "is_available <> is_upcoming"} + arguments: {expression: "is_available or is_upcoming"} columns: - {name: tmdb_id, description: "{{ doc('col_tmdb_id') }}", data_type: bigint, tests: [not_null]} - name: content_type diff --git a/warehouse/models/marts/catalog_availability.sql b/warehouse/models/marts/catalog_availability.sql index 9f7d625..57037fe 100644 --- a/warehouse/models/marts/catalog_availability.sql +++ b/warehouse/models/marts/catalog_availability.sql @@ -1,5 +1,15 @@ with availability as ( - select * from {{ ref('int_current_availability') }} + select current.* + from {{ ref('int_current_availability') }} as current + where not exists ( + select 1 + from {{ ref('int_upcoming_availability') }} as upcoming + where upcoming.tmdb_id = current.tmdb_id + and upcoming.content_type = current.content_type + and upcoming.region = current.region + and upcoming.provider_key = current.provider_key + and upcoming.monetization_type = current.monetization_type + ) union all select * from {{ ref('int_upcoming_availability') }} ) @@ -13,6 +23,8 @@ select content.release_date, content.release_year, content.runtime_minutes, + content.episode_count, + content.season_count, content.original_language, content.genre_ids, content.tmdb_rating, diff --git a/warehouse/models/marts/dim_content.sql b/warehouse/models/marts/dim_content.sql index 3caf4de..49a5e7f 100644 --- a/warehouse/models/marts/dim_content.sql +++ b/warehouse/models/marts/dim_content.sql @@ -7,6 +7,8 @@ select release_date, release_year, runtime_minutes, + episode_count, + season_count, genre_ids, original_language, tmdb_rating, diff --git a/warehouse/models/marts/streaming_availability.sql b/warehouse/models/marts/streaming_availability.sql index c5fbfe5..0d7dae8 100644 --- a/warehouse/models/marts/streaming_availability.sql +++ b/warehouse/models/marts/streaming_availability.sql @@ -1,3 +1,13 @@ -select * from {{ ref('int_current_availability') }} +select current.* +from {{ ref('int_current_availability') }} as current +where not exists ( + select 1 + from {{ ref('int_upcoming_availability') }} as upcoming + where upcoming.tmdb_id = current.tmdb_id + and upcoming.content_type = current.content_type + and upcoming.region = current.region + and upcoming.provider_key = current.provider_key + and upcoming.monetization_type = current.monetization_type +) union all select * from {{ ref('int_upcoming_availability') }} diff --git a/warehouse/models/staging/_staging.yml b/warehouse/models/staging/_staging.yml index d6fa01f..05bd780 100644 --- a/warehouse/models/staging/_staging.yml +++ b/warehouse/models/staging/_staging.yml @@ -1,6 +1,62 @@ version: 2 models: + - name: stg_tmdb_watch_providers + description: "{{ doc('stg_tmdb_watch_providers') }}" + meta: + owner: watchpulse + source: tmdb + grain: one row per content, region, provider, and monetization type + tests: + - dbt_utils.unique_combination_of_columns: + arguments: + combination_of_columns: [tmdb_id, content_type, region, provider_key, monetization_type] + columns: + - {name: tmdb_id, description: "{{ doc('col_tmdb_id') }}", data_type: bigint, tests: [not_null]} + - {name: content_type, description: "{{ doc('col_content_type') }}", data_type: varchar, tests: [not_null]} + - {name: region, description: "{{ doc('col_region') }}", data_type: varchar, tests: [not_null]} + - {name: provider_key, description: "{{ doc('col_provider_key') }}", data_type: varchar, tests: [not_null]} + - {name: monetization_type, description: "{{ doc('col_monetization_type') }}", data_type: varchar, tests: [not_null]} + - {name: source_updated_at, description: "{{ doc('col_source_updated_at') }}", data_type: timestamptz, tests: [not_null]} + - {name: source_file, description: "{{ doc('col_source_file') }}", data_type: varchar, tests: [not_null]} + + - name: stg_tmdb_metadata + description: "{{ doc('stg_tmdb_metadata') }}" + meta: + owner: watchpulse + source: tmdb + grain: one row per tmdb_id and content_type + tests: + - dbt_utils.unique_combination_of_columns: + arguments: + combination_of_columns: [tmdb_id, content_type] + columns: + - {name: tmdb_id, description: "{{ doc('col_tmdb_id') }}", data_type: bigint, tests: [not_null]} + - name: content_type + description: "{{ doc('col_content_type') }}" + data_type: varchar + tests: + - not_null + - accepted_values: + arguments: + values: [movie, tv] + - {name: title, description: "{{ doc('col_title') }}", data_type: varchar, tests: [not_null]} + - {name: original_title, description: "{{ doc('col_original_title') }}", data_type: varchar} + - {name: overview, description: "{{ doc('col_overview') }}", data_type: varchar} + - {name: release_date, description: "{{ doc('col_release_date') }}", data_type: date} + - {name: runtime_minutes, description: "{{ doc('col_runtime_minutes') }}", data_type: integer} + - {name: episode_count, description: "{{ doc('col_episode_count') }}", data_type: integer} + - {name: season_count, description: "{{ doc('col_season_count') }}", data_type: integer} + - {name: genre_ids, description: "{{ doc('col_genre_ids') }}", data_type: json} + - {name: original_language, description: "{{ doc('col_original_language') }}", data_type: varchar} + - {name: tmdb_rating, description: "{{ doc('col_tmdb_rating') }}", data_type: double} + - {name: vote_count, description: "{{ doc('col_vote_count') }}", data_type: bigint, tests: [not_null]} + - {name: tmdb_popularity, description: "{{ doc('col_tmdb_popularity') }}", data_type: double} + - {name: poster_path, description: "{{ doc('col_poster_path') }}", data_type: varchar} + - {name: backdrop_path, description: "{{ doc('col_backdrop_path') }}", data_type: varchar} + - {name: source_updated_at, description: "{{ doc('col_source_updated_at') }}", data_type: timestamptz, tests: [not_null]} + - {name: source_file, description: "{{ doc('col_source_file') }}", data_type: varchar, tests: [not_null]} + - name: stg_streaming_links description: "{{ doc('stg_streaming_links') }}" meta: diff --git a/warehouse/models/staging/stg_tmdb_metadata.sql b/warehouse/models/staging/stg_tmdb_metadata.sql new file mode 100644 index 0000000..7b39c37 --- /dev/null +++ b/warehouse/models/staging/stg_tmdb_metadata.sql @@ -0,0 +1,74 @@ +{% set lake_root = var('lake_root', '../data/lake') %} + +with raw_records as ( + select + filename as source_file, + cast(fetched_at as timestamptz) as fetched_at, + cast(request_params as json) as request_params, + cast(payload as json) as payload + from read_parquet( + '{{ lake_root }}/raw/source=tmdb/endpoint=metadata/entity_type=*/country=ALL/date=*/*.parquet', + filename = true, + hive_partitioning = false, + union_by_name = true + ) +), + +normalized as ( + select + cast(json_extract_string(request_params, '$.tmdb_id') as bigint) as tmdb_id, + json_extract_string(request_params, '$.entity_type') as content_type, + coalesce( + json_extract_string(payload, '$.title'), + json_extract_string(payload, '$.name') + ) as title, + coalesce( + json_extract_string(payload, '$.original_title'), + json_extract_string(payload, '$.original_name') + ) as original_title, + nullif(json_extract_string(payload, '$.overview'), '') as overview, + try_cast(coalesce( + json_extract_string(payload, '$.release_date'), + json_extract_string(payload, '$.first_air_date') + ) as date) as release_date, + try_cast(json_extract_string(payload, '$.runtime') as integer) as movie_runtime, + try_cast(json_extract_string(payload, '$.episode_run_time[0]') as integer) as series_runtime, + try_cast(json_extract_string(payload, '$.number_of_episodes') as integer) as episode_count, + try_cast(json_extract_string(payload, '$.number_of_seasons') as integer) as season_count, + json_extract(payload, '$.genres') as genres, + json_extract_string(payload, '$.original_language') as original_language, + try_cast(json_extract_string(payload, '$.vote_average') as double) as tmdb_rating, + coalesce(try_cast(json_extract_string(payload, '$.vote_count') as bigint), 0) as vote_count, + try_cast(json_extract_string(payload, '$.popularity') as double) as tmdb_popularity, + json_extract_string(payload, '$.poster_path') as poster_path, + json_extract_string(payload, '$.backdrop_path') as backdrop_path, + fetched_at as source_updated_at, + source_file + from raw_records +) + +select + tmdb_id, + content_type, + title, + original_title, + overview, + release_date, + coalesce(nullif(movie_runtime, 0), nullif(series_runtime, 0)) as runtime_minutes, + episode_count, + season_count, + to_json(list_transform(cast(genres as json[]), genre -> cast(json_extract_string(genre, '$.id') as integer))) as genre_ids, + original_language, + tmdb_rating, + vote_count, + tmdb_popularity, + poster_path, + backdrop_path, + source_updated_at, + source_file +from normalized +where tmdb_id is not null and content_type in ('movie', 'tv') +qualify row_number() over ( + partition by tmdb_id, content_type + order by source_updated_at desc, source_file desc +) = 1 diff --git a/warehouse/models/staging/stg_tmdb_watch_providers.sql b/warehouse/models/staging/stg_tmdb_watch_providers.sql new file mode 100644 index 0000000..bbe65e1 --- /dev/null +++ b/warehouse/models/staging/stg_tmdb_watch_providers.sql @@ -0,0 +1,56 @@ +{% set lake_root = var('lake_root', '../data/lake') %} + +with raw as ( + select + cast(request_params as json) as request_params, + cast(payload as json) as payload, + cast(fetched_at as timestamptz) as source_updated_at, + filename as source_file + from read_parquet( + '{{ lake_root }}/raw/source=tmdb/endpoint=watch_providers/entity_type=*/country=ALL/date=*/*.parquet', + filename = true, + hive_partitioning = false, + union_by_name = true + ) +), + +flattened as ( + select + cast(json_extract_string(raw.request_params, '$.tmdb_id') as bigint) as tmdb_id, + json_extract_string(raw.request_params, '$.entity_type') as content_type, + upper(region.key) as region, + cast(json_extract_string(provider.value, '$.provider_id') as varchar) as source_provider_id, + raw.source_updated_at, + raw.source_file + from raw, + json_each(raw.payload, '$.results') as region, + json_each(region.value, '$.flatrate') as provider +), + +mapped as ( + select + flattened.tmdb_id, + flattened.content_type, + flattened.region, + provider.provider_key, + 'subscription' as monetization_type, + flattened.source_updated_at, + flattened.source_file + from flattened + inner join {{ ref('provider_source_reference') }} as provider + on provider.source = 'tmdb' + and provider.region = flattened.region + and provider.source_provider_id = flattened.source_provider_id +) + +select * exclude row_rank +from ( + select + *, + row_number() over ( + partition by tmdb_id, content_type, region, provider_key, monetization_type + order by source_updated_at desc, source_file desc + ) as row_rank + from mapped +) +where row_rank = 1 diff --git a/warehouse/seeds/genre_reference.csv b/warehouse/seeds/genre_reference.csv index 5af8d18..975d3a3 100644 --- a/warehouse/seeds/genre_reference.csv +++ b/warehouse/seeds/genre_reference.csv @@ -25,6 +25,7 @@ tv,37,Western tv,80,Crime tv,99,Documentary tv,9648,Mystery +tv,10749,Romance tv,10751,Family tv,10759,Action & Adventure tv,10762,Kids diff --git a/watchpulse/api/filters.py b/watchpulse/api/filters.py index cf08889..417233d 100644 --- a/watchpulse/api/filters.py +++ b/watchpulse/api/filters.py @@ -129,6 +129,21 @@ def validate_provider_keys(cls, providers: tuple[str, ...]) -> tuple[str, ...]: DiscoveryFiltersQuery = Annotated[DiscoveryFilters, Query()] """FastAPI query binding reused by every discovery endpoint.""" + +class TitleSearchFilters(DiscoveryFilters): + """Global discovery filters plus a validated local title query.""" + + query: str = Field(min_length=2, max_length=100) + + @field_validator("query", mode="before") + @classmethod + def normalize_query(cls, value: object) -> object: + return value.strip() if isinstance(value, str) else value + + +TitleSearchFiltersQuery = Annotated[TitleSearchFilters, Query()] +"""FastAPI query binding for title search.""" + CatalogScopeQuery = Annotated[CatalogScope, Query()] """FastAPI query binding for region/provider-aware reference options.""" diff --git a/watchpulse/api/models.py b/watchpulse/api/models.py index 510206c..0f4889b 100644 --- a/watchpulse/api/models.py +++ b/watchpulse/api/models.py @@ -112,6 +112,8 @@ class CatalogItemResponse(BaseModel): metadata_source: str last_updated_at: datetime availabilities: tuple[AvailabilityResponse, ...] + episode_count: int | None = None + season_count: int | None = None class RankedCatalogItemResponse(CatalogItemResponse): @@ -162,3 +164,10 @@ class LeavingSoonResponse(BaseModel): class TitleDetailsResponse(CatalogItemResponse): region: str + + +class TitleSearchResponse(BaseModel): + query: str + filters: DiscoveryFilters + count: int + items: tuple[CatalogItemResponse, ...] diff --git a/watchpulse/api/query.py b/watchpulse/api/query.py index f4eb10a..7071521 100644 --- a/watchpulse/api/query.py +++ b/watchpulse/api/query.py @@ -40,6 +40,7 @@ class DiscoveryRequest: expires_from: datetime | None = None expires_to: datetime | None = None tmdb_id: int | None = None + title_query: str | None = None def __post_init__(self) -> None: if not 1 <= self.limit <= 100: @@ -48,6 +49,8 @@ def __post_init__(self) -> None: raise ValueError("offset must be zero or greater") if self.tmdb_id is not None and self.tmdb_id <= 0: raise ValueError("tmdb_id must be positive") + if self.title_query is not None and not 2 <= len(self.title_query.strip()) <= 100: + raise ValueError("title_query must contain between 2 and 100 characters") if ( self.release_date_from is not None and self.release_date_to is not None @@ -78,8 +81,8 @@ class DiscoveryQueryBuilder: """Build discovery SQL using only fixed fragments and bound values.""" _availability_predicates = { - AvailabilityState.CURRENT: ("catalog.is_available = true and catalog.is_upcoming = false"), - AvailabilityState.UPCOMING: ("catalog.is_upcoming = true and catalog.is_available = false"), + AvailabilityState.CURRENT: "catalog.is_available = true", + AvailabilityState.UPCOMING: "catalog.is_upcoming = true", AvailabilityState.ANY: "true", } _sort_expressions = { @@ -162,6 +165,13 @@ def build(self, request: DiscoveryRequest) -> ParameterizedQuery: if request.tmdb_id is not None: predicates.append("catalog.tmdb_id = ?") parameters.append(request.tmdb_id) + if request.title_query is not None: + predicates.append( + "(contains(lower(catalog.title), lower(?)) " + "or contains(lower(coalesce(catalog.original_title, '')), lower(?)))" + ) + normalized_query = request.title_query.strip() + parameters.extend((normalized_query, normalized_query)) parameters.extend((request.limit, request.offset)) sql = f""" @@ -204,7 +214,9 @@ def build(self, request: DiscoveryRequest) -> ParameterizedQuery: ) order by catalog.provider_name, catalog.provider_key, catalog.monetization_type - ) as availabilities + ) as availabilities, + catalog.episode_count, + catalog.season_count from main_marts.catalog_availability as catalog where {" and ".join(predicates)} group by @@ -223,7 +235,9 @@ def build(self, request: DiscoveryRequest) -> ParameterizedQuery: catalog.popularity_score, catalog.poster_path, catalog.backdrop_path, - catalog.metadata_source + catalog.metadata_source, + catalog.episode_count, + catalog.season_count order by {self._sort_expressions[request.sort]}, catalog.tmdb_id, catalog.content_type limit ? offset ? diff --git a/watchpulse/api/repository.py b/watchpulse/api/repository.py index 2145f4b..3fd8d20 100644 --- a/watchpulse/api/repository.py +++ b/watchpulse/api/repository.py @@ -88,6 +88,8 @@ class CatalogItem: metadata_source: str last_updated_at: datetime availabilities: tuple[CatalogAvailability, ...] + episode_count: int | None = None + season_count: int | None = None class CatalogRepository: @@ -227,6 +229,8 @@ def _catalog_item(row: tuple[Any, ...]) -> CatalogItem: metadata_source=row[16], last_updated_at=row[17], availabilities=availabilities, + episode_count=row[19], + season_count=row[20], ) @staticmethod diff --git a/watchpulse/api/routers/discovery.py b/watchpulse/api/routers/discovery.py index b47aaf5..2154ed1 100644 --- a/watchpulse/api/routers/discovery.py +++ b/watchpulse/api/routers/discovery.py @@ -11,6 +11,7 @@ DiscoveryFilters, DiscoveryFiltersQuery, TitleScopeQuery, + TitleSearchFiltersQuery, ) from watchpulse.api.models import ( CatalogItemResponse, @@ -19,6 +20,7 @@ RankedCatalogItemResponse, RecentlyAddedResponse, TitleDetailsResponse, + TitleSearchResponse, TopTenResponse, UpcomingResponse, ) @@ -32,6 +34,28 @@ def _repository(request: Request) -> CatalogRepository: return request.app.state.catalog_repository +@router.get("/search", response_model=TitleSearchResponse) +async def search_titles( + request: Request, + search: TitleSearchFiltersQuery, +) -> TitleSearchResponse: + filters = DiscoveryFilters.model_validate(search.model_dump(exclude={"query"})) + discovery = DiscoveryRequest( + filters=filters, + availability=AvailabilityState.ANY, + sort=DiscoverySort.POPULARITY, + limit=8, + title_query=search.query, + ) + items = _repository(request).discover(discovery) + return TitleSearchResponse( + query=search.query, + filters=filters, + count=len(items), + items=tuple(CatalogItemResponse.model_validate(item) for item in items), + ) + + @router.get("/top-10", response_model=TopTenResponse) async def top_ten(request: Request, filters: DiscoveryFiltersQuery) -> TopTenResponse: query = DiscoveryRequest( diff --git a/watchpulse/config.py b/watchpulse/config.py index 800023b..bdb6efc 100644 --- a/watchpulse/config.py +++ b/watchpulse/config.py @@ -44,6 +44,12 @@ class Settings: new_release_days: int recently_added_days: int leaving_soon_days: int + tmdb_enrichment_max_titles_per_run: int + tmdb_enrichment_upcoming_refresh_days: int + tmdb_enrichment_recent_refresh_days: int + tmdb_enrichment_series_refresh_days: int + tmdb_enrichment_movie_refresh_days: int + tmdb_enrichment_provider_refresh_days: int @classmethod def from_env(cls, env: Mapping[str, str] | None = None) -> Settings: @@ -89,6 +95,24 @@ def from_env(cls, env: Mapping[str, str] | None = None) -> Settings: new_release_days=_positive_int(values, "NEW_RELEASE_DAYS", 90), recently_added_days=_positive_int(values, "RECENTLY_ADDED_DAYS", 30), leaving_soon_days=_positive_int(values, "LEAVING_SOON_DAYS", 30), + tmdb_enrichment_max_titles_per_run=_positive_int( + values, "TMDB_ENRICHMENT_MAX_TITLES_PER_RUN", 250 + ), + tmdb_enrichment_upcoming_refresh_days=_positive_int( + values, "TMDB_ENRICHMENT_UPCOMING_REFRESH_DAYS", 7 + ), + tmdb_enrichment_recent_refresh_days=_positive_int( + values, "TMDB_ENRICHMENT_RECENT_REFRESH_DAYS", 30 + ), + tmdb_enrichment_series_refresh_days=_positive_int( + values, "TMDB_ENRICHMENT_SERIES_REFRESH_DAYS", 90 + ), + tmdb_enrichment_movie_refresh_days=_positive_int( + values, "TMDB_ENRICHMENT_MOVIE_REFRESH_DAYS", 180 + ), + tmdb_enrichment_provider_refresh_days=_positive_int( + values, "TMDB_ENRICHMENT_PROVIDER_REFRESH_DAYS", 60 + ), ) diff --git a/watchpulse/warehouse_publish.py b/watchpulse/warehouse_publish.py index d6e83a2..1f80b64 100644 --- a/watchpulse/warehouse_publish.py +++ b/watchpulse/warehouse_publish.py @@ -45,7 +45,9 @@ def validate_candidate(candidate_path: Path) -> None: catalog_count, current_count, upcoming_count, built_at, source_updated_at = freshness_rows[0] if actual_row_count <= 0 or catalog_count != actual_row_count: raise ValueError("Candidate catalog is empty or its recorded count is invalid") - if current_count + upcoming_count != catalog_count: + if catalog_count > current_count + upcoming_count or catalog_count < max( + current_count, upcoming_count + ): raise ValueError("Candidate availability-state counts are inconsistent") if built_at is None or source_updated_at is None: raise ValueError("Candidate freshness timestamps must not be null")