from __future__ import annotations import re from copy import deepcopy from typing import Any from .resolver import SourceResolver, pointer_lookup from .schema import SourceBinding, unresolved_binding # These mappings are keyed by the stable business projection shape, never by # fuzzy title matching. Values point at the richer source modules built by the # card-specific projectors. _MODULE_MAP: dict[str, dict[str, list[str]]] = { "objective": { "block:direction": ["direction"], "block:inputs": ["inputs"], }, "round-goal": { "block:goal": ["goal", "why"], "block:dependencies": ["dependencies"], }, "implementation-plan": { "block:plan": ["mode", "routes", "revisions"], "block:basis": ["basis", "revisions"], }, "implementation-task": { "section:scope": ["scope"], "section:method": ["method"], "section:goal": ["goal"], "section:constraints": ["constraints"], "section:avoid": ["avoid"], "section:deliverable": ["deliverable"], "section:action": ["action"], "section:request": ["request"], "section:result": ["result"], }, "creative": { "block:basis": ["basis"], "block:process": ["process"], "block:available": ["available"], "block:changes": ["changes"], "block:candidate": ["candidate"], "block:uncertainty": ["uncertainty"], "block:record-note": ["record-note"], }, "data-decision": { "block:0": ["sources"], "block:1": ["decision", "tradeoff"], "block:2": ["reason"], }, "multipath-decision": { "block:0": ["strength"], "block:1": ["scope"], "block:2": ["comparison"], "block:3": ["comparison"], "block:4": ["reason"], "block:5": ["strength"], }, "domain-facts": { "section:output": ["facts", "verification", "sources", "usage"], "section:assessment": [], }, "candidate-output-fallback": { "section:output": ["output"], "section:assessment": ["assessment"], }, "round-result": { "section:goal": ["completion"], "section:outcomes": ["dispositions"], "section:facts": ["domain"], "section:evaluation": ["review"], "section:changes": ["changes"], "section:next": ["next"], }, "final-result": { "section:conclusion": ["conclusion"], "section:summary": ["summary"], "section:statistics": ["stats"], "section:artifact": ["scale", "artifact"], "section:version": ["version"], }, "retrieval-agent": { "section:task": ["task"], "section:queries": ["queries"], "section:hits": ["raw"], "section:screening": ["screening"], "special:activities": ["activities"], }, "retrieval-direct-group": { "section:purpose": ["purpose"], "section:sources": ["calls"], "section:result": ["results", "failures"], "special:activities": ["calls", "results", "failures"], }, "retrieval-event": { "special:conditions": ["purpose"], "special:outcome": ["status"], "special:results": ["result", "results"], "special:completeness": ["status"], }, } def business_only_projection(detail: dict[str, Any]) -> dict[str, Any]: """Keep exactly what the readable Inspector uses, without duplicating raw JSON.""" hidden = {"technical", "changes"} return deepcopy({key: value for key, value in detail.items() if key not in hidden}) def align_business_projection( *, detail_ref: str, card_kind: str, business: dict[str, Any], source_modules: list[dict[str, Any]], resolver: SourceResolver, bundle: dict[str, Any], ) -> list[dict[str, Any]]: """Turn the readable Inspector projection into row-aligned source modules.""" projection = business_only_projection(business) normalized = _normalize_business(projection, card_kind=card_kind) source_by_id = {str(module.get("id")): module for module in source_modules} aligned: list[dict[str, Any]] = [] for module_index, module in enumerate(normalized, 1): semantic_key = str(module.pop("_semanticKey")) source_ids = _source_module_ids(card_kind, semantic_key, module_index) candidates = [source_by_id[source_id] for source_id in source_ids if source_id in source_by_id] if not candidates and card_kind == "implementation-task" and "historical-task" in source_by_id: candidates = [source_by_id["historical-task"]] rows: list[dict[str, Any]] = [] module_bindings: list[SourceBinding] = [] for row_index, row in enumerate(module.get("rows") or [], 1): found, business_value = pointer_lookup( projection, str(row.get("businessSelector") or "") ) if card_kind == "domain-facts" and semantic_key == "section:assessment": branch_id = _branch_id(detail_ref) branch = next((item for item in bundle.get("branches") or [] if _integer(item.get("branch_id")) == branch_id), None) bindings = [resolver.binding_for_field({ "id": f"branch:{branch_id}:assessment", "label": "实现者说明", "value": (branch or {}).get("self_assessment"), "source": { "kind": "database", "label": "script_build_branch", "ref": detail_ref, "fieldPath": "self_assessment", }, "relation": "persisted-output", "completeness": "complete" if branch and branch.get("self_assessment") else "partial", }, binding_id=f"{row['id']}:source:1", role="output")] elif semantic_key in {"special:question", "block:record-note"}: bindings = [resolver.binding_for_field({ "id": f"{module['id']}:display-text", "label": "业务详情确定性文案", "value": business_value if found else None, "source": { "kind": "calculation", "label": "业务详情确定性说明", "ref": f"calculation:{module['id']}:display-text", "fieldPath": "value", }, "relation": "persisted-output", "completeness": "complete", }, binding_id=f"{row['id']}:source:1", role="status")] elif card_kind == "retrieval-event" and semantic_key in { "special:outcome", "special:completeness", }: event_id = _event_id(detail_ref) output_key = str(row.get("id") or "").rsplit(":", 1)[-1] path = "output.content" if semantic_key == "special:outcome" else "" binding = resolver.binding_for_event( event_id or 0, binding_id=f"{row['id']}:source:1", path=path, role="output" if semantic_key == "special:outcome" else "status", availability=( "returned-to-agent" if semantic_key == "special:outcome" else "produced-by-run" ), ) binding["transform"] = { "kind": "parsed", "operation": ( "retrieval-outcome-v1" if semantic_key == "special:outcome" else "event-body-completeness-v1" ), "outputKey": output_key, } binding["evidence"]["confidence"] = "deterministic" bindings = [binding] elif semantic_key == "special:activities": ref = "" if found and isinstance(business_value, dict): ref = str( business_value.get("detailRef") or business_value.get("id") or "" ) event_id = _event_id(ref) if event_id is not None: binding = resolver.binding_for_event( event_id, binding_id=f"{row['id']}:source:1", role="process", availability="produced-by-run", ) binding["transform"] = { "kind": "parsed", "operation": "activity-summary-v1", "outputKey": str(row.get("id") or "activity"), } binding["evidence"]["confidence"] = "deterministic" bindings = [binding] elif found and ( business_value == [] or str(row.get("businessSelector") or "") == "/missingActivityNotice" ): empty_binding = resolver.binding_for_field( { "id": f"{row['id']}:empty", "label": "可见查询运行记录数量", "value": business_value, "source": { "kind": "calculation", "label": "取数 Agent 结构化查询记录确定性检查", "ref": f"calculation:{row['id']}:empty", "fieldPath": "value", }, "relation": "persisted-output", "completeness": "complete", }, binding_id=f"{row['id']}:source:1", role="status", ) empty_source = resolver.sources.get(empty_binding["sourceId"]) if empty_source is not None: empty_source["resultState"] = "empty" empty_source.setdefault("locator", {})["resultCount"] = 0 bindings = [empty_binding] else: bindings = [] else: bindings = _bindings_for_row( module_id=str(module["id"]), row=row, row_index=row_index, candidates=candidates, resolver=resolver, business_value=business_value if found else None, narrow_event_text=card_kind == "implementation-task", ) if card_kind == "implementation-plan" and semantic_key == "block:plan": revision = _implementation_plan_revision_binding( row=row, business_value=business_value if found else None, candidates=candidates, resolver=resolver, ) if revision is not None: bindings.append(revision) if not bindings: binding = unresolved_binding( f"{module['id']}:row:{row_index}:unresolved", reason=f"业务详情「{module['title']}」的这一项没有保存可安全定位的原始记录", code="binding-map-missing", ) resolver.add_unresolved_source(binding) bindings = [binding] _mark_row_transform( bindings, business_value if found else None, resolver, str(module["id"]), ) row = {key: value for key, value in row.items() if not key.startswith("_")} row["bindingIds"] = [binding["id"] for binding in bindings] rows.append(row) module_bindings.extend(bindings) module["rows"] = rows module["bindings"] = _unique_bindings(module_bindings) module["runtimeRefs"] = list(dict.fromkeys( binding["sourceId"] for binding in module["bindings"] if binding["sourceId"].startswith("event:") )) aligned.append(module) return aligned def _source_module_ids(card_kind: str, semantic_key: str, index: int) -> list[str]: explicit = _MODULE_MAP.get(card_kind, {}).get(semantic_key) if explicit is not None: return explicit if semantic_key.startswith("section:"): return [semantic_key.split(":", 1)[1]] if semantic_key.startswith("block:"): try: return [f"block-{int(semantic_key.split(':', 1)[1]) + 1}"] except ValueError: return [f"block-{index}"] return [semantic_key.split(":", 1)[-1]] def _normalize_business( projection: dict[str, Any], *, card_kind: str = "" ) -> list[dict[str, Any]]: modules: list[dict[str, Any]] = [] question = projection.get("question") if question not in (None, ""): modules.append(_module( "question", "这一步要回答什么", "/question", "special:question", [_row("question:value", None, "/question")], default_open=True, )) for index, section in enumerate(projection.get("businessSections") or []): if not isinstance(section, dict): continue section_id = str(section.get("id") or f"section-{index + 1}") path = f"/businessSections/{index}" rows = [] if isinstance(section.get("items"), list): for item_index, item in enumerate(section.get("items") or []): label = str((item or {}).get("label") or f"第 {item_index + 1} 项") if isinstance(item, dict) else f"第 {item_index + 1} 项" item_path = f"{path}/items/{item_index}" rows.append(_row(f"{section_id}:item:{item_index + 1}", label, item_path, item_index=item_index)) elif section.get("content") is not None: rows.append(_row(f"{section_id}:content", None, f"{path}/content")) semantic_id = _section_semantic_id(section_id) modules.append(_module( section_id, str(section.get("title") or f"业务模块 {index + 1}"), path, f"section:{semantic_id}", rows, presentation=str(section.get("presentation") or "text"), default_open=not bool(section.get("collapsible")) or bool(section.get("defaultOpen")), )) for index, block in enumerate(projection.get("blocks") or []): if not isinstance(block, dict): continue path = f"/blocks/{index}" semantic_id = _block_semantic_id(card_kind, block, index) module_id = semantic_id or f"block-{index + 1}" modules.append(_module( module_id, str(block.get("title") or "记录说明"), path, f"block:{semantic_id}" if semantic_id else f"block:{index}", _block_rows(module_id, path, block), presentation=str(block.get("presentation") or block.get("type") or "text"), default_open=not bool(block.get("collapsible")) or bool(block.get("defaultOpen")), )) conditions = projection.get("queryConditions") if projection.get("detailKind") == "retrieval-event" and isinstance(conditions, list): rows = [ _row( f"conditions:item:{index + 1}", str(item.get("label") or item.get("key") or f"条件 {index + 1}"), f"/queryConditions/{index}/value", item_index=index, ) for index, item in enumerate(conditions) if isinstance(item, dict) ] if not rows: rows = [_row("conditions:empty", None, "/queryConditions")] modules.append(_module("conditions", "查询条件", "/queryConditions", "special:conditions", rows)) outcome = projection.get("outcome") if isinstance(outcome, dict): rows = [] for key, label in (("message", "执行结果"), ("count", "返回数量"), ("errorKind", "失败类型")): if outcome.get(key) is not None: rows.append(_row(f"outcome:{key}", label, f"/outcome/{key}")) modules.append(_module("outcome", "查询结果", "/outcome", "special:outcome", rows)) result_items = projection.get("items") if isinstance(result_items, list) and result_items: rows = [ _row( f"results:item:{index + 1}", str((item or {}).get("title") or f"结果 {index + 1}") if isinstance(item, dict) else f"结果 {index + 1}", f"/items/{index}", item_index=index, ) for index, item in enumerate(result_items) ] modules.append(_module("results", "返回结果", "/items", "special:results", rows)) activities = projection.get("activities") if isinstance(activities, list): rows = [ _row( f"activities:item:{index + 1}", str((item or {}).get("label") or f"运行记录 {index + 1}") if isinstance(item, dict) else f"运行记录 {index + 1}", f"/activities/{index}", item_index=index, ) for index, item in enumerate(activities) ] if not rows: notice_path = "/missingActivityNotice" if projection.get("missingActivityNotice") else "/activities" rows = [_row("activities:empty", None, notice_path)] modules.append(_module("activities", "运行记录", "/activities", "special:activities", rows, presentation="process")) completeness = projection.get("completeness") if projection.get("detailKind") == "retrieval-event" and isinstance(completeness, dict): rows = [] if completeness.get("bodyAvailable") is False: rows.append(_row("completeness:body", "完整返回", "/completeness/bodyAvailable")) if completeness.get("truncated"): rows.append(_row( "completeness:truncated", "截断说明", "/completeness/omittedCharacters" if completeness.get("omittedCharacters") is not None else "/completeness/truncated", )) if rows: modules.append(_module("completeness", "记录完整性", "/completeness", "special:completeness", rows, presentation="notice")) return modules def _block_rows(module_id: str, path: str, block: dict[str, Any]) -> list[dict[str, Any]]: block_type = str(block.get("type") or "summary") if block_type in {"summary", "notice"}: return [_row(f"{module_id}:value", None, f"{path}/value")] if block_type == "items": return [ _row( f"{module_id}:item:{index + 1}", str( item.get("label") or item.get("title") or item.get("subject") or item.get("summary") or f"第 {index + 1} 项" ) if isinstance(item, dict) else f"第 {index + 1} 项", f"{path}/items/{index}", item_index=index, ) for index, item in enumerate(block.get("items") or []) ] if block_type == "implementation-plan": rows = [] if block.get("mode") is not None: rows.append(_row(f"{module_id}:mode", "规划方式", f"{path}/mode", binding_slot="mode:0")) for route_index, route in enumerate(block.get("routes") or []): if not isinstance(route, dict): continue route_number = route.get("pathIndex") or route_index + 1 for key, label in (("target", "作用范围"), ("method", "实施方法"), ("action", "具体动作"), ("emphasis", "路线侧重"), ("pathType", "路线类型")): if route.get(key) is None: continue source_key = {"pathIndex": "path_index", "pathType": "path_type"}.get(key, key) rows.append(_row( f"{module_id}:route:{route_index + 1}:{key}", f"路线 {route_number} · {label}", f"{path}/routes/{route_index}/{key}", binding_slot="routes:0", source_suffix=f"/{route_index}/{source_key}", )) if block.get("reasoning") is not None: rows.append(_row(f"{module_id}:reasoning", "为什么这样规划", f"{path}/reasoning", binding_slot="mode:1")) return rows if block_type == "creative-process": rows = [] for step_index, step in enumerate(block.get("steps") or []): if not isinstance(step, dict): continue number = step.get("stepIndex") or step_index + 1 for key, label in (("action", "动作"), ("summary", "结果"), ("plan", "执行计划"), ("reasoning", "思考依据")): if step.get(key) is not None: row = _row(f"{module_id}:step:{step_index + 1}:{key}", f"步骤 {number} · {label}", f"{path}/steps/{step_index}/{key}", item_index=step_index) row["_fieldKey"] = key rows.append(row) return rows if block_type == "evaluation-branches": rows = [] for branch_index, branch in enumerate(block.get("branches") or []): if not isinstance(branch, dict): continue subject = str(branch.get("subject") or f"方案 {branch_index + 1}") for key, label in (("conclusion", "结论"), ("achievements", "已达成"), ("problems", "需关注"), ("evidence", "依据")): if branch.get(key) not in (None, [], ""): rows.append(_row(f"{module_id}:branch:{branch_index + 1}:{key}", f"{subject} · {label}", f"{path}/branches/{branch_index}/{key}", item_index=branch_index)) return rows if block_type == "evaluation-matrix": return [ _row(f"{module_id}:row:{index + 1}", str((row or {}).get("criterion") or f"标准 {index + 1}"), f"{path}/rows/{index}", item_index=index) for index, row in enumerate(block.get("rows") or []) ] if block_type == "comparison": return [ _row(f"{module_id}:row:{index + 1}", str((row or {}).get("candidate") or f"方案 {index + 1}"), f"{path}/rows/{index}", item_index=index) for index, row in enumerate(block.get("rows") or []) ] if block_type == "artifact": return [ _row(f"{module_id}:artifact:{index + 1}", str((ref or {}).get("label") or f"产物 {index + 1}"), f"{path}/refs/{index}", item_index=index) for index, ref in enumerate(block.get("refs") or []) ] return [_row(f"{module_id}:record", None, path)] def _bindings_for_row( *, module_id: str, row: dict[str, Any], row_index: int, candidates: list[dict[str, Any]], resolver: SourceResolver, business_value: Any, narrow_event_text: bool = False, ) -> list[SourceBinding]: slot = str(row.get("_bindingSlot") or "") bindings: list[SourceBinding] = [] if slot: source_id, _, raw_index = slot.partition(":") source_module = next((module for module in candidates if str(module.get("id")) == source_id), None) source_bindings = list((source_module or {}).get("bindings") or []) try: binding = source_bindings[int(raw_index or 0)] except (IndexError, ValueError): binding = None if binding is not None: cloned = _clone_binding( binding, binding_id=f"{row['id']}:source:1", source_suffix=str(row.get("_sourceSuffix") or ""), resolver=resolver, ) bindings = [ _narrow_event_text_binding(cloned, business_value, resolver, module_id) if narrow_event_text else cloned ] if not bindings: item_index = row.get("_itemIndex") for source_module in candidates: items = ((source_module.get("business") or {}).get("items") or []) source_bindings = list(source_module.get("bindings") or []) if isinstance(item_index, int) and item_index < len(items): item = items[item_index] or {} item_bindings = list(item.get("bindings") or []) field_key = str(row.get("_fieldKey") or "") if field_key and item_bindings: event_id = _event_id(item_bindings[0].get("sourceId")) event_paths = { "action": "input.content.action", "summary": "input.content.thought_summary", "plan": "input.content.plan", "reasoning": "input.content.thought", } if event_id and field_key in event_paths: binding = resolver.binding_for_event( event_id, binding_id=f"{row['id']}:source:1", path=event_paths[field_key], role="process", ) found, selected = pointer_lookup( (resolver.sources.get(binding["sourceId"]) or {}).get("rawRecord"), str((binding.get("selector") or {}).get("path") or ""), ) if not found or selected != business_value: binding["transform"] = { "kind": "parsed", "operation": "creative-step-business-label-v1", "outputKey": field_key, } binding["evidence"]["confidence"] = "deterministic" return [binding] matched = _bindings_matching_value( item.get("value"), item_bindings, business_value ) bindings.extend(matched or item_bindings) elif ( isinstance(item_index, int) and isinstance(((source_module.get("business") or {}).get("value")), list) and item_index < len(source_bindings) ): # The readable Inspector may split one source list into one row # per item while CardData keeps it as one aggregate module. # The persisted source order is explicit, so bind the same # index instead of repeating every source on every row. bindings.append(source_bindings[item_index]) elif isinstance(item_index, int) and item_index < len(items): bindings.extend((items[item_index] or {}).get("bindings") or []) if not bindings: bindings = [ binding for source_module in candidates for binding in source_module.get("bindings") or [] ] physical_bindings = _unique_physical_bindings(bindings) usable_bindings = [ binding for binding in physical_bindings if binding.get("resolution") != "missing" ] if usable_bindings: physical_bindings = usable_bindings bindings = [ _narrow_event_text_binding(cloned, business_value, resolver, module_id) if narrow_event_text else cloned for index, binding in enumerate(physical_bindings, 1) for cloned in [ _clone_binding( binding, binding_id=f"{row['id']}:source:{index}", source_suffix="", resolver=resolver, ) ] ] return bindings def _mark_row_transform( bindings: list[SourceBinding], business_value: Any, resolver: SourceResolver, module_id: str, ) -> None: """Never call a normalized/combined business row a direct projection.""" mismatched: list[SourceBinding] = [] for binding in bindings: if (binding.get("transform") or {}).get("kind") != "direct": continue source = resolver.sources.get(str(binding.get("sourceId") or "")) or {} selector = binding.get("selector") or {} selected: Any = None found = False if selector.get("kind") == "json-pointer": found, selected = pointer_lookup( source.get("rawRecord"), str(selector.get("path") or "") ) elif selector.get("kind") == "whole-record": found, selected = True, source.get("rawRecord") elif selector.get("kind") == "text-span": selected = selector.get("exactText") found = True if found and selected != business_value: mismatched.append(binding) for binding in mismatched: binding["transform"] = ( { "kind": "combined", "operation": f"business-row:{module_id}", } if len(bindings) > 1 else { "kind": "parsed", "operation": "business-row-projection-v1", "outputKey": module_id, } ) binding["evidence"]["confidence"] = "deterministic" def _section_semantic_id(section_id: str) -> str: """Map task section instance ids to the stable business module ids.""" if not section_id.startswith("task-"): return section_id match = re.match(r"^task-([a-z-]+)-\d+$", section_id) kind = match.group(1) if match else "notes" return { "scope": "scope", "method": "method", "goal": "goal", "materials": "constraints", "context": "constraints", "constraints": "constraints", "avoid": "avoid", "notes": "deliverable", }.get(kind, "deliverable") def _block_semantic_id( card_kind: str, block: dict[str, Any], index: int ) -> str | None: """Return a stable semantic id for optional/reordered decision blocks.""" explicit = str(block.get("semanticId") or "").strip() if explicit: return explicit block_type = str(block.get("type") or "").strip() title = str(block.get("title") or "").strip() by_title = ({ "当前保存的创作方向": "direction", "最终方向": "direction", "根据以下信息做出的决策": "inputs", } if card_kind == "objective" else { "明确依据": "basis", "明确采用的数据取舍": "basis", "创作处理": "process", "可确认的上游信息": "available", "实际脚本改动": "changes", "实际产出": "candidate", "产出的候选表": "candidate", "存疑项": "uncertainty", "记录说明": "record-note", } if card_kind == "creative" else { "本轮目标构成": "goal", "根据以下信息做出的决策": "dependencies", } if card_kind == "round-goal" else { "本轮实施路线": "plan", "根据以下信息做出的决策": "basis", } if card_kind == "implementation-plan" else {}) if title in by_title: return by_title[title] if card_kind == "creative" and block_type == "creative-process": return "process" if card_kind == "creative" and block_type == "notice": return "record-note" return None def _narrow_event_text_binding( binding: SourceBinding, business_value: Any, resolver: SourceResolver, module_id: str = "", ) -> SourceBinding: """Narrow a complete implementation task binding to its exact visible section.""" if not isinstance(business_value, str) or not business_value.strip(): return binding event_id = _event_id(binding.get("sourceId")) selector = binding.get("selector") or {} if event_id is None or selector.get("kind") != "json-pointer": return binding path = str(selector.get("path") or "") source = resolver.sources.get(str(binding.get("sourceId") or "")) or {} found, raw_text = pointer_lookup(source.get("rawRecord"), path) if not found or not isinstance(raw_text, str): return binding exact_text = business_value if business_value in raw_text else _task_source_section(raw_text, module_id) if not exact_text: return binding narrowed = resolver.text_span_binding( event_id, binding_id=str(binding.get("id") or "implementation-task:span"), path=path.lstrip("/").replace("/", "."), exact_text=exact_text, role=str(binding.get("role") or "input"), ) if exact_text != business_value: narrowed["transform"] = { "kind": "parsed", "operation": "implementation-task-section-normalization-v1", "outputKey": module_id, } narrowed["evidence"]["confidence"] = "deterministic" return narrowed def _task_source_section(task: str, module_id: str) -> str | None: """Return the exact raw task section behind a normalized business block.""" module_id = _section_semantic_id(module_id) line_keys = { "scope": ("target", "实现范围"), "method": ("method", "实施方法"), } if module_id in line_keys: keys = "|".join(re.escape(key) for key in line_keys[module_id]) match = re.search(rf"(?im)^(?:{keys})\s*[:：]\s*(.+)$", task) return match.group(1).strip() if match else None headings = { "goal": (r"目标",), "constraints": ( r"取证方向与约束", r"参考依据与约束", r"严格约束", r"约束(?:（[^）]*）)?", ), "avoid": (r"禁止事项", r"需要避免"), "deliverable": (r"预期交付", r"交付"), } constraint_pattern = "|".join(headings["constraints"]) if module_id == "deliverable" and not re.search( rf"(?im)^(?:{'|'.join(headings['deliverable'])})\s*[:：]", task ): # Unlabelled task notes are the exact prefix before a later explicit # constraint section. The readable Inspector may normalize terms # inside it, so retain the raw prefix as a parsed text span. match = re.search( rf"(?ims)^(.*?)(?=\n\s*\n(?:{constraint_pattern})\s*[:：]|\Z)", task, ) return match.group(1).strip() if match else task.strip() names = headings.get(module_id) if not names: return None all_names = [name for values in headings.values() for name in values] start_pattern = "|".join(names) next_pattern = "|".join(all_names) match = re.search( rf"(?ims)^(?:{start_pattern})\s*[:：]\s*(.*?)(?=\n\s*\n(?:{next_pattern})\s*[:：]|\Z)", task, ) return match.group(1).strip() if match else None def _implementation_plan_revision_binding( *, row: dict[str, Any], business_value: Any, candidates: list[dict[str, Any]], resolver: SourceResolver, ) -> SourceBinding | None: """Bind the displayed final plan row to the revision that formed it. A round may call ``record_multipath_plan`` several times. The database value is authoritative, while the latest revision containing the exact displayed value is the formation record. This prevents round 4 from stopping at Events 4301/4369 and omitting the final Event 4434. """ revision = next( (module for module in candidates if str(module.get("id")) == "revisions"), None, ) event_ids = sorted( { event_id for binding in (revision or {}).get("bindings") or [] for event_id in [_event_id(binding.get("sourceId"))] if event_id is not None }, reverse=True, ) row_id = str(row.get("id") or "") path = "" if row_id.endswith(":mode"): path = "input.content.mode" elif row_id.endswith(":reasoning"): path = "input.content.note" else: match = re.search(r":route:(\d+):([A-Za-z]+)$", row_id) if match: route_index = int(match.group(1)) - 1 key = {"pathType": "path_type"}.get(match.group(2), match.group(2)) path = f"input.content.paths.{route_index}.{key}" if not path: return None fallback_event: int | None = None for event_id in event_ids: event = resolver.event_details.get(event_id) or {} found, selected = pointer_lookup(event, "/" + path.replace(".", "/")) if not found: continue fallback_event = fallback_event or event_id if selected != business_value: continue return resolver.binding_for_event( event_id, binding_id=f"{row_id}:revision", path=path, role="process", availability="produced-by-run", ) # The readable mode can normalize a one-route saved plan to "单路". # Retain the latest concrete revision but mark that normalization openly. if row_id.endswith(":mode") and fallback_event: binding = resolver.binding_for_event( fallback_event, binding_id=f"{row_id}:revision", path=path, role="process", availability="produced-by-run", ) binding["transform"] = { "kind": "parsed", "operation": "single-route-mode-normalization-v1", "outputKey": "mode", } binding["evidence"]["confidence"] = "deterministic" return binding return None def _bindings_matching_value( source_value: Any, bindings: list[SourceBinding], business_value: Any, ) -> list[SourceBinding]: source_leaves = _leaf_texts(source_value) wanted = set(_leaf_texts(business_value)) if not wanted or len(source_leaves) != len(bindings): return [] return [ binding for leaf, binding in zip(source_leaves, bindings) if leaf in wanted ] def _leaf_texts(value: Any) -> list[str]: if isinstance(value, str): return [value.strip()] if value.strip() else [] if isinstance(value, dict): result: list[str] = [] for child in value.values(): result.extend(_leaf_texts(child)) return result if isinstance(value, list): result = [] for child in value: result.extend(_leaf_texts(child)) return result return [str(value)] if value is not None else [] def _event_id(source_id: Any) -> int | None: value = str(source_id or "") if not value.startswith("event:"): return None try: return int(value.split(":", 1)[1]) except ValueError: return None def _clone_binding( binding: SourceBinding, *, binding_id: str, source_suffix: str, resolver: SourceResolver, ) -> SourceBinding: cloned: SourceBinding = deepcopy(binding) cloned["id"] = binding_id selector = cloned.get("selector") or {} if source_suffix and selector.get("kind") == "json-pointer": pointer = f"{str(selector.get('path') or '').rstrip('/')}{source_suffix}" source = resolver.sources.get(cloned["sourceId"]) or {} found, selected = pointer_lookup(source.get("rawRecord"), pointer) if found: cloned["selector"] = {"kind": "json-pointer", "path": pointer} source.setdefault("selectedValues", []).append({ "bindingId": binding_id, "fieldId": binding_id, "label": binding_id, "path": pointer, "value": selected, }) else: cloned["selector"] = { "kind": "unresolved", "reason": f"原始记录中无法定位字段 {pointer}", "code": "field-path-mismatch", } cloned["resolution"] = "partial" cloned["evidence"]["confidence"] = "unconfirmed" return cloned def _module( module_id: str, title: str, business_path: str, semantic_key: str, rows: list[dict[str, Any]], *, presentation: str = "text", default_open: bool = True, ) -> dict[str, Any]: return { "id": module_id, "title": title, "businessPath": business_path, "presentation": presentation, "rows": rows, "defaultOpen": default_open, "_semanticKey": semantic_key, } def _row( row_id: str, label: str | None, business_selector: str, *, item_index: int | None = None, binding_slot: str | None = None, source_suffix: str | None = None, ) -> dict[str, Any]: row: dict[str, Any] = { "id": row_id, "label": label, "businessSelector": business_selector, } if item_index is not None: row["_itemIndex"] = item_index if binding_slot is not None: row["_bindingSlot"] = binding_slot if source_suffix is not None: row["_sourceSuffix"] = source_suffix return row def _unique_bindings(bindings: list[SourceBinding]) -> list[SourceBinding]: result: list[SourceBinding] = [] seen: set[str] = set() for binding in bindings: key = str(binding.get("id") or "") if not key or key in seen: continue seen.add(key) result.append(binding) return result def _unique_physical_bindings(bindings: list[SourceBinding]) -> list[SourceBinding]: result: list[SourceBinding] = [] seen: set[str] = set() for binding in bindings: selector = binding.get("selector") or {} key = "|".join(( str(binding.get("sourceId") or ""), str(selector.get("kind") or ""), str(selector.get("path") or selector.get("exactText") or selector.get("reason") or selector), str(binding.get("role") or ""), )) if not key or key in seen: continue seen.add(key) result.append(binding) return result def _branch_id(value: str) -> int | None: import re match = re.search(r":branch:(\d+)", value) return _integer(match.group(1)) if match else None def _integer(value: Any) -> int | None: try: return int(value) except (TypeError, ValueError): return None