| """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() |
|
|
