from __future__ import annotations from .shared import * from .charter_seed_utils import _build_signal_context_text, _safe_threshold, _to_bullet_text, _to_string_list from .charter_prefill_utils import _default_ai_validations, _default_escalation_trigger, _generate_charter_seed _MODULE_FIELD_KEYS: dict[str, tuple[str, ...]] = { "module_1": ("problem_statement", "goal_statement", "kpis"), "module_2": ("in_scope", "out_of_scope", "alignment_qa"), "module_3": ("primary_model", "fallback_model", "capability_table", "dependency_map", "business_view", "engineering_view"), "module_4": ("milestones", "role_effort_estimates", "capability_owner_assignment", "buffer_policy"), "module_5": ("risk_register", "trigger_rules", "execution_rule"), "module_6": ("attachment_metadata", "attachment_summary", "search_tags", "governance_rule"), } _STATUS_REASON_MAP = { "evidence_insufficient": "缺少可验证证据", "generation_failed": "AI 生成失败", "generic_text": "文本存在但证据不足", "conflict_blocked": "存在未决冲突", } def _field_required(module_key: str, field_key: str) -> bool: return field_key in _CHARTER_REQUIRED_FIELDS.get(module_key, ()) def _field_title(module_key: str, field_key: str) -> str: return f"{module_key}.{field_key}" def _stringify_field_value(value: Any) -> str: if value is None: return "" if isinstance(value, str): return value.strip() if isinstance(value, (list, tuple)): items = _to_string_list(value) if not items: return "" return _to_bullet_text(items) return str(value).strip() def _normalize_evidence_items(value: Any) -> list[dict[str, Any]]: if not isinstance(value, list): return [] normalized: list[dict[str, Any]] = [] for item in value: if isinstance(item, dict): try: signal_id = int(item.get("signal_id")) if item.get("signal_id") is not None else None except (TypeError, ValueError): signal_id = None normalized.append( { "signal_id": signal_id, "summary": str(item.get("summary") or item.get("excerpt") or "").strip(), "quote": str(item.get("quote") or item.get("snippet") or "").strip(), "reason": str(item.get("reason") or "").strip(), "confidence": _safe_threshold(item.get("confidence"), 0.0), } ) continue text = str(item or "").strip() if text: normalized.append( { "signal_id": None, "summary": text, "quote": "", "reason": "", "confidence": 0.0, } ) return normalized def _legacy_seed_modules(seed: dict[str, Any]) -> dict[str, dict[str, Any]]: return { "module_1": { "problem_statement": seed.get("summary"), "goal_statement": _to_bullet_text(_to_string_list(seed.get("objectives"))), "kpis": _to_bullet_text( _to_string_list(seed.get("success_metrics") or seed.get("success_criteria")) ), }, "module_2": { "in_scope": _to_bullet_text(_to_string_list(seed.get("in_scope"))), "out_of_scope": _to_bullet_text(_to_string_list(seed.get("out_of_scope"))), "alignment_qa": str(seed.get("alignment_qa") or "").strip(), }, "module_3": { "primary_model": seed.get("primary_model"), "fallback_model": seed.get("fallback_model"), "capability_table": seed.get("capability_table"), "dependency_map": _to_bullet_text(_to_string_list(seed.get("dependencies"))), "business_view": seed.get("business_view"), "engineering_view": seed.get("engineering_view"), }, "module_4": { "milestones": _to_bullet_text(_to_string_list(seed.get("milestones") or seed.get("mvp_items"))), "role_effort_estimates": _to_bullet_text(_to_string_list(seed.get("team_setup"))), "capability_owner_assignment": seed.get("owner_suggestion"), "buffer_policy": seed.get("buffer_policy"), }, "module_5": { "risk_register": _to_bullet_text(_to_string_list(seed.get("risks"))), "trigger_rules": _to_bullet_text(_to_string_list(seed.get("trigger_rules"))), "execution_rule": _to_bullet_text(_to_string_list(seed.get("execution_rule"))), }, "module_6": { "attachment_metadata": _to_bullet_text(_to_string_list(seed.get("attachment_metadata"))), "attachment_summary": seed.get("attachment_summary"), "search_tags": _to_bullet_text(_to_string_list(seed.get("search_tags"))), "governance_rule": seed.get("governance_rule"), }, } def _derive_status( *, value: str, evidence: list[dict[str, Any]], required: bool, explicit_status: str, generation_failed: bool, ) -> str: if explicit_status: return explicit_status if value and evidence: return "valid" if value: return "generic_text" if generation_failed: return "generation_failed" if required: return "evidence_insufficient" return "generation_failed" def _readiness_detail_for_status(module_key: str, field_key: str, status: str) -> str: reason = _STATUS_REASON_MAP.get(status, "未通过校验") return f"{_field_title(module_key, field_key)} {reason}" def _build_verification_metrics(modules: dict[str, dict[str, str]], field_status: dict[str, dict[str, str]]) -> list[dict[str, Any]]: if field_status.get("module_1", {}).get("kpis") != "valid": return [] items = _to_string_list(modules.get("module_1", {}).get("kpis")) return [ { "name": item, "expected": item, "source_field": "module_1.kpis", } for item in items ] def _build_risk_gates(modules: dict[str, dict[str, str]], field_status: dict[str, dict[str, str]]) -> list[dict[str, Any]]: trigger_status = field_status.get("module_5", {}).get("trigger_rules") execution_status = field_status.get("module_5", {}).get("execution_rule") if trigger_status != "valid" or execution_status != "valid": return [] triggers = _to_string_list(modules.get("module_5", {}).get("trigger_rules")) action = modules.get("module_5", {}).get("execution_rule", "").strip() return [ { "name": trigger, "action": action, "source_fields": ["module_5.trigger_rules", "module_5.execution_rule"], } for trigger in triggers ] async def _build_ai_first_charter_content( *, db: AsyncSession, tenant_id: int, project_id: int, signal_rows: list[Signal], signal_ids: list[int], raw_text: str = "", ) -> dict[str, Any]: signal_context = _build_signal_context_text(signal_rows, raw_text=raw_text) seed = await _generate_charter_seed( db=db, tenant_id=tenant_id, project_id=project_id, signal_context=signal_context, ) generation_error = str(seed.get("generation_error") or "").strip() seed_modules = seed.get("modules") if not isinstance(seed_modules, dict): seed_modules = _legacy_seed_modules(seed) modules: dict[str, dict[str, str]] = {} field_status: dict[str, dict[str, str]] = {} field_evidence: dict[str, dict[str, list[dict[str, Any]]]] = {} blocking_fields: list[str] = [] blocking_reasons: list[str] = [] for module_key, field_keys in _MODULE_FIELD_KEYS.items(): modules[module_key] = {} field_status[module_key] = {} field_evidence[module_key] = {} module_seed = seed_modules.get(module_key) if not isinstance(module_seed, dict): module_seed = {} for field_key in field_keys: raw_field = module_seed.get(field_key) explicit_status = "" if isinstance(raw_field, dict): explicit_status = str(raw_field.get("status") or "").strip() raw_value = raw_field.get("value") if raw_value is None: raw_value = raw_field.get("text") else: raw_value = raw_field value = _stringify_field_value(raw_value) evidence = _normalize_evidence_items(raw_field.get("evidence")) if isinstance(raw_field, dict) else [] status = _derive_status( value=value, evidence=evidence, required=_field_required(module_key, field_key), explicit_status=explicit_status, generation_failed=bool(generation_error), ) modules[module_key][field_key] = value field_status[module_key][field_key] = status field_evidence[module_key][field_key] = evidence if _field_required(module_key, field_key) and status != "valid": blocking_fields.append(_field_title(module_key, field_key)) if isinstance(raw_field, dict): reason = str(raw_field.get("blocking_reason") or "").strip() else: reason = "" blocking_reasons.append(reason or _readiness_detail_for_status(module_key, field_key, status)) source_refs = [f"signal:{sid}" for sid in signal_ids] now_iso = now_cn_aware().isoformat() validations_m1 = _default_ai_validations("模块1：触发与目标") validations_m2 = _default_ai_validations("模块2：范围与对齐") validations_m3 = _default_ai_validations("模块3：模型与依赖") validations_m4 = _default_ai_validations("模块4：计划与资源") validations_m5 = _default_ai_validations("模块5：风险与兜底") validations_m6 = _default_ai_validations("模块6：附件与治理") escalation = _default_escalation_trigger() for module_key, validations in ( ("module_1", validations_m1), ("module_2", validations_m2), ("module_3", validations_m3), ("module_4", validations_m4), ("module_5", validations_m5), ("module_6", validations_m6), ): modules[module_key]["ai_validations"] = validations modules[module_key]["escalation_trigger"] = escalation explicit_readiness = seed.get("readiness") if isinstance(seed.get("readiness"), dict) else {} for item in explicit_readiness.get("blocking_fields") or []: field = str(item or "").strip() if field and field not in blocking_fields: blocking_fields.append(field) for item in explicit_readiness.get("blocking_reasons") or []: reason = str(item or "").strip() if reason and reason not in blocking_reasons: blocking_reasons.append(reason) required_total = sum(len(fields) for fields in _CHARTER_REQUIRED_FIELDS.values()) valid_required = 0 for module_key, fields in _CHARTER_REQUIRED_FIELDS.items(): for field_key in fields: if field_status.get(module_key, {}).get(field_key) == "valid": valid_required += 1 readiness = { "ready_for_confirm": not blocking_fields and explicit_readiness.get("ready_for_confirm") is not False, "blocking_fields": blocking_fields, "blocking_reasons": blocking_reasons, } derived_artifacts = seed.get("derived_artifacts") if isinstance(seed.get("derived_artifacts"), dict) else {} if not isinstance(derived_artifacts.get("task_seeds"), list): derived_artifacts["task_seeds"] = [] if not isinstance(derived_artifacts.get("verification_metrics"), list) or not derived_artifacts.get("verification_metrics"): derived_artifacts["verification_metrics"] = _build_verification_metrics(modules, field_status) if not isinstance(derived_artifacts.get("risk_gates"), list) or not derived_artifacts.get("risk_gates"): derived_artifacts["risk_gates"] = _build_risk_gates(modules, field_status) summary = modules["module_1"]["problem_statement"] background = str(seed.get("background") or "").strip() objectives = _to_string_list(modules["module_1"]["goal_statement"]) scope = str(seed.get("scope") or "").strip() in_scope_items = _to_string_list(modules["module_2"]["in_scope"]) out_of_scope_items = _to_string_list(modules["module_2"]["out_of_scope"]) success_metrics = _to_string_list(modules["module_1"]["kpis"]) risks = _to_string_list(modules["module_5"]["risk_register"]) return { "schema_version": "charter_ai_first_v1", "auto_generated_by_ai": True, "metadata": { "status": "draft", "confidence_score": _safe_threshold( seed.get("confidence_score") or (valid_required / required_total if required_total else 0.0), 0.0, ), "last_validated_at": now_iso, "evidence_refs": source_refs, "escalation_reason": blocking_reasons[0] if blocking_reasons else generation_error or None, }, "modules": modules, "field_status": field_status, "field_evidence": field_evidence, "readiness": readiness, "derived_artifacts": derived_artifacts, "signal_ids": signal_ids, "summary": summary, "background": background, "objectives": objectives, "scope": scope, "success_metrics": success_metrics, "core_metrics": success_metrics, "rationale": summary, "decision": _to_bullet_text(objectives), "in_scope": in_scope_items, "out_of_scope": out_of_scope_items, "risks": risks, "raw_source_text": raw_text.strip() or None, }