Spaces:
Sleeping
Sleeping
| """Mechanical first-pass topic <-> syntopicon-term mapping. | |
| Reads index.json (skipping syntopicon-kind topics) and a slim term registry, | |
| maps each topic to its term via case-insensitive alias matching in words[], | |
| and writes topic-to-entry.json (confirmed) + _unresolved.json (gaps). | |
| Atomic writes via tempfile + rename. Deterministic output (sorted keys). | |
| """ | |
| from __future__ import annotations | |
| import json | |
| import os | |
| from dataclasses import dataclass | |
| from datetime import datetime, timezone | |
| from pathlib import Path | |
| _NON_CONTENT_KINDS = {"syntopicon_entry", "syntopicon_registry"} | |
| class LinkStats: | |
| topics_total: int | |
| topics_resolved: int | |
| topics_unresolved: int | |
| def _topic_kinds(topic_record: dict) -> set[str]: | |
| return {f.get("kind", "") for f in topic_record.get("files", [])} | |
| def _candidates_for_topic(topic: str, registry: dict) -> list[str]: | |
| """Return list of term slugs whose words[] contains a case-insensitive match for topic.""" | |
| needle = topic.strip().lower() | |
| out: list[str] = [] | |
| for slug, term in registry.get("terms", {}).items(): | |
| for word in term.get("words", []) or []: | |
| if word.strip().lower() == needle: | |
| out.append(slug) | |
| break | |
| return sorted(out) | |
| def _atomic_write(path: Path, data: dict) -> None: | |
| tmp = path.with_suffix(path.suffix + ".tmp") | |
| tmp.write_text(json.dumps(data, indent=2, sort_keys=True, ensure_ascii=False) + "\n", encoding="utf-8") | |
| os.replace(tmp, path) | |
| def run(index_path: Path, registry_path: Path, output_dir: Path) -> LinkStats: | |
| """Generate topic-to-entry.json + _unresolved.json from index + registry.""" | |
| index = json.loads(index_path.read_text(encoding="utf-8")) | |
| registry = json.loads(registry_path.read_text(encoding="utf-8")) | |
| output_dir.mkdir(parents=True, exist_ok=True) | |
| mappings: dict[str, dict] = {} | |
| unresolved: dict[str, dict] = {} | |
| now = datetime.now(timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ") | |
| for topic, record in sorted(index.get("topics", {}).items()): | |
| kinds = _topic_kinds(record) | |
| if kinds.issubset(_NON_CONTENT_KINDS): | |
| # Skip topics that ONLY come from syntopicon entries themselves | |
| continue | |
| candidates = _candidates_for_topic(topic, registry) | |
| if len(candidates) == 1: | |
| mappings[topic] = {"entry": candidates[0], "confidence": "exact_alias"} | |
| elif len(candidates) > 1: | |
| unresolved[topic] = {"reason": "multiple_candidates", "candidates": candidates} | |
| else: | |
| unresolved[topic] = {"reason": "no_alias_match", "candidates": []} | |
| _atomic_write(output_dir / "topic-to-entry.json", { | |
| "schema_version": 1, | |
| "generated_utc": now, | |
| "mappings": mappings, | |
| }) | |
| _atomic_write(output_dir / "_unresolved.json", { | |
| "schema_version": 1, | |
| "generated_utc": now, | |
| "unresolved": unresolved, | |
| }) | |
| return LinkStats( | |
| topics_total=len(mappings) + len(unresolved), | |
| topics_resolved=len(mappings), | |
| topics_unresolved=len(unresolved), | |
| ) | |