diff --git a/Makefile b/Makefile index 2ec4639..c30550a 100644 --- a/Makefile +++ b/Makefile @@ -1,6 +1,6 @@ PYTHON ?= python -.PHONY: install install-hooks api-dev dbt-deps dbt-debug dbt-parse dbt-validate dbt-publish frontend-install frontend-dev frontend-lint frontend-test frontend-build frontend-ci lint precommit pre-commit prepush pre-push syntax test coverage ci +.PHONY: install install-hooks api-dev catalog-refresh dbt-deps dbt-debug dbt-parse dbt-validate dbt-publish frontend-install frontend-dev frontend-lint frontend-test frontend-build frontend-ci lint precommit pre-commit prepush pre-push syntax test coverage ci install: $(PYTHON) -m pip install -e ".[dev,warehouse]" @@ -11,6 +11,9 @@ install-hooks: api-dev: uvicorn watchpulse.api:create_app --factory --reload +catalog-refresh: + $(PYTHON) -m ingestion.full_refresh --country GR --summary-output data/full-refresh-summary.json + frontend-install: cd frontend && npm install diff --git a/README.md b/README.md index cc64e60..5aab7ca 100644 --- a/README.md +++ b/README.md @@ -196,6 +196,46 @@ 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. +## Full catalog refresh + +Run the complete Greece refresh for Netflix, Disney+, Prime Video, and Apple TV+: + +```bash +make catalog-refresh +``` + +The guarded workflow performs complete TMDB movie/TV discovery, publishes a +validated catalog snapshot, ingests all pages of Streaming Availability `new` +and `upcoming` changes within a 100-request run budget, incrementally enriches +missing/stale TMDB metadata, then atomically publishes the final DuckDB. Broad +watch-provider enrichment is opt-in. The previous serving database remains live +if discovery, dbt validation, or candidate publication fails. + +Every discovery page belongs to a run manifest. dbt selects only the latest +complete run for each region/provider/content-type combination, so samples and +interrupted refreshes never leak into the frontend. A title absent from the +newest complete snapshot is no longer considered currently available. + +Useful bounded variants: + +```bash +# Current catalog only; no lifecycle or per-title enrichment. +python -m ingestion.full_refresh --country GR --skip-streaming --skip-enrichment + +# Clean metadata backfill after discovery; provider payloads remain disabled. +python -m ingestion.full_refresh --country GR --enrichment-mode backfill + +# Explicit exceptional provider enrichment. +python -m ingestion.full_refresh --country GR --include-watch-providers + +# Refresh one provider while preserving the latest complete snapshots of the others. +python -m ingestion.full_refresh --country GR --provider netflix +``` + +The default full-refresh summary is retained locally at +`data/full-refresh-summary.json`. Streaming request usage is also recorded in +the operational DuckDB and constrained by the configured monthly cap. + ## Dependency management `pyproject.toml` is the canonical project configuration. It contains package @@ -342,6 +382,10 @@ Add `--metadata-only` to skip watch-provider payloads. Add inspection. Planning never calls TMDB; only execution does. Completed batches are retained as immutable Parquet files, allowing later plans to resume. +See the [catalog refresh runbook](docs/catalog-refresh-runbook.md) for the +twice-weekly operating flow, clean rebuild procedure, diagrams, success gates, +request expectations, and recovery commands. + To override the configured regions for one run, repeat `--country`: ```bash diff --git a/docs/README.md b/docs/README.md index a06efcb..4811101 100644 --- a/docs/README.md +++ b/docs/README.md @@ -6,6 +6,8 @@ Each document has one primary responsibility: - [Data model](data-model.md): grains, keys, fields, history, and data tests. - [Decisions](decisions.md): append-only architecture decision records (ADRs). - [Roadmap](roadmap.md): version-by-version scope and exit criteria. +- [Catalog refresh runbook](catalog-refresh-runbook.md): routine refreshes, + clean rebuilds, validation, publication, and recovery. - [Release notes](releases/): completed version outcomes and limitations. `../AGENTS.md` is the authoritative product brief. When implementation changes a diff --git a/docs/catalog-refresh-runbook.md b/docs/catalog-refresh-runbook.md new file mode 100644 index 0000000..1d21b29 --- /dev/null +++ b/docs/catalog-refresh-runbook.md @@ -0,0 +1,251 @@ +# WatchPulse catalog refresh runbook + +This runbook defines the supported operational flows for Greece and the four +configured subscription providers: Netflix, Disney+, Prime Video, and Apple +TV+. External APIs are ingestion sources only. The frontend and FastAPI always +query the last successfully published DuckDB. + +## Data-source responsibilities + +| Source | Responsibility | +|---|---| +| TMDB discovery | Current region/provider subscription membership plus frequently changing rating, vote count, and popularity | +| TMDB title details | Movie runtime and TV season/episode totals; also fills missing canonical metadata | +| TMDB watch providers | Optional reconciliation and rent/buy/free evidence; disabled by default | +| Streaming Availability | `new` and `upcoming` lifecycle events, dates, and direct links when supplied | + +Provider logos do not require per-title TMDB watch-provider requests. A logo is +clickable only when a verified `watch_url` exists; otherwise it remains a +non-clickable availability badge. + +## Routine refresh: twice per week + +Use this flow every two or three days. + +```mermaid +flowchart TD + A[Start routine refresh] --> B[Complete TMDB discovery\nGR × 4 providers × movie/TV] + B --> C{All eight queries complete\nand below 500 pages?} + C -- No --> X[Keep previous serving DB\nrecord failure] + C -- Yes --> D[Write complete snapshot manifest] + D --> E[Build and publish current catalog candidate] + E --> F[Streaming API: new + upcoming\nwithin request/monthly budgets] + F --> G[Plan incremental metadata locally] + G --> H[Enrich new, missing, or stale metadata\nno broad provider calls] + H --> I[Enrich lifecycle-only titles missing TMDB metadata] + I --> J[dbt build + 236 quality tests] + J --> K{Candidate valid?} + K -- No --> X + K -- Yes --> L[Atomically publish DuckDB] + L --> M[Frontend reads new snapshot] +``` + +### One-command routine run + +From the repository root: + +```bash +make catalog-refresh +``` + +Equivalent explicit command: + +```bash +python -m ingestion.full_refresh \ + --country GR \ + --enrichment-mode incremental \ + --streaming-max-requests 100 \ + --summary-output data/full-refresh-summary.json +``` + +The default does **not** call the TMDB watch-provider endpoint. Use +`--include-watch-providers` only for an explicit reconciliation investigation. + +### What incremental metadata selects + +The local plan makes no API calls. It selects: + +- titles never enriched; +- newly discovered titles; +- movies whose runtime is missing and whose retry window is due; +- TV titles whose season/episode totals are missing and due for retry; +- upcoming, recent, active, or otherwise stale titles based on configured + refresh windows. + +Ratings, vote counts, and popularity come from the newest complete discovery +snapshot. They do not require title-detail enrichment. + +Inspect the plan without making requests: + +```bash +python -m ingestion.enrich_catalog \ + --mode incremental \ + --metadata-only \ + --dry-run \ + --plan-output data/incremental-enrichment-plan.json +``` + +### Lifecycle-only metadata + +An upcoming title may correctly be absent from discovery because it is not yet +streaming. Enrich missing metadata for both lifecycle types: + +```bash +python -m ingestion.enrich_streaming_metadata --event-type new --country GR +python -m ingestion.enrich_streaming_metadata --event-type upcoming --country GR +``` + +These commands request TMDB title details only by default. They do not request +TMDB watch-provider payloads. + +## Clean rebuild from scratch + +A clean rebuild recreates the complete raw and serving state when no retained +catalog or metadata exists. Back up the current `data/` directory before any +manual cleanup. This runbook intentionally does not prescribe a deletion +command; confirm the exact target and recovery plan first. + +```mermaid +flowchart TD + A[Empty, explicitly selected data paths] --> B[Full TMDB discovery] + B --> C[Complete snapshot manifest] + C --> D[Initial dbt publication\ncreates serving catalog] + D --> E[Streaming new + upcoming] + E --> F[Backfill TMDB title details\nonce per unique title] + F --> G[Target lifecycle-only missing titles] + G --> H[Final dbt build and tests] + H --> I[Atomic serving DB publication] + I --> J[Count, freshness, and API verification] +``` + +### Step 1: verify configuration + +Confirm `.env` contains valid keys and the intended scope: + +```dotenv +SUPPORTED_REGIONS=GR +SUPPORTED_PROVIDERS=netflix,disney_plus,prime_video,apple_tv_plus +``` + +Never print or commit API keys. + +### Step 2: preview discovery configuration + +```bash +python -m ingestion.run --help +``` + +Do not use `--max-pages` for a complete rebuild. Page-capped runs are samples +and cannot become the latest live snapshot. + +### Step 3: execute the guarded rebuild + +```bash +python -m ingestion.full_refresh \ + --country GR \ + --enrichment-mode backfill \ + --enrichment-max-titles 20000 \ + --streaming-max-requests 100 \ + --summary-output data/full-rebuild-summary.json +``` + +This performs metadata-only backfill by default. For the current catalog size, +the approximate request shape is: + +```text +TMDB discovery: ~700 shared page requests +TMDB title details: ~13,000 one-time title requests +TMDB watch providers: 0 by default +Streaming lifecycle: ~12 currently observed requests +``` + +### Step 4: verify the result + +```bash +make dbt-publish +make test +``` + +Inspect the serving database: + +```bash +python -m ingestion.inspect --country GR --limit 20 +``` + +The run succeeds only when: + +- every provider/content discovery query is complete; +- no query is truncated by TMDB's 500-page ceiling; +- dbt builds a non-empty candidate and every data test passes; +- the candidate freshness record is valid; +- atomic publication succeeds. + +## Publication and failure behavior + +```mermaid +sequenceDiagram + participant Job as Refresh job + participant Lake as Append-only Parquet + participant Candidate as Candidate DuckDB + participant Live as Published DuckDB + participant UI as FastAPI / frontend + + UI->>Live: Continue serving last good snapshot + Job->>Lake: Append API responses and manifests + Job->>Candidate: dbt build and tests + alt validation passes + Candidate->>Live: Atomic replacement + UI->>Live: Read new snapshot + else ingestion or validation fails + Job-->>Live: No replacement + UI->>Live: Keep reading previous snapshot + end +``` + +Completed enrichment batches are retained every 25 titles. To resume an +interrupted clean metadata backfill: + +```bash +python -m ingestion.enrich_catalog \ + --mode backfill \ + --metadata-only \ + --max-titles 20000 + +make dbt-publish +``` + +The planner skips successfully retained title details. + +## Monitoring + +Long enrichment runs log progress every minute: + +```text +run_id +titles processed / planned +remaining titles +API request count and observed rate +metadata rows persisted +provider rows persisted +elapsed seconds +estimated remaining seconds +``` + +Pipeline-run metadata also records status, request counts, row counts, errors, +and endpoint summaries in the operational DuckDB. + +## Optional watch-provider reconciliation + +Broad provider enrichment is not part of either normal flow. Run it only when +rent/buy/free evidence or independent provider reconciliation is explicitly +required: + +```bash +python -m ingestion.enrich_catalog \ + --mode incremental \ + --max-titles 100 \ + --dry-run +``` + +Remove `--dry-run` only after reviewing the bounded plan. This endpoint does +not guarantee direct Netflix, Disney+, Prime Video, or Apple TV+ playback URLs. diff --git a/ingestion/enrich_catalog.py b/ingestion/enrich_catalog.py index 71926ca..938d0ea 100644 --- a/ingestion/enrich_catalog.py +++ b/ingestion/enrich_catalog.py @@ -33,6 +33,7 @@ logger = logging.getLogger("ingestion.enrich_catalog") EnrichmentMode = Literal["backfill", "incremental"] BATCH_SIZE = 25 +PROGRESS_LOG_INTERVAL_SECONDS = 60.0 @dataclass(frozen=True) @@ -263,6 +264,8 @@ def execute_enrichment( provider_records: list[RawRecord] = [] metadata_written = 0 providers_written = 0 + titles_processed = 0 + last_progress_at = started_at if runs: runs.start(run_id=run_id, job_name="tmdb_catalog_enrichment", source="tmdb") try: @@ -296,6 +299,31 @@ def execute_enrichment( endpoint="watch_providers", run_id=run_id, ) + titles_processed += 1 + now = time.monotonic() + if now - last_progress_at >= PROGRESS_LOG_INTERVAL_SECONDS: + elapsed = now - started_at + rate = source.request_count / elapsed if elapsed else 0.0 + remaining = len(plan) - titles_processed + eta_seconds = ( + remaining / (titles_processed / elapsed) if titles_processed else 0 + ) + logger.info( + "enrichment progress run_id=%s titles=%d/%d remaining=%d " + "requests=%d request_rate=%.2f/s metadata_persisted=%d " + "providers_persisted=%d elapsed_seconds=%.1f eta_seconds=%.1f", + run_id, + titles_processed, + len(plan), + remaining, + source.request_count, + rate, + metadata_written, + providers_written, + elapsed, + eta_seconds, + ) + last_progress_at = now metadata_written += _write_batch( metadata_records, lake_root=lake_root, endpoint="metadata", run_id=run_id ) @@ -309,6 +337,7 @@ def execute_enrichment( "run_id": run_id, "mode": mode, "titles_planned": len(plan), + "titles_processed": titles_processed, "metadata_records_written": metadata_written, "provider_records_written": providers_written, "api_request_count": source.request_count, @@ -323,7 +352,7 @@ def execute_enrichment( details=summary, ) return summary - except Exception as error: + except BaseException as error: if runs: runs.fail( run_id=run_id, diff --git a/ingestion/enrich_streaming_metadata.py b/ingestion/enrich_streaming_metadata.py index 4153187..96bf954 100644 --- a/ingestion/enrich_streaming_metadata.py +++ b/ingestion/enrich_streaming_metadata.py @@ -100,13 +100,18 @@ def run( event_type: str = "upcoming", country: str | None = None, max_titles: int | None = None, + include_watch_providers: bool = False, 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" + availability_refs = ( + pending_title_refs( + lake_root, event_type=event_type, country=country, endpoint="watch_providers" + ) + if include_watch_providers + else () ) if max_titles is not None: allowed = set( @@ -176,6 +181,7 @@ def main() -> None: parser.add_argument("--event-type", choices=("new", "upcoming"), default="upcoming") parser.add_argument("--country") parser.add_argument("--max-titles", type=int) + parser.add_argument("--include-watch-providers", action="store_true") 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: @@ -188,6 +194,7 @@ def main() -> None: event_type=args.event_type, country=args.country, max_titles=args.max_titles, + include_watch_providers=args.include_watch_providers, tmdb_base_url=settings.tmdb_base_url, ) logger.info("enrichment complete: %s", summary) diff --git a/ingestion/full_refresh.py b/ingestion/full_refresh.py new file mode 100644 index 0000000..bdcbd2a --- /dev/null +++ b/ingestion/full_refresh.py @@ -0,0 +1,215 @@ +"""Run a guarded, end-to-end WatchPulse catalog refresh. + +The workflow keeps external ingestion out of the serving path and publishes +only validated DuckDB candidates. TMDB discovery is mandatory; lifecycle and +enrichment stages are independently configurable and bounded. +""" + +from __future__ import annotations + +import argparse +import json +import logging +from dataclasses import asdict +from pathlib import Path + +from dotenv import load_dotenv + +from ingestion.enrich_catalog import EnrichmentMode, execute_enrichment, plan_enrichment +from ingestion.enrich_streaming_metadata import run as enrich_lifecycle_metadata +from ingestion.run import run as run_tmdb_discovery +from ingestion.run_streaming_availability import run as run_streaming_lifecycle +from ingestion.sources.streaming_availability.config import DEFAULT_CHANGE_TYPES +from ingestion.sources.tmdb.config import PROVIDERS +from watchpulse.config import Settings +from watchpulse.warehouse_publish import publish_warehouse + +logger = logging.getLogger("ingestion.full_refresh") + + +def run_full_refresh( + *, + settings: Settings, + country: str, + provider_keys: tuple[str, ...], + streaming_max_requests: int, + enrichment_max_titles: int, + include_watch_providers: bool, + enrichment_mode: EnrichmentMode = "incremental", + run_streaming: bool = True, + run_enrichment: bool = True, + project_dir: Path = Path("warehouse"), +) -> dict[str, object]: + """Refresh raw sources and atomically publish validated serving data.""" + if not settings.tmdb_api_key: + raise ValueError("TMDB_API_KEY is required") + unknown = set(provider_keys) - PROVIDERS.keys() + if unknown: + raise ValueError(f"Unknown provider keys: {sorted(unknown)}") + country = country.upper() + summary: dict[str, object] = { + "country": country, + "provider_keys": list(provider_keys), + "stages": {}, + } + stages = summary["stages"] + assert isinstance(stages, dict) + + logger.info( + "full refresh started country=%s providers=%s streaming=%s enrichment=%s", + country, + ",".join(provider_keys), + run_streaming, + run_enrichment, + ) + logger.info("stage started: tmdb_discovery") + discovery = run_tmdb_discovery( + lake_root=settings.lake_root, + database_path=settings.database_path, + api_key=settings.tmdb_api_key, + countries=(country,), + providers={key: PROVIDERS[key] for key in provider_keys}, + enrich=False, + tmdb_base_url=settings.tmdb_base_url, + ) + stages["tmdb_discovery"] = discovery + if not discovery["discovery_complete"]: + raise RuntimeError("TMDB discovery was incomplete; serving database was not published") + logger.info( + "stage completed: tmdb_discovery requests=%s pages=%s", + discovery["api_request_count"], + discovery["discover_pages_written"], + ) + + logger.info("stage started: catalog_publication") + publish_warehouse(project_dir, settings.serving_database_path) + stages["catalog_publication"] = {"status": "success"} + logger.info("stage completed: catalog_publication") + + if run_streaming: + if not settings.streaming_availability_api_key: + raise ValueError("STREAMING_AVAILABILITY_API_KEY is required") + logger.info("stage started: streaming_lifecycle") + stages["streaming_lifecycle"] = run_streaming_lifecycle( + lake_root=settings.lake_root, + database_path=settings.database_path, + api_key=settings.streaming_availability_api_key, + base_url=settings.streaming_availability_base_url, + countries=(country,), + provider_keys=provider_keys, + max_requests_per_run=streaming_max_requests, + monthly_cap=settings.streaming_availability_monthly_cap, + change_types=DEFAULT_CHANGE_TYPES, + max_pages_per_type=None, + ) + logger.info( + "stage completed: streaming_lifecycle requests=%s events=%s", + stages["streaming_lifecycle"]["api_request_count"], + stages["streaming_lifecycle"]["events_written"], + ) + + if run_enrichment: + logger.info("stage started: tmdb_enrichment_planning") + plan = plan_enrichment( + database_path=settings.serving_database_path, + lake_root=settings.lake_root, + mode=enrichment_mode, + max_titles=enrichment_max_titles, + include_providers=include_watch_providers, + 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, + ) + stages["tmdb_enrichment_plan"] = { + "titles": len(plan), + "metadata_requests": sum(item.metadata_due for item in plan), + "provider_requests": sum(item.providers_due for item in plan), + "sample": [asdict(item) for item in plan[:10]], + } + logger.info( + "stage completed: tmdb_enrichment_planning titles=%d requests=%d", + len(plan), + sum(item.metadata_due + item.providers_due for item in plan), + ) + logger.info("stage started: tmdb_enrichment") + stages["tmdb_enrichment"] = execute_enrichment( + plan, + lake_root=settings.lake_root, + api_key=settings.tmdb_api_key, + mode=enrichment_mode, + tmdb_base_url=settings.tmdb_base_url, + database_path=settings.database_path, + ) + logger.info( + "stage completed: tmdb_enrichment requests=%s", + stages["tmdb_enrichment"]["api_request_count"], + ) + + if run_streaming: + logger.info("stage started: lifecycle_metadata") + lifecycle_enrichment = {} + for event_type in DEFAULT_CHANGE_TYPES: + lifecycle_enrichment[event_type] = enrich_lifecycle_metadata( + lake_root=settings.lake_root, + api_key=settings.tmdb_api_key, + event_type=event_type, + country=country, + include_watch_providers=False, + tmdb_base_url=settings.tmdb_base_url, + ) + stages["lifecycle_metadata"] = lifecycle_enrichment + logger.info("stage completed: lifecycle_metadata") + + logger.info("stage started: final_publication") + publish_warehouse(project_dir, settings.serving_database_path) + stages["final_publication"] = {"status": "success"} + logger.info("stage completed: final_publication") + logger.info("full refresh completed country=%s", country) + return summary + + +def build_parser() -> argparse.ArgumentParser: + parser = argparse.ArgumentParser(description=__doc__) + parser.add_argument("--country", default="GR") + parser.add_argument("--provider", action="append", dest="providers") + parser.add_argument("--streaming-max-requests", type=int, default=100) + parser.add_argument("--enrichment-max-titles", type=int, default=20_000) + parser.add_argument( + "--enrichment-mode", choices=("backfill", "incremental"), default="incremental" + ) + parser.add_argument("--include-watch-providers", action="store_true") + parser.add_argument("--skip-streaming", action="store_true") + parser.add_argument("--skip-enrichment", action="store_true") + parser.add_argument("--summary-output", type=Path) + return parser + + +def main() -> int: + logging.basicConfig(level=logging.INFO, format="%(asctime)s %(levelname)s %(message)s") + logging.getLogger("httpx").setLevel(logging.WARNING) + load_dotenv() + settings = Settings.from_env() + args = build_parser().parse_args() + providers = tuple(args.providers) if args.providers else settings.supported_providers + summary = run_full_refresh( + settings=settings, + country=args.country, + provider_keys=providers, + streaming_max_requests=args.streaming_max_requests, + enrichment_max_titles=args.enrichment_max_titles, + include_watch_providers=args.include_watch_providers, + enrichment_mode=args.enrichment_mode, + run_streaming=not args.skip_streaming, + run_enrichment=not args.skip_enrichment, + ) + rendered = json.dumps(summary, indent=2, default=str) + if args.summary_output: + args.summary_output.write_text(rendered + "\n", encoding="utf-8") + print(rendered) + return 0 + + +if __name__ == "__main__": # pragma: no cover + raise SystemExit(main()) diff --git a/ingestion/run.py b/ingestion/run.py index 46d6a50..9dab8eb 100644 --- a/ingestion/run.py +++ b/ingestion/run.py @@ -22,7 +22,7 @@ from dotenv import load_dotenv -from ingestion.core.lake import write_raw_batch +from ingestion.core.lake import RawRecord, write_raw_batch from ingestion.sources.tmdb.client import TMDBSource from ingestion.sources.tmdb.config import ( DEFAULT_COUNTRIES, @@ -194,6 +194,28 @@ def run( "metadata_records_written": metadata_written, "availability_records_written": availability_written, } + # The manifest is the publication contract for discovery snapshots. + # dbt only promotes provider/content combinations marked complete here; + # raw pages from interrupted or deliberately bounded runs stay auditable + # without accidentally becoming the live catalog. + write_raw_batch( + [ + RawRecord( + request_params={"run_id": run_id}, + payload={ + "run_id": run_id, + "discovery_complete": summary["discovery_complete"], + "queries": discovery_queries, + }, + ) + ], + lake_root=lake_root, + source="tmdb", + endpoint="discovery_manifest", + entity_type="catalog", + country="ALL", + run_id=run_id, + ) if runs: runs.succeed( run_id=run_id, @@ -229,6 +251,12 @@ def main() -> None: dest="countries", help="ISO2 country code; repeatable. Defaults to configured launch countries.", ) + parser.add_argument( + "--provider", + action="append", + dest="providers", + help="Configured provider key; repeatable. Defaults to all configured providers.", + ) parser.add_argument( "--enrich", action="store_true", @@ -251,13 +279,14 @@ def main() -> None: raise SystemExit("TMDB_API_KEY is not set. Copy .env.example to .env and fill it in.") countries = tuple(args.countries) if args.countries else settings.supported_regions - unknown_providers = set(settings.supported_providers) - PROVIDERS.keys() + selected_providers = tuple(args.providers) if args.providers else settings.supported_providers + unknown_providers = set(selected_providers) - PROVIDERS.keys() if unknown_providers: raise SystemExit( "SUPPORTED_PROVIDERS contains unknown provider keys: " + ", ".join(sorted(unknown_providers)) ) - providers = {key: PROVIDERS[key] for key in settings.supported_providers} + providers = {key: PROVIDERS[key] for key in selected_providers} run( lake_root=Path(args.lake_root), diff --git a/ingestion/sources/tmdb/client.py b/ingestion/sources/tmdb/client.py index 4bf433d..fe3210a 100644 --- a/ingestion/sources/tmdb/client.py +++ b/ingestion/sources/tmdb/client.py @@ -31,7 +31,11 @@ def __init__( base_url: str = TMDB_BASE_URL, ) -> None: self._api_key = api_key - self._client = http_client or RateLimitedClient(base_url=base_url, min_interval_seconds=0.3) + # Stay below TMDB's protective request ceiling. The client remains + # sequential and backs off automatically on 429/5xx. + self._client = http_client or RateLimitedClient( + base_url=base_url, min_interval_seconds=1 / 30 + ) def close(self) -> None: self._client.close() diff --git a/ingestion/tests/test_full_refresh.py b/ingestion/tests/test_full_refresh.py new file mode 100644 index 0000000..3d25356 --- /dev/null +++ b/ingestion/tests/test_full_refresh.py @@ -0,0 +1,85 @@ +from pathlib import Path +from types import SimpleNamespace + +from ingestion.full_refresh import run_full_refresh + + +def _settings(tmp_path: Path) -> SimpleNamespace: + return SimpleNamespace( + tmdb_api_key="tmdb-key", # pragma: allowlist secret + tmdb_base_url="https://tmdb.test", + streaming_availability_api_key="streaming-key", # pragma: allowlist secret + streaming_availability_base_url="https://streaming.test", + streaming_availability_monthly_cap=1000, + lake_root=tmp_path / "lake", + database_path=tmp_path / "ops.duckdb", + serving_database_path=tmp_path / "serving.duckdb", + 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, + ) + + +def test_full_refresh_runs_mandatory_discovery_and_two_atomic_publications( + tmp_path: Path, monkeypatch +) -> None: + publications = [] + monkeypatch.setattr( + "ingestion.full_refresh.run_tmdb_discovery", + lambda **kwargs: { + "run_id": "discovery", + "discovery_complete": True, + "api_request_count": 2, + "discover_pages_written": 2, + }, + ) + monkeypatch.setattr( + "ingestion.full_refresh.publish_warehouse", + lambda project, output: publications.append((project, output)), + ) + + summary = run_full_refresh( + settings=_settings(tmp_path), + country="gr", + provider_keys=("netflix",), + streaming_max_requests=100, + enrichment_max_titles=10, + include_watch_providers=False, + enrichment_mode="incremental", + run_streaming=False, + run_enrichment=False, + ) + + assert summary["country"] == "GR" + assert summary["stages"]["tmdb_discovery"]["run_id"] == "discovery" + assert len(publications) == 2 + + +def test_incomplete_discovery_is_never_published(tmp_path: Path, monkeypatch) -> None: + monkeypatch.setattr( + "ingestion.full_refresh.run_tmdb_discovery", + lambda **kwargs: {"run_id": "partial", "discovery_complete": False}, + ) + monkeypatch.setattr( + "ingestion.full_refresh.publish_warehouse", + lambda *args: (_ for _ in ()).throw(AssertionError("must not publish")), + ) + + try: + run_full_refresh( + settings=_settings(tmp_path), + country="GR", + provider_keys=("netflix",), + streaming_max_requests=100, + enrichment_max_titles=10, + include_watch_providers=False, + enrichment_mode="incremental", + run_streaming=False, + run_enrichment=False, + ) + except RuntimeError as error: + assert "incomplete" in str(error) + else: + raise AssertionError("incomplete discovery should fail") diff --git a/ingestion/tests/test_streaming_metadata_enrichment.py b/ingestion/tests/test_streaming_metadata_enrichment.py index bb181e6..72b6880 100644 --- a/ingestion/tests/test_streaming_metadata_enrichment.py +++ b/ingestion/tests/test_streaming_metadata_enrichment.py @@ -80,7 +80,13 @@ def fetch_availability(self, *, entity_type: str, source_title_id: int) -> RawRe FakeTMDBSource, ) - summary = run(lake_root=tmp_path, api_key="", country="GR", max_titles=1) + summary = run( + lake_root=tmp_path, + api_key="", + country="GR", + max_titles=1, + include_watch_providers=True, + ) assert summary["metadata_titles_requested"] == 1 assert summary["availability_titles_requested"] == 1 diff --git a/ingestion/tests/test_tmdb_run.py b/ingestion/tests/test_tmdb_run.py index 32fe4b4..3954e90 100644 --- a/ingestion/tests/test_tmdb_run.py +++ b/ingestion/tests/test_tmdb_run.py @@ -1,5 +1,8 @@ +import json from pathlib import Path +import duckdb + from ingestion.core.lake import RawRecord from ingestion.run import run @@ -79,6 +82,17 @@ def test_discovery_only_is_default_and_reports_complete_query(tmp_path: Path, mo "truncated_by_source_limit": False, } ] + manifests = list( + (tmp_path / "lake").glob("raw/source=tmdb/endpoint=discovery_manifest/**/*.parquet") + ) + assert len(manifests) == 1 + payload = json.loads( + duckdb.connect() + .execute("select cast(payload as json) from read_parquet(?)", [[str(manifests[0])]]) + .fetchone()[0] + ) + assert payload["run_id"] == summary["run_id"] + assert payload["discovery_complete"] is True def test_bounded_discovery_is_partial_and_enrichment_is_opt_in(tmp_path: Path, monkeypatch) -> None: diff --git a/warehouse/models/intermediate/int_content.sql b/warehouse/models/intermediate/int_content.sql index 92ffc1d..0475089 100644 --- a/warehouse/models/intermediate/int_content.sql +++ b/warehouse/models/intermediate/int_content.sql @@ -84,6 +84,21 @@ candidates as ( select * from streaming_candidates ), +latest_tmdb_volatile_fields as ( + select + tmdb_id, + content_type, + arg_max(tmdb_rating, source_updated_at) + filter (where tmdb_rating is not null) as tmdb_rating, + arg_max(vote_count, source_updated_at) + filter (where vote_count is not null) as vote_count, + arg_max(tmdb_popularity, source_updated_at) + filter (where tmdb_popularity is not null) as tmdb_popularity + from candidates + where metadata_source = 'tmdb' + group by 1, 2 +), + ranked as ( select *, @@ -98,28 +113,35 @@ ranked as ( partition by tmdb_id, content_type ) as last_observed_at from candidates +), + +selected as ( + select * + from ranked + where observation_rank = 1 ) select - tmdb_id, - content_type, - title, - original_title, - overview, - release_date, - release_year, - runtime_minutes, - episode_count, - season_count, - genre_ids, - original_language, - tmdb_rating, - vote_count, - tmdb_popularity, - poster_path, - backdrop_path, - metadata_source, - first_observed_at, - last_observed_at as source_updated_at -from ranked -where observation_rank = 1 + selected.tmdb_id, + selected.content_type, + selected.title, + selected.original_title, + selected.overview, + selected.release_date, + selected.release_year, + selected.runtime_minutes, + selected.episode_count, + selected.season_count, + selected.genre_ids, + selected.original_language, + coalesce(volatile.tmdb_rating, selected.tmdb_rating) as tmdb_rating, + coalesce(volatile.vote_count, selected.vote_count) as vote_count, + coalesce(volatile.tmdb_popularity, selected.tmdb_popularity) as tmdb_popularity, + selected.poster_path, + selected.backdrop_path, + selected.metadata_source, + selected.first_observed_at, + selected.last_observed_at as source_updated_at +from selected +left join latest_tmdb_volatile_fields as volatile + using (tmdb_id, content_type) diff --git a/warehouse/models/staging/_staging.yml b/warehouse/models/staging/_staging.yml index 05bd780..74b1c2f 100644 --- a/warehouse/models/staging/_staging.yml +++ b/warehouse/models/staging/_staging.yml @@ -1,6 +1,36 @@ version: 2 models: + - name: stg_tmdb_discovery_snapshots + description: Complete and partial TMDB discovery-query manifests used to select live snapshots. + meta: + owner: watchpulse + source: tmdb + grain: one row per run, region, provider, and content type + tests: + - dbt_utils.unique_combination_of_columns: + arguments: + combination_of_columns: [run_id, region, provider_key, content_type] + columns: + - {name: run_id, description: Ingestion run identifier., data_type: varchar, tests: [not_null]} + - {name: completed_at, description: Time the discovery manifest was retained., data_type: timestamptz, 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: content_type + description: "{{ doc('col_content_type') }}" + data_type: varchar + tests: + - not_null + - accepted_values: + arguments: + values: [movie, tv] + - {name: upstream_total_pages, description: Pages reported by TMDB., data_type: integer, tests: [not_null]} + - {name: expected_pages, description: Pages required within the source ceiling., data_type: integer, tests: [not_null]} + - {name: pages_fetched, description: Pages retained by WatchPulse., data_type: integer, tests: [not_null]} + - {name: total_results, description: Results reported by TMDB., data_type: integer, tests: [not_null]} + - {name: is_complete, description: Whether all expected pages were retained., data_type: boolean, tests: [not_null]} + - {name: truncated_by_source_limit, description: Whether the TMDB 500-page ceiling prevented completeness., data_type: boolean, tests: [not_null]} + - name: stg_tmdb_watch_providers description: "{{ doc('stg_tmdb_watch_providers') }}" meta: diff --git a/warehouse/models/staging/stg_tmdb_discovery.sql b/warehouse/models/staging/stg_tmdb_discovery.sql index 5992817..0ac721a 100644 --- a/warehouse/models/staging/stg_tmdb_discovery.sql +++ b/warehouse/models/staging/stg_tmdb_discovery.sql @@ -3,6 +3,7 @@ with raw_pages as ( select filename as source_file, + regexp_extract(filename, 'part-([^-]+)-', 1) as run_id, cast(fetched_at as timestamptz) as fetched_at, entity_type as content_type, upper(country) as region, @@ -16,6 +17,23 @@ with raw_pages as ( ) ), +latest_complete_snapshots as ( + select run_id, region, provider_key, content_type + from {{ ref('stg_tmdb_discovery_snapshots') }} + where is_complete and not truncated_by_source_limit + qualify row_number() over ( + partition by region, provider_key, content_type + order by completed_at desc, run_id desc + ) = 1 +), + +published_pages as ( + select raw_pages.* + from raw_pages + inner join latest_complete_snapshots + using (run_id, region, provider_key, content_type) +), + flattened as ( select cast(json_extract_string(result.value, '$.id') as bigint) as tmdb_id, @@ -46,7 +64,7 @@ flattened as ( json_extract_string(result.value, '$.backdrop_path') as backdrop_path, fetched_at as source_updated_at, source_file - from raw_pages, + from published_pages, json_each(json_extract(payload, '$.results')) as result ) diff --git a/warehouse/models/staging/stg_tmdb_discovery_snapshots.sql b/warehouse/models/staging/stg_tmdb_discovery_snapshots.sql new file mode 100644 index 0000000..1441667 --- /dev/null +++ b/warehouse/models/staging/stg_tmdb_discovery_snapshots.sql @@ -0,0 +1,38 @@ +{% set lake_root = var('lake_root', '../data/lake') %} + +with manifests as ( + select + cast(fetched_at as timestamptz) as completed_at, + cast(payload as json) as payload + from read_parquet( + '{{ lake_root }}/raw/source=tmdb/endpoint=discovery_manifest/entity_type=catalog/country=ALL/date=*/*.parquet', + union_by_name = true + ) +), + +queries as ( + select + json_extract_string(payload, '$.run_id') as run_id, + completed_at, + upper(json_extract_string(query.value, '$.country')) as region, + json_extract_string(query.value, '$.provider_key') as provider_key, + json_extract_string(query.value, '$.content_type') as content_type, + try_cast(json_extract_string(query.value, '$.upstream_total_pages') as integer) + as upstream_total_pages, + try_cast(json_extract_string(query.value, '$.expected_pages') as integer) + as expected_pages, + try_cast(json_extract_string(query.value, '$.pages_fetched') as integer) + as pages_fetched, + try_cast(json_extract_string(query.value, '$.total_results') as integer) + as total_results, + coalesce(try_cast(json_extract_string(query.value, '$.complete') as boolean), false) + as is_complete, + coalesce( + try_cast(json_extract_string(query.value, '$.truncated_by_source_limit') as boolean), + false + ) as truncated_by_source_limit + from manifests, json_each(payload, '$.queries') as query +) + +select * +from queries diff --git a/warehouse/models/staging/stg_tmdb_metadata.sql b/warehouse/models/staging/stg_tmdb_metadata.sql index 7b39c37..0ec5365 100644 --- a/warehouse/models/staging/stg_tmdb_metadata.sql +++ b/warehouse/models/staging/stg_tmdb_metadata.sql @@ -55,8 +55,8 @@ select overview, release_date, coalesce(nullif(movie_runtime, 0), nullif(series_runtime, 0)) as runtime_minutes, - episode_count, - season_count, + nullif(episode_count, 0) as episode_count, + nullif(season_count, 0) as 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,