from __future__ import annotations from typing import Any from .common import ( EventLoader, context_input_fields, decision_context, decision_event_refs, event_notices, event_unit, find_raw_round, find_round, load_events, ) from .schema import field, payload def project_objective( bundle: dict[str, Any], view: dict[str, Any], load_event: EventLoader ) -> dict[str, Any]: record = bundle.get("record") or {} node = view.get("objective") or {} context = decision_context(node) events = load_events(load_event, decision_event_refs(context)) inputs = context_input_fields(context, "objective", events) output = field( "objective:current-value", "完整创作方向", record.get("script_direction"), source_kind="database", source_label="script_build_record", source_ref="run:objective", field_path="script_direction", relation="persisted-output", ) outputs = [output] if context.get("decisionItems"): outputs.append(field( "objective:goals", "创作目标", context.get("decisionItems"), source_kind="calculation", source_label="创作方向结构化解析", source_ref="run:objective", field_path="decision.decisionItems", relation="persisted-output", )) if context.get("constraints"): outputs.append(field( "objective:dimensions", "评估维度与约束", context.get("constraints"), source_kind="calculation", source_label="创作方向结构化解析", source_ref="run:objective", field_path="decision.constraints", relation="persisted-output", )) notices = event_notices(events) if inputs and any(item["relation"] == "available-upstream" for item in inputs): notices.append({"level": "info", "message": "可确认是上游信息，但无法确认是否被本次决策采用。"}) return payload( "objective", "business-record", inputs=inputs, units=[event_unit(item) for item in events], outputs=outputs, runtime_summary=(f"找到 {len(events)} 个保存/修订事件。" if events else "只能确认当前保存的业务值，未找到完整形成过程。"), notices=notices, completeness=context.get("completeness") if context.get("completeness") in {"complete", "partial", "missing"} else None, business_modules=[ {"id": "direction", "title": "完整创作方向原文", "sourceIds": [output["id"]]}, {"id": "inputs", "title": "形成前读取", "sourceIds": [item["id"] for item in inputs]}, {"id": "structure", "title": "目标与评估维度", "sourceIds": [item["id"] for item in outputs[1:]]}, {"id": "revisions", "title": "修订记录", "sourceIds": [output["id"]]}, ], card_fields=[ {"key": "headline", "label": "核心目标", "sourceIds": [output["id"]], "transform": "使用结构化解析的 headline"}, {"key": "targetCount", "label": "目标数", "sourceIds": ["objective:goals"], "transform": "列表计数"}, {"key": "dimensionCount", "label": "评估维度数", "sourceIds": ["objective:dimensions"], "transform": "列表计数"}, ], ) def project_round_goal( detail_ref: str, round_index: int, bundle: dict[str, Any], view: dict[str, Any], load_event: EventLoader, ) -> dict[str, Any]: raw = find_raw_round(bundle, round_index) or {} round_ = find_round(view, round_index) or {} node = round_.get("goal") or {} context = decision_context(node) refs = decision_event_refs(context) refs.extend( f"event:{item.get('id')}" for item in bundle.get("events") or [] if item.get("event_name") == "begin_round" ) events = load_events(load_event, refs) inputs = context_input_fields(context, f"round:{round_index}:goal", events) begin_round_fields = [ field( f"round:{round_index}:goal:begin:{event.get('id')}", "进入本轮时提交的目标", ((event.get("input") or {}).get("content") or {}).get("goal") if isinstance((event.get("input") or {}).get("content"), dict) else (event.get("input") or {}).get("content"), source_kind="runtime-event", source_label="begin_round 运行事件", source_ref=f"event:{event.get('id')}", field_path="input.content.goal", relation="run-output", ) for event in events if event.get("event_name") == "begin_round" and ( ( isinstance((event.get("output") or {}).get("content"), dict) and int(((event.get("output") or {}).get("content") or {}).get("round_index") or 0) == round_index ) or ( isinstance((event.get("input") or {}).get("content"), dict) and ((event.get("input") or {}).get("content") or {}).get("goal") == raw.get("goal") ) ) ] outputs = [field( f"round:{round_index}:goal:value", "Goal 原文", raw.get("goal"), source_kind="database", source_label="script_build_round", source_ref=detail_ref, field_path="goal", relation="persisted-output", )] return payload( "round-goal", "business-record", inputs=[*inputs, *begin_round_fields], units=[event_unit(item) for item in events], outputs=outputs, runtime_summary=(f"通过 {len(events)} 个 begin_round/读取事件形成本轮目标。" if events else "本轮目标有业务记录，但形成过程未完整保存。"), notices=event_notices(events) + ([{"level": "info", "message": "可确认是上游信息，但无法确认是否被本次决策采用。"}] if any(item["relation"] == "available-upstream" for item in inputs) else []), completeness=context.get("completeness") if context.get("completeness") in {"complete", "partial", "missing"} else None, business_modules=[ {"id": "goal", "title": "Goal 原文", "sourceIds": [outputs[0]["id"]]}, {"id": "why", "title": "进入本轮时提交的目标", "sourceIds": [item["id"] for item in begin_round_fields]}, {"id": "dependencies", "title": "本轮形成前可见的上游", "sourceIds": [item["id"] for item in inputs]}, ], card_fields=[ {"key": "headline", "label": "本轮目标", "sourceIds": [outputs[0]["id"]], "transform": "使用结构化目标首项"}, {"key": "basis", "label": "形成前依据", "sourceIds": [item["id"] for item in inputs[:1]], "transform": "取第一个明确上游依据"}, ], ) def project_implementation_plan( detail_ref: str, round_index: int, bundle: dict[str, Any], view: dict[str, Any], load_event: EventLoader, ) -> dict[str, Any]: raw = find_raw_round(bundle, round_index) or {} round_ = find_round(view, round_index) or {} node = ((round_.get("plan") or {}).get("node") or {}) context = decision_context(node) refs = decision_event_refs(context) refs.extend( f"event:{item.get('id')}" for item in bundle.get("events") or [] if item.get("event_name") == "record_multipath_plan" and int(item.get("round_index") or 0) == round_index ) events = load_events(load_event, refs) inputs = context_input_fields(context, f"round:{round_index}:plan", events) outputs = [ field(f"round:{round_index}:plan:mode", "规划方式", raw.get("race_or_divide"), source_kind="database", source_label="script_build_round", source_ref=detail_ref, field_path="race_or_divide", relation="persisted-output"), field(f"round:{round_index}:plan:routes", "实现路线", raw.get("multipath_plan") or [], source_kind="database", source_label="script_build_round", source_ref=detail_ref, field_path="multipath_plan", relation="persisted-output"), field(f"round:{round_index}:plan:reason", "规划理由", raw.get("plan_note"), source_kind="database", source_label="script_build_round", source_ref=detail_ref, field_path="plan_note", relation="persisted-output"), ] revision_fields = [ field( f"round:{round_index}:plan:revision:{event.get('id')}", f"规划保存 Event {event.get('id')}", (event.get("input") or {}).get("content"), source_kind="runtime-event", source_label="record_multipath_plan 运行事件", source_ref=f"event:{event.get('id')}", field_path="input.content", relation="run-output", ) for event in events if event.get("event_name") == "record_multipath_plan" ] return payload( "implementation-plan", "business-record", inputs=inputs, units=[event_unit(item) for item in events], outputs=[*outputs, *revision_fields], runtime_summary=(f"找到 {len(events)} 个规划保存/修订事件。" if events else "展示数据库当前保存的实现规划。"), notices=event_notices(events), completeness=context.get("completeness") if context.get("completeness") in {"complete", "partial", "missing"} else None, business_modules=[ {"id": "mode", "title": "规划模式与理由", "sourceIds": [outputs[0]["id"], outputs[2]["id"]]}, {"id": "routes", "title": "逐条实现路线", "sourceIds": [outputs[1]["id"]]}, {"id": "basis", "title": "规划依据", "sourceIds": [item["id"] for item in inputs]}, {"id": "revisions", "title": "规划保存与修订记录", "sourceIds": [item["id"] for item in revision_fields]}, ], card_fields=[ {"key": "mode", "label": "模式", "sourceIds": [outputs[0]["id"]], "transform": "规范化为单路/赛马/分工"}, {"key": "routeCount", "label": "路线数", "sourceIds": [outputs[1]["id"]], "transform": "列表计数"}, {"key": "routeScope", "label": "路线作用范围", "sourceIds": [outputs[1]["id"]], "transform": "取每路结构化 target"}, ], )