from __future__ import annotations import json import re from typing import Any from ...models import TruthNode def _as_dict(value: Any) -> dict[str, Any]: if isinstance(value, dict): return value if isinstance(value, str): try: parsed = json.loads(value) except (TypeError, ValueError): return {"text": value} return parsed if isinstance(parsed, dict) else {"text": value} return {} def _compact_text(value: object, *, max_len: int = 120) -> str: text = str(value or "").strip() if not text: return "" text = re.sub(r"\s+", " ", text) return text[:max_len] def _compact_items( value: object, *, max_items: int = 6, max_len: int = 120, ) -> list[str]: if isinstance(value, list): raw_items = value elif isinstance(value, str): raw_items = re.split(r"[\n\r,,;;、/|]+", value) else: raw_items = [] items: list[str] = [] seen: set[str] = set() for raw in raw_items: text = _compact_text(raw, max_len=max_len) if not text: continue key = text.lower() if key in seen: continue seen.add(key) items.append(text) if len(items) >= max_items: break return items def _build_charter_context_payload(charter: TruthNode) -> dict[str, Any]: content = _as_dict(charter.content) modules = content.get("modules") if not isinstance(modules, dict): modules = {} module_1 = modules.get("module_1") if isinstance(modules.get("module_1"), dict) else {} module_2 = modules.get("module_2") if isinstance(modules.get("module_2"), dict) else {} module_4 = modules.get("module_4") if isinstance(modules.get("module_4"), dict) else {} module_5 = modules.get("module_5") if isinstance(modules.get("module_5"), dict) else {} module_6 = modules.get("module_6") if isinstance(modules.get("module_6"), dict) else {} signal_ids = content.get("signal_ids") normalized_signal_ids: list[int] = [] if isinstance(signal_ids, list): for sid in signal_ids: try: normalized_signal_ids.append(int(sid)) except Exception: continue return { "id": int(charter.id), "title": _compact_text(charter.title or "", max_len=120) or "未命名宪章", "version": int(charter.version or 1), "status": _compact_text(charter.status or "", max_len=32) or "unknown", "summary": _compact_text( content.get("summary") or content.get("rationale") or content.get("background") or module_1.get("problem_statement") or "", max_len=180, ), "objectives": _compact_items( content.get("objectives") or module_1.get("goal_statement") or [], max_items=5, max_len=100, ), "scope": _compact_text( content.get("scope") or module_2.get("alignment_qa") or "", max_len=240, ), "in_scope": _compact_items( content.get("in_scope") or module_2.get("in_scope") or [], max_items=5, max_len=100, ), "out_of_scope": _compact_items( content.get("out_of_scope") or module_2.get("out_of_scope") or [], max_items=5, max_len=100, ), "success_criteria": _compact_items( content.get("success_criteria") or content.get("success_metrics") or content.get("core_metrics") or module_1.get("kpis") or [], max_items=6, max_len=100, ), "constraints": _compact_items( content.get("constraints") or module_5.get("execution_rule") or module_6.get("governance_rule") or module_2.get("alignment_qa") or [], max_items=6, max_len=100, ), "milestones": _compact_items( content.get("milestones") or module_4.get("milestones") or [], max_items=5, max_len=100, ), "risks": _compact_items( content.get("risks") or module_5.get("risk_register") or [], max_items=5, max_len=100, ), "signal_ids": normalized_signal_ids, } __all__ = ["_build_charter_context_payload"]