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

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