"""Safely rebuild deployable recommendation artifacts outside the live Space. The script is intentionally publication-agnostic: it creates and validates a complete local ``data/processed`` snapshot. A CI workflow can then upload that snapshot to the Hugging Face Space in one commit, so users continue to receive the old, valid version while the refresh is running. """ from __future__ import annotations import argparse from datetime import datetime, timezone import hashlib import json from pathlib import Path import subprocess import sys DEFAULT_ARTIFACTS = ( "movie_catalog.csv", "catalog_metrics.csv", "item_neighbors.csv", "item_cf_metrics.csv", "item_cf_eval.csv", "semantic_index.npz", "ranking_eval.csv", "hybrid_weights.json", ) def main() -> None: parser = argparse.ArgumentParser( description="Rebuild and validate MovieLens/TMDb recommendation artifacts for a manual deployment." ) parser.add_argument("--variant", choices=("small",), default="small") parser.add_argument("--min-year", type=int, default=2018) parser.add_argument("--pages-per-year", type=int, default=50) parser.add_argument("--enrich-limit-per-year", type=int, default=100) parser.add_argument("--tmdb-sleep", type=float, default=0.30) parser.add_argument("--enrich-sleep", type=float, default=0.15) parser.add_argument( "--max-tmdb-requests", type=int, default=2000, help="Fail before making requests when the worst-case TMDb call count exceeds this limit.", ) parser.add_argument( "--existing-manifest", type=Path, help="Previously deployed manifest used to skip a refresh that is not due yet.", ) parser.add_argument("--max-age-days", type=int, default=90) parser.add_argument("--force", action="store_true", help="Refresh even if the manifest is recent.") parser.add_argument("--state-path", type=Path, help="Optional JSON file recording whether a refresh ran.") parser.add_argument("--dry-run", action="store_true", help="Print the planned commands without changing files.") args = parser.parse_args() validate_arguments(args) if not args.force and is_recent(args.existing_manifest, args.max_age_days): message = f"Skipping refresh: deployed artifacts are newer than {args.max_age_days} days." print(message) write_state(args.state_path, updated=False, message=message) return current_year = datetime.now(timezone.utc).year estimated_requests = 1 + (current_year - args.min_year + 1) * ( args.pages_per_year + args.enrich_limit_per_year ) if estimated_requests > args.max_tmdb_requests: raise ValueError( f"Worst-case TMDb requests ({estimated_requests}) exceed --max-tmdb-requests " f"({args.max_tmdb_requests}). Reduce pages/years or explicitly raise the guard." ) print( f"TMDb safety plan: at most {estimated_requests} requests, " f"discover paced at <= {1 / args.tmdb_sleep:.1f} req/s and details at <= {1 / args.enrich_sleep:.1f} req/s." ) processed = Path("data/processed") commands = build_commands(args, current_year, processed) for command in commands: print("+", " ".join(command)) if not args.dry_run: subprocess.run(command, check=True) if args.dry_run: write_state(args.state_path, updated=False, message="Dry run; no artifacts were rebuilt.") return write_manifest(processed, args, current_year) write_state(args.state_path, updated=True, message="Artifacts rebuilt and validated.") print(f"Wrote {processed / 'update_manifest.json'}") def validate_arguments(args: argparse.Namespace) -> None: current_year = datetime.now(timezone.utc).year if args.min_year < 1888 or args.min_year > current_year: raise ValueError(f"--min-year must be between 1888 and {current_year}.") if args.pages_per_year < 1 or args.enrich_limit_per_year < 0: raise ValueError("--pages-per-year must be positive and --enrich-limit-per-year cannot be negative.") if args.tmdb_sleep < 0.1 or args.enrich_sleep < 0.1: raise ValueError("Refresh pacing must be at least 0.1 seconds per TMDb request.") if args.max_tmdb_requests < 1 or args.max_age_days < 0: raise ValueError("--max-tmdb-requests must be positive and --max-age-days cannot be negative.") def is_recent(manifest_path: Path | None, max_age_days: int) -> bool: if manifest_path is None or not manifest_path.exists(): return False try: payload = json.loads(manifest_path.read_text(encoding="utf-8")) generated_at = datetime.fromisoformat(str(payload["generated_at"]).replace("Z", "+00:00")) except (KeyError, TypeError, ValueError, json.JSONDecodeError, OSError): print(f"Ignoring unreadable manifest: {manifest_path}") return False if generated_at.tzinfo is None: generated_at = generated_at.replace(tzinfo=timezone.utc) age_days = max((datetime.now(timezone.utc) - generated_at).total_seconds() / 86_400, 0) print(f"Deployed artifact age: {age_days:.1f} days.") return age_days < max_age_days def build_commands(args: argparse.Namespace, current_year: int, processed: Path) -> list[list[str]]: python = sys.executable movie_lens_catalog = processed / "movielens_small_catalog.csv" tmdb_catalog = processed / "tmdb_broad_recent_catalog.csv" catalog = processed / "movie_catalog.csv" ratings = Path("data/raw/ml-latest-small/ratings.csv") return [ [ python, "scripts/prepare_movielens.py", "--variant", args.variant, "--catalog-output-path", str(movie_lens_catalog), ], [ python, "scripts/build_tmdb_catalog.py", "--min-year", str(args.min_year), "--max-year", str(current_year), "--pages", str(args.pages_per_year), "--per-year", "--sleep", str(args.tmdb_sleep), "--output-path", str(tmdb_catalog), ], [ python, "scripts/enrich_tmdb_metadata.py", "--input-path", str(tmdb_catalog), "--output-path", str(tmdb_catalog), "--limit-per-year", str(args.enrich_limit_per_year), "--sleep", str(args.enrich_sleep), ], [ python, "scripts/merge_catalogs.py", "--inputs", str(movie_lens_catalog), str(tmdb_catalog), "--output-path", str(catalog), ], [python, "scripts/train_item_cf.py", "--ratings-path", str(ratings), "--catalog-path", str(catalog)], [python, "scripts/build_semantic_index.py", "--catalog-path", str(catalog)], [ python, "scripts/evaluate_item_cf.py", "--ratings-path", str(ratings), "--neighbors-path", str(processed / "item_neighbors.csv"), "--k", "10", ], [ python, "scripts/evaluate_recommenders.py", "--ratings-path", str(ratings), "--catalog-path", str(catalog), "--semantic-path", str(processed / "semantic_index.npz"), "--k", "10", ], [python, "scripts/validate_runtime.py", "--require-learned-embeddings"], ] def write_manifest(processed: Path, args: argparse.Namespace, current_year: int) -> None: missing = [name for name in DEFAULT_ARTIFACTS if not (processed / name).exists()] if missing: raise FileNotFoundError(f"Refresh did not produce expected artifacts: {', '.join(missing)}") payload = { "schema_version": 1, "generated_at": datetime.now(timezone.utc).replace(microsecond=0).isoformat().replace("+00:00", "Z"), "movie_lens": {"variant": args.variant}, "tmdb": { "min_year": args.min_year, "max_year": current_year, "pages_per_year": args.pages_per_year, "enrich_limit_per_year": args.enrich_limit_per_year, "max_requests": args.max_tmdb_requests, }, "artifacts": {name: describe_file(processed / name) for name in DEFAULT_ARTIFACTS}, } (processed / "update_manifest.json").write_text(json.dumps(payload, indent=2, sort_keys=True) + "\n", encoding="utf-8") def describe_file(path: Path) -> dict[str, int | str]: digest = hashlib.sha256() with path.open("rb") as handle: for chunk in iter(lambda: handle.read(1024 * 1024), b""): digest.update(chunk) return {"bytes": path.stat().st_size, "sha256": digest.hexdigest()} def write_state(path: Path | None, updated: bool, message: str) -> None: if path is None: return path.parent.mkdir(parents=True, exist_ok=True) path.write_text(json.dumps({"updated": updated, "message": message}) + "\n", encoding="utf-8") if __name__ == "__main__": main()