asa / scripts /lib /catalog_loader.py
defenestrate's picture
ASA Space: squashed history (instructions and settings live in the private pack)
526950e
Raw History Blame Contribute Delete
2.82 kB
"""Pure offline catalog loader. No network."""
from __future__ import annotations
import json
import re
from dataclasses import dataclass, field
from pathlib import Path
from scripts.lib.term_extraction import build_regex
@dataclass
class Catalog:
records: list[dict]
term_to_docs: dict[str, list[str]]
concept_graph: dict[str, dict]
embedding_model: str
embedding_dim: int
term_regex: re.Pattern
registry: dict = field(default_factory=dict)
chunk_to_referencing_cards: dict[str, list[dict]] = field(default_factory=dict)
def load_catalog(derived_dir: Path, *, repo_root: Path | None = None) -> Catalog:
"""Load student-chat catalog and supporting layers from derived/ artifacts.
repo_root defaults to derived_dir.parent.parent (assumes <repo>/derived/student-chat/
layout); pass explicitly for non-standard paths.
"""
if repo_root is None:
repo_root = derived_dir.parent.parent
catalog_path = derived_dir / "catalog.jsonl"
if not catalog_path.is_file():
raise FileNotFoundError(f"catalog.jsonl not found at {catalog_path}")
records = [json.loads(line) for line in catalog_path.read_text(encoding="utf-8").splitlines() if line.strip()]
if not records:
raise ValueError(f"catalog.jsonl at {catalog_path} is empty")
term_to_docs: dict[str, list[str]] = {}
if (derived_dir / "term-to-docs.json").is_file():
term_to_docs = json.loads((derived_dir / "term-to-docs.json").read_text(encoding="utf-8"))
concept_graph: dict[str, dict] = {}
if (derived_dir / "concept-graph.json").is_file():
concept_graph = json.loads((derived_dir / "concept-graph.json").read_text(encoding="utf-8"))
chunk_to_referencing_cards: dict[str, list[dict]] = {}
rev_path = derived_dir / "chunk_to_referencing_cards.json"
if rev_path.is_file():
chunk_to_referencing_cards = json.loads(rev_path.read_text(encoding="utf-8"))
cfg_path = derived_dir / "build-config.json"
if not cfg_path.is_file():
raise FileNotFoundError(f"build-config.json not found at {cfg_path}")
cfg = json.loads(cfg_path.read_text(encoding="utf-8"))
embedding_model = cfg["embedding_model"]
embedding_dim = cfg["embedding_dimensions"]
reg_path = repo_root / "content" / "syntopicon-mgrecon780" / "_registry.json"
registry: dict = {}
if reg_path.is_file():
registry = json.loads(reg_path.read_text(encoding="utf-8"))
term_regex = build_regex(registry) if registry else re.compile(r"$^")
return Catalog(
records=records, term_to_docs=term_to_docs, concept_graph=concept_graph,
embedding_model=embedding_model, embedding_dim=embedding_dim,
term_regex=term_regex, registry=registry,
chunk_to_referencing_cards=chunk_to_referencing_cards,
)