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"), )