cinematch / app /engine /corpus.py
Alluri Lakshman Narendra
Deploy CineMatch backend (FastAPI + hybrid retrieval)
22b7d63
Raw
History Blame Contribute Delete
1.38 kB
"""corpus.py — loads corpus.json and exposes movie records + lookups."""
from __future__ import annotations
import json
from pathlib import Path
from app.schemas import Movie
_DISPLAY_FIELDS = (
"id", "tmdb_id", "title", "year", "genres", "overview", "tagline",
"poster_url", "vote_average", "vote_count", "popularity", "language",
"cast", "director", "keywords", "runtime",
)
class Corpus:
def __init__(self, path: Path) -> None:
if not path.exists():
raise FileNotFoundError(
f"corpus not found at {path}. Run: python scripts/build_corpus.py"
)
self.records: list[dict] = json.loads(Path(path).read_text())
self.by_id: dict[str, int] = {r["id"]: i for i, r in enumerate(self.records)}
self.size = len(self.records)
def __len__(self) -> int:
return self.size
def index_of(self, movie_id: str) -> int | None:
return self.by_id.get(movie_id)
def semantic_texts(self) -> list[str]:
return [r["semantic_text"] for r in self.records]
def lexical_texts(self) -> list[str]:
return [r["lexical_text"] for r in self.records]
def to_movie(self, idx: int) -> Movie:
r = self.records[idx]
return Movie(**{k: r.get(k) for k in _DISPLAY_FIELDS})
def record(self, idx: int) -> dict:
return self.records[idx]