from __future__ import annotations import hashlib import json from app.core.knowledge.contracts import KnowledgeChunkDraft, KnowledgeSnapshot def _hash(payload: object) -> str: value = json.dumps( payload, ensure_ascii=False, sort_keys=True, separators=(",", ":") ) return hashlib.sha256(value.encode("utf-8")).hexdigest() def _chunk_kind(semantic_path: str) -> str: root = semantic_path.split("/", 1)[0] return { "name": "summary", "aliases": "summary", "purpose": "definition", "definition": "definition", "owner": "field", "fields": "field", "rules": "field", }.get(root, "relation") def build_chunks(snapshot: KnowledgeSnapshot) -> tuple[KnowledgeChunkDraft, ...]: chunks = [] for point in snapshot.points: point_set_hash = _hash( { "point_key": point.point_key, "content_hash": point.content_hash, "metadata_hash": point.metadata_hash, "permission_hash": point.permission_hash, } ) chunks.append( KnowledgeChunkDraft( chunk_key=_hash( { "point_key": point.point_key, "point_set_hash": point_set_hash, } ), chunk_kind=_chunk_kind(point.semantic_path), section_path=point.semantic_path, content=point.content, content_hash=point.content_hash, lexical_text=point.content, primary_point_key=point.point_key, point_keys=(point.point_key,), point_set_hash=point_set_hash, metadata={ **point.metadata, "source_type": snapshot.source_type, "source_uid": snapshot.source_uid, "source_revision": snapshot.source_revision, "permission_scope": point.permission_scope, }, ) ) return tuple(chunks) def embedding_input_hash(chunk: KnowledgeChunkDraft, profile_key: str) -> str: return _hash( { "profile_key": profile_key, "content_hash": chunk.content_hash, "content": chunk.content, } )