src/gcmd_classifier/vocabulary/loader.py

"""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,  # type: ignore[arg-type]
            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