Spaces:
Sleeping
Sleeping
| """Walk content/, apply classification rules, emit index/index.json.""" | |
| from __future__ import annotations | |
| import fnmatch | |
| import hashlib | |
| import json | |
| from datetime import datetime, timezone | |
| from pathlib import Path | |
| import yaml | |
| from lib import classify | |
| from lib.paths import repo_root | |
| SCHEMA_VERSION = 1 | |
| class UnmatchedFilesError(RuntimeError): | |
| pass | |
| def _load_rules() -> dict: | |
| return yaml.safe_load((repo_root() / "config" / "course.yaml").read_text(encoding="utf-8")) | |
| def _hash_file(p: Path) -> tuple[str, int]: | |
| data = p.read_bytes() | |
| return hashlib.sha256(data).hexdigest(), len(data) | |
| def _walk_content(content_dir: Path) -> list[Path]: | |
| return sorted( | |
| (p for p in content_dir.rglob("*") if p.is_file() and p.name != ".gitkeep"), | |
| key=lambda p: p.as_posix(), | |
| ) | |
| def generate(source_commit: str, *, rules: dict | None = None, root: Path | None = None) -> dict: | |
| """Generate the full index.json structure (does not write to disk).""" | |
| if rules is None: | |
| rules = _load_rules() | |
| if root is None: | |
| root = repo_root() | |
| content_dir = root / "content" | |
| topics: dict[str, dict] = {} | |
| unmatched: list[str] = [] | |
| for fp in _walk_content(content_dir): | |
| rel_posix = fp.relative_to(root).as_posix() | |
| cls = classify.classify_path(rel_posix, rules) | |
| if cls is None: | |
| unmatched.append(rel_posix) | |
| continue | |
| sha, size = _hash_file(fp) | |
| record = { | |
| "path": rel_posix, | |
| "kind": cls.kind, | |
| "topic": cls.topic, | |
| "sha256": sha, | |
| "size_bytes": size, | |
| "visibility": cls.visibility, | |
| "student_access": cls.student_access, | |
| "source_system": _source_system_for(cls.kind), | |
| "release_status": cls.release_status, | |
| } | |
| if cls.order is not None: | |
| record["order"] = cls.order | |
| if cls.lecture_date is not None: | |
| record["lecture_date"] = cls.lecture_date | |
| topic_entry = topics.setdefault(cls.topic, {"title": _titlecase(cls.topic), "files": []}) | |
| if cls.kind == "syntopicon_entry": | |
| topic_entry.setdefault("syntopicon_entries", []).append({ | |
| "id": f"syntopicon:{fp.stem}", | |
| "path": rel_posix, | |
| "sha256": sha, | |
| "source_system": "syntopicon", | |
| }) | |
| else: | |
| topic_entry["files"].append(record) | |
| waivers = rules.get("diagnostic_waivers") or {"unmatched_files": [], "topic_collisions": []} | |
| waiver_patterns = waivers.get("unmatched_files") or [] | |
| blocking_unmatched = [ | |
| u for u in unmatched | |
| if not any(fnmatch.fnmatch(u, p) for p in waiver_patterns) | |
| ] | |
| if blocking_unmatched: | |
| raise UnmatchedFilesError( | |
| f"unmatched files (waivers must be added to course.yaml diagnostic_waivers if intentional): " | |
| f"{blocking_unmatched}" | |
| ) | |
| for t in topics.values(): | |
| t["files"].sort(key=lambda r: (r.get("order", 1_000_000), r["path"])) | |
| if "syntopicon_entries" in t: | |
| t["syntopicon_entries"].sort(key=lambda r: r["path"]) | |
| canon = json.dumps(topics, sort_keys=True, separators=(",", ":")).encode("utf-8") | |
| fingerprint = "sha256:" + hashlib.sha256(canon).hexdigest() | |
| return { | |
| "schema_version": SCHEMA_VERSION, | |
| "generator_version": rules.get("index_generator_version", "1.0.0"), | |
| "generated_utc": datetime.now(timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ"), | |
| "source_commit": source_commit, | |
| "content_fingerprint": fingerprint, | |
| "topics": topics, | |
| "diagnostics": { | |
| "unmatched_files": unmatched, | |
| "topic_collisions": [], | |
| }, | |
| } | |
| def write_index(data: dict) -> Path: | |
| """Write index/index.json with stable ordering.""" | |
| out = repo_root() / "index" / "index.json" | |
| out.parent.mkdir(parents=True, exist_ok=True) | |
| out.write_text(json.dumps(data, indent=2, sort_keys=True), encoding="utf-8") | |
| return out | |
| def _source_system_for(kind: str) -> str: | |
| """Map a record kind to its source-state key.""" | |
| mapping = { | |
| "lecture_note": "togo", | |
| "slide": "togo", | |
| "slide_answer": "togo", | |
| "question": "togo", | |
| "course_meta": "togo", | |
| "transcript": "lectures_jsonl", | |
| "lecture_session": "lectures_jsonl", | |
| "syntopicon_entry": "syntopicon", | |
| "syntopicon_registry": "syntopicon", | |
| } | |
| if kind not in mapping: | |
| raise ValueError(f"no source system for kind: {kind!r}") | |
| return mapping[kind] | |
| def _titlecase(topic: str) -> str: | |
| return " ".join(part.capitalize() for part in topic.replace("_", " ").split()) | |