DataOps-platform/app/core/knowledge/point_builder.py

from __future__ import annotations

import hashlib
import json
from collections.abc import Iterable, Mapping
from typing import Any

from app.core.knowledge.contracts import (
    KnowledgeDependencyDraft,
    KnowledgePointDraft,
    KnowledgeSnapshot,
)

SECRET_MARKERS = (
    "password",
    "credential",
    "api_key",
    "apikey",
    "token",
    "authorization",
    "connection_string",
    "secret",
)
VOLATILE_KEYS = {"created_at", "updated_at"}
SUPPORTED_TYPES = {
    "BusinessDomain",
    "DataFlow",
    "DataMeta",
    "DataStandard",
    "Label",
    "DataLabel",
}


def _is_secret(key: str) -> bool:
    normalized = key.casefold().replace("-", "_")
    return any(marker in normalized for marker in SECRET_MARKERS)


def _canonical(value: Any, key: str = "") -> Any:
    if _is_secret(key):
        return "[redacted]"
    if isinstance(value, Mapping):
        return {
            str(item_key): _canonical(item_value, str(item_key))
            for item_key, item_value in sorted(
                value.items(), key=lambda item: str(item[0])
            )
            if str(item_key).casefold() not in VOLATILE_KEYS
        }
    if isinstance(value, (list, tuple, set, frozenset)):
        items = [_canonical(item) for item in value]
        return sorted(items, key=_canonical_json)
    return value


def _canonical_json(value: Any) -> str:
    return json.dumps(value, ensure_ascii=False, sort_keys=True, separators=(",", ":"))


def _hash(value: Any) -> str:
    return hashlib.sha256(_canonical_json(value).encode("utf-8")).hexdigest()


def _first(source: Mapping[str, Any], *keys: str) -> Any:
    for key in keys:
        value = source.get(key)
        if value is not None and value != "":
            return value
    return None


def _scope(source: Mapping[str, Any]) -> dict[str, Any]:
    explicit = source.get("permission_scope")
    if isinstance(explicit, Mapping):
        return dict(_canonical(explicit))
    domain_uid = source.get("business_domain_uid")
    if domain_uid:
        return {"business_domains": [str(domain_uid)]}
    return {"business_domains": []}


def _content(value: Any) -> str:
    safe = _canonical(value)
    return safe if isinstance(safe, str) else _canonical_json(safe)


def _build_point(
    *,
    prefix: str,
    semantic_path: str,
    value: Any,
    permission_scope: Mapping[str, Any],
    metadata: Mapping[str, Any] | None = None,
) -> KnowledgePointDraft:
    content = _content(value)
    safe_metadata = dict(_canonical(metadata or {}))
    safe_scope = dict(_canonical(permission_scope))
    return KnowledgePointDraft(
        point_key=f"{prefix}/{semantic_path}",
        semantic_path=semantic_path,
        content=content,
        content_hash=_hash(content),
        metadata=safe_metadata,
        metadata_hash=_hash(safe_metadata),
        permission_scope=safe_scope,
        permission_hash=_hash(safe_scope),
    )


def _require_child_uid(item: Mapping[str, Any], kind: str) -> str:
    uid = item.get("uid")
    if not uid:
        raise ValueError(f"{kind} requires a stable uid")
    return str(uid)


def _iter_dicts(value: Any) -> Iterable[Mapping[str, Any]]:
    if not value:
        return ()
    if not isinstance(value, (list, tuple)):
        raise ValueError("nested governance values must be a list")
    if not all(isinstance(item, Mapping) for item in value):
        raise ValueError("nested governance values must contain objects")
    return value


