| """Deterministic loader for the GCMD Science Keyword hierarchy.""" |
|
|
| from __future__ import annotations |
|
|
| import hashlib |
| import json |
| from pathlib import Path |
| from typing import Any |
|
|
| from gcmd_classifier.errors import ( |
| DuplicateCanonicalPathError, |
| DuplicateUUIDError, |
| InvalidHierarchyTransitionError, |
| MalformedHierarchyError, |
| ) |
| from gcmd_classifier.models import CanonicalConceptRecord |
| from gcmd_classifier.vocabulary.index import VocabularyIndex |
|
|
| ROOT_LEVEL = "Category" |
| ROOT_NAME = "EARTH SCIENCE" |
| CANONICAL_PATH_SEPARATOR = " > " |
|
|
| _ALLOWED_LEVELS = { |
| ROOT_LEVEL, |
| "Topic", |
| "Term", |
| "Variable_Level_1", |
| "Variable_Level_2", |
| "Variable_Level_3", |
| } |
| _VALID_CHILD_LEVELS = { |
| ROOT_LEVEL: "Topic", |
| "Topic": "Term", |
| "Term": "Variable_Level_1", |
| "Variable_Level_1": "Variable_Level_2", |
| "Variable_Level_2": "Variable_Level_3", |
| "Variable_Level_3": None, |
| } |
| _ASSIGNABLE_LEVELS = { |
| "Topic", |
| "Term", |
| "Variable_Level_1", |
| "Variable_Level_2", |
| "Variable_Level_3", |
| } |
|
|
|
|
| def calculate_file_hash(path: str | Path) -> str: |
| """Return a reproducible SHA-256 hash of a vocabulary file's raw bytes.""" |
| return hashlib.sha256(Path(path).read_bytes()).hexdigest() |
|
|
|
|
| def load_vocabulary(path: str | Path) -> VocabularyIndex: |
| """Load a GCMD hierarchy JSON file and return a canonical vocabulary index.""" |
| hierarchy_path = Path(path) |
| raw_bytes = hierarchy_path.read_bytes() |
| try: |
| data = json.loads(raw_bytes) |
| except json.JSONDecodeError as exc: |
| raise MalformedHierarchyError(f"Invalid JSON in vocabulary file: {hierarchy_path}") from exc |
| vocabulary_version = hashlib.sha256(raw_bytes).hexdigest() |
| return build_vocabulary_index(data, vocabulary_version=vocabulary_version) |
|
|
|
|
| def build_vocabulary_index(data: Any, vocabulary_version: str) -> VocabularyIndex: |
| """Build a canonical vocabulary index from an already-loaded hierarchy object.""" |
| builder = _VocabularyIndexBuilder(vocabulary_version=vocabulary_version) |
| return builder.build(data) |
|
|
|
|
| class _VocabularyIndexBuilder: |
| def __init__(self, vocabulary_version: str) -> None: |
| self.vocabulary_version = vocabulary_version |
| self.records_by_uuid: dict[str, CanonicalConceptRecord] = {} |
| self.records_by_path: dict[str, CanonicalConceptRecord] = {} |
| self.parent_by_uuid: dict[str, str | None] = {} |
| self.children_by_uuid: dict[str | None, tuple[str, ...]] = {} |
| self.topic_uuids: list[str] = [] |
| self.terms_by_topic_uuid: dict[str, tuple[str, ...]] = {} |
| self.variables_by_parent_uuid: dict[str, tuple[str, ...]] = {} |
|
|
| def build(self, data: Any) -> VocabularyIndex: |
| if not isinstance(data, dict): |
| raise MalformedHierarchyError("Hierarchy root must be a JSON object.") |
| root_level = self._required_string(data, "level", path="<root>") |
| root_name = self._required_string(data, "name", path="<root>") |
| if root_level != ROOT_LEVEL or root_name != ROOT_NAME: |
| raise MalformedHierarchyError( |
| f"Hierarchy root must be {ROOT_LEVEL!r} named {ROOT_NAME!r}; " |
| f"found level={root_level!r}, name={root_name!r}." |
| ) |
| if data.get("UUID"): |
| raise MalformedHierarchyError("The known root Category must not define a UUID.") |
|
|
| root_children = self._children(data, path=ROOT_NAME) |
| self._validate_children_transition(root_level, root_children, path=ROOT_NAME) |
| topic_uuids = self._walk_children( |
| root_children, |
| parent_level=root_level, |
| parent_uuid=None, |
| parent_name=root_name, |
| path_components=(), |
| topic=None, |
| term=None, |
| source_path=root_name, |
| ) |
| self.topic_uuids = list(topic_uuids) |
| self.children_by_uuid[None] = topic_uuids |
|
|
| return VocabularyIndex( |
| records_by_uuid=self.records_by_uuid, |
| records_by_path=self.records_by_path, |
| parent_by_uuid=self.parent_by_uuid, |
| children_by_uuid=self.children_by_uuid, |
| topic_uuids=tuple(self.topic_uuids), |
| terms_by_topic_uuid=self.terms_by_topic_uuid, |
| variables_by_parent_uuid=self.variables_by_parent_uuid, |
| vocabulary_version=self.vocabulary_version, |
| root_level=root_level, |
| root_name=root_name, |
| ) |
|
|
| def _walk_children( |
| self, |
| children: list[Any], |
| *, |
| parent_level: str, |
| parent_uuid: str | None, |
| parent_name: str, |
| path_components: tuple[str, ...], |
| topic: str | None, |
| term: str | None, |
| source_path: str, |
| ) -> tuple[str, ...]: |
| child_uuids: list[str] = [] |
| for child in children: |
| child_uuid = self._walk_node( |
| child, |
| parent_level=parent_level, |
| parent_uuid=parent_uuid, |
| parent_name=parent_name, |
| path_components=path_components, |
| topic=topic, |
| term=term, |
| source_path=source_path, |
| ) |
| child_uuids.append(child_uuid) |
| return tuple(child_uuids) |
|
|
| def _walk_node( |
| self, |
| node: Any, |
| *, |
| parent_level: str, |
| parent_uuid: str | None, |
| parent_name: str, |
| path_components: tuple[str, ...], |
| topic: str | None, |
| term: str | None, |
| source_path: str, |
| ) -> str: |
| if not isinstance(node, dict): |
| raise MalformedHierarchyError( |
| f"Hierarchy node under {source_path!r} must be an object." |
| ) |
|
|
| level = self._required_string(node, "level", path=source_path) |
| name = self._required_string(node, "name", path=source_path) |
| node_path = f"{source_path} / {name}" |
| self._validate_level(level, path=node_path) |
| self._validate_transition(parent_level, level, path=node_path) |
|
|
| uuid = self._required_uuid(node, level=level, path=node_path) |
| definition = self._optional_string(node, "definition", path=node_path) |
| children = self._children(node, path=node_path) |
| self._validate_children_transition(level, children, path=node_path) |
|
|
| next_path_components = (*path_components, name) |
| canonical_path = CANONICAL_PATH_SEPARATOR.join(next_path_components) |
| next_topic = name if level == "Topic" else topic |
| next_term = name if level == "Term" else term |
|
|
| direct_child_uuids = self._walk_children( |
| children, |
| parent_level=level, |
| parent_uuid=uuid, |
| parent_name=name, |
| path_components=next_path_components, |
| topic=next_topic, |
| term=next_term, |
| source_path=node_path, |
| ) |
|
|
| if uuid in self.records_by_uuid: |
| existing = self.records_by_uuid[uuid] |
| raise DuplicateUUIDError( |
| f"Duplicate UUID {uuid!r} at {canonical_path!r}; " |
| f"already used by {existing.canonical_path!r}." |
| ) |
| existing_path = self.records_by_path.get(canonical_path) |
| if existing_path is not None and existing_path.UUID != uuid: |
| raise DuplicateCanonicalPathError( |
| f"Canonical path {canonical_path!r} is assigned to both " |
| f"{existing_path.UUID!r} and {uuid!r}." |
| ) |
|
|
| record = CanonicalConceptRecord( |
| UUID=uuid, |
| name=name, |
| level=level, |
| topic=next_topic or name, |
| term=next_term, |
| path_components=next_path_components, |
| canonical_path=canonical_path, |
| parent_uuid=parent_uuid, |
| parent_name=parent_name, |
| child_uuids=direct_child_uuids, |
| has_children=bool(children), |
| assignable=True, |
| definition=definition, |
| vocabulary_version=self.vocabulary_version, |
| ) |
| self.records_by_uuid[uuid] = record |
| self.records_by_path[canonical_path] = record |
| self.parent_by_uuid[uuid] = parent_uuid |
| self.children_by_uuid[uuid] = direct_child_uuids |
| if level == "Topic": |
| self.terms_by_topic_uuid[uuid] = direct_child_uuids |
| elif level != "Variable_Level_3": |
| self.variables_by_parent_uuid[uuid] = direct_child_uuids |
| return uuid |
|
|
| def _required_string(self, node: dict[str, Any], field: str, *, path: str) -> str: |
| value = node.get(field) |
| if not isinstance(value, str) or not value.strip(): |
| raise MalformedHierarchyError(f"Node at {path!r} must contain non-empty {field!r}.") |
| return value |
|
|
| def _optional_string(self, node: dict[str, Any], field: str, *, path: str) -> str | None: |
| if field not in node: |
| return None |
| value = node[field] |
| if value is None: |
| return None |
| if not isinstance(value, str): |
| raise MalformedHierarchyError(f"Optional field {field!r} at {path!r} must be a string.") |
| return value |
|
|
| def _required_uuid(self, node: dict[str, Any], *, level: str, path: str) -> str: |
| value = node.get("UUID") |
| if not isinstance(value, str) or not value.strip(): |
| raise MalformedHierarchyError( |
| f"UUID-less hierarchy node at {path!r} is not allowed for level {level!r}." |
| ) |
| if level in {"Topic", "Term"} and not value: |
| raise MalformedHierarchyError(f"{level} at {path!r} must contain a UUID.") |
| return value |
|
|
| def _children(self, node: dict[str, Any], *, path: str) -> list[Any]: |
| children = node.get("children", []) |
| if children is None: |
| return [] |
| if not isinstance(children, list): |
| raise MalformedHierarchyError(f"Children for node at {path!r} must be a list.") |
| return children |
|
|
| def _validate_level(self, level: str, *, path: str) -> None: |
| if level not in _ALLOWED_LEVELS: |
| raise MalformedHierarchyError(f"Unsupported hierarchy level {level!r} at {path!r}.") |
| if level == ROOT_LEVEL: |
| raise MalformedHierarchyError("Category is only allowed as the known hierarchy root.") |
|
|
| def _validate_transition(self, parent_level: str, child_level: str, *, path: str) -> None: |
| expected = _VALID_CHILD_LEVELS.get(parent_level) |
| if child_level != expected: |
| raise InvalidHierarchyTransitionError( |
| f"Invalid hierarchy transition at {path!r}: {parent_level!r} -> {child_level!r}; " |
| f"expected child level {expected!r}." |
| ) |
|
|
| def _validate_children_transition(self, level: str, children: list[Any], *, path: str) -> None: |
| expected = _VALID_CHILD_LEVELS[level] |
| if expected is None and children: |
| raise InvalidHierarchyTransitionError( |
| f"Node at {path!r} has children, but {level!r} cannot have child concepts." |
| ) |
| for child in children: |
| if not isinstance(child, dict): |
| raise MalformedHierarchyError(f"Child under {path!r} must be an object.") |
| child_level = child.get("level") |
| if not isinstance(child_level, str) or not child_level.strip(): |
| raise MalformedHierarchyError( |
| f"Child under {path!r} must contain non-empty 'level'." |
| ) |
| if child_level != expected: |
| raise InvalidHierarchyTransitionError( |
| f"Invalid hierarchy transition under {path!r}: {level!r} -> {child_level!r}; " |
| f"expected child level {expected!r}." |
| ) |
|
|
|
|
| def is_assignable_level(level: str) -> bool: |
| """Return whether a source hierarchy level is assignable when UUID-bearing.""" |
| return level in _ASSIGNABLE_LEVELS |
|
|
