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 [""]