def build_knowledge_snapshot(
    object_type: str, source: Mapping[str, Any]
) -> KnowledgeSnapshot:
    if object_type not in SUPPORTED_TYPES:
        raise ValueError(f"unsupported governance object type: {object_type}")
    source_uid = source.get("uid")
    if not source_uid:
        raise ValueError("governance object uid is required")
    source_uid = str(source_uid)
    source_revision = int(source.get("version", 1))
    permission_scope = _scope(source)
    prefix = f"{object_type}/{source_uid}"
    points: list[KnowledgePointDraft] = []
    dependencies: list[KnowledgeDependencyDraft] = []

    scalar_specs = (
        ("name", ("name_zh", "name", "name_en")),
        ("definition", ("definition", "description")),
        ("purpose", ("purpose", "script_requirement")),
        ("owner", ("owner", "owner_name")),
        ("data_type", ("data_type", "type")),
    )
    for semantic_path, keys in scalar_specs:
        value = _first(source, *keys)
        if value is not None:
            points.append(
                _build_point(
                    prefix=prefix,
                    semantic_path=semantic_path,
                    value=value,
                    permission_scope=permission_scope,
                )
            )

    aliases = source.get("aliases") or []
    if not isinstance(aliases, (list, tuple, set, frozenset)):
        raise ValueError("aliases must be a list")
    for alias in sorted({str(item).strip() for item in aliases if str(item).strip()}):
        alias_id = hashlib.sha256(alias.casefold().encode("utf-8")).hexdigest()[:16]
        points.append(
            _build_point(
                prefix=prefix,
                semantic_path=f"aliases/{alias_id}",
                value=alias,
                permission_scope=permission_scope,
            )
        )

    for relation in _iter_dicts(source.get("relations")):
        target_uid = relation.get("target_uid")
        relation_type = str(relation.get("type") or "").strip().upper()
        if not target_uid or not relation_type:
            raise ValueError("relationship requires type and stable target uid")
        semantic_path = f"{relation_type.casefold()}/{target_uid}"
        point = _build_point(
            prefix=prefix,
            semantic_path=semantic_path,
            value=relation,
            permission_scope=permission_scope,
            metadata={"relation_type": relation_type, "target_uid": str(target_uid)},
        )
        points.append(point)
        target_type = str(relation.get("target_type") or "BusinessDomain")
        dependencies.append(
            KnowledgeDependencyDraft(
                from_point_key=point.point_key,
                to_point_key=f"{target_type}/{target_uid}/definition",
                relation_type=relation_type.casefold(),
                source="governance",
            )
        )

    for collection_name in ("fields", "rules"):
        for item in _iter_dicts(source.get(collection_name)):
            item_uid = _require_child_uid(item, collection_name[:-1])
            for semantic_name, keys in (
                ("name", ("name_zh", "name", "name_en")),
                ("definition", ("definition", "description")),
                ("data_type", ("data_type", "type")),
            ):
                value = _first(item, *keys)
                if value is None:
                    continue
                points.append(
                    _build_point(
                        prefix=prefix,
                        semantic_path=f"{collection_name}/{item_uid}/{semantic_name}",
                        value=value,
                        permission_scope=permission_scope,
                        metadata={"child_uid": item_uid, "child_kind": collection_name},
                    )
                )

    points.sort(key=lambda point: point.point_key)
    if len({point.point_key for point in points}) != len(points):
        raise ValueError("knowledge snapshot contains duplicate point keys")
    dependencies.sort(
        key=lambda item: (
            item.from_point_key,
            item.to_point_key,
            item.relation_type,
            item.source,
        )
    )
    safe_source = _canonical(source)
    point_set_payload = [
        {
            "point_key": point.point_key,
            "content_hash": point.content_hash,
            "metadata_hash": point.metadata_hash,
            "permission_hash": point.permission_hash,
        }
        for point in points
    ]
    return KnowledgeSnapshot(
        source_type=object_type,
        source_uid=source_uid,
        source_revision=source_revision,
        source_snapshot_hash=_hash(safe_source),
        point_set_hash=_hash(point_set_payload),
        permission_scope=permission_scope,
        points=tuple(points),
        dependencies=tuple(dependencies),
        source_updated_at=source.get("updated_at"),
    )