PMFlow/backend/app/services/sstc/charter_payload.py

156 lines
4.5 KiB
Python
Raw Permalink Blame History

This file contains ambiguous Unicode characters

This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.

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