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

from __future__ import annotations
import hashlib
import json
from typing import Any

SECRET_KEYS = {"password", "credentials", "api_key", "token", "authorization", "connection_string"}


def _safe(value: Any, key: str = "") -> Any:
    if key.lower() in SECRET_KEYS:
        return "[redacted]"
    if isinstance(value, dict):
        return {k: _safe(v, k) for k, v in sorted(value.items()) if k not in {"updated_at", "created_at"}}
    if isinstance(value, list):
        items = [_safe(item) for item in value]
        return sorted(items, key=lambda item: json.dumps(item, sort_keys=True, ensure_ascii=False))
    return value


def build_document(object_type: str, source: dict[str, Any]) -> dict[str, Any]:
    uid = source.get("uid")
    if not uid:
        raise ValueError("governance object uid is required")
    safe = _safe(source)
    content = json.dumps({"object_type": object_type, "source": safe}, sort_keys=True, ensure_ascii=False, separators=(",", ":"))
    return {"object_type": object_type, "object_uid": uid, "object_version": int(source.get("version", 1)), "object_name": source.get("name_zh") or source.get("name") or uid, "business_domain_uid": source.get("business_domain_uid"), "content": content, "content_hash": hashlib.sha256(content.encode()).hexdigest(), "source_updated_at": source.get("updated_at")}


def chunk_document(content: str, max_chars: int = 1200) -> list[str]:
    return [content[index:index + max_chars] for index in range(0, len(content), max_chars)] or [""]