from __future__ import annotations import json import re from typing import Any from .evaluation_report_parser import EvaluationReportParser from .runtime_tool_catalog import TOOL_CATALOG _STATUS_LABELS = { "merged": "已采用", "parked": "暂存", "discarded": "未采用", "open": "等待处理", "running": "进行中", "success": "成功", "failure": "失败", "failed": "失败", "ok": "完成", "completed": "完成", } _FIELD_LABELS = { "full_description": "详细执行描述", "content_range": "内容范围", "theme_elements": "主题信息", "form_elements": "形式信息", "function_elements": "作用信息", "feeling_elements": "感受信息", "element_count": "元素数量", "theme": "主题", "form": "形式", "function": "作用", "feeling": "感受", } def business_text(value: Any) -> str: """Translate stored business prose without inventing a new conclusion.""" text = _string(value) if not text: return "" text = _decode_visible_escapes(text) for field, label in _FIELD_LABELS.items(): text = re.sub( rf"(? str: """Remove implementation headers while preserving the complete business brief.""" text = business_text(value) if not text: return "" kept: list[str] = [] for line in text.splitlines(): stripped = line.strip() if re.match(r"^(?:路序号|action|path_type)\s*[：:]", stripped, re.I): continue if re.match(r"^任务目标\s*[：:]", stripped): stripped = re.sub(r"^任务目标\s*[：:]\s*", "", stripped) if stripped: kept.extend(["### 实现范围", stripped]) continue stripped = re.sub(r"^取证方向与约束\s*[：:]?", "### 参考依据与约束", stripped) stripped = re.sub(r"^总目标方向(?:（[^）]*）)?\s*[：:]?", "### 整体方向", stripped) stripped = re.sub(r"^本轮当前目标\s*[：:]?", "**本轮重点：**", stripped) stripped = re.sub(r"^禁止事项\s*[：:]?\s*", "### 需要避免\n", stripped) kept.append(stripped) return _tidy("\n".join(kept)) def branch_task_sections(value: Any) -> list[dict[str, str]]: """Split a delegated task into optional display sections without data loss. Real runs mostly follow the dispatch contract, but some repair tasks are free-form. No field is therefore required: recognized labels become sections and every remaining line stays in a generic task note. """ text = branch_task_text(value) if not text: return [] sections: list[dict[str, str]] = [] current: dict[str, Any] | None = None recognized = False def flush() -> None: nonlocal current if not current: return content = _tidy("\n".join(current.pop("lines", []))) if content: sections.append({**current, "content": content}) current = None def begin(kind: str, title: str, inline: str = "") -> None: nonlocal current, recognized flush() recognized = True current = {"kind": kind, "title": title, "lines": []} if inline.strip(): current["lines"].append(inline.strip()) for line in text.splitlines(): stripped = line.strip() heading = re.match(r"^#{1,6}\s+(.+?)\s*$", stripped) if heading: label = heading.group(1).strip() if label == "实现范围": begin("scope", label) continue if label in {"参考依据与约束", "整体方向"}: begin("constraints", label) continue if label == "需要避免": begin("avoid", label) continue labelled = _task_section_label(stripped) if labelled: kind, title, inline = labelled begin(kind, title, inline) continue if current is None: current = {"kind": "notes", "title": "任务说明", "lines": []} current["lines"].append(line) flush() if not recognized: return [{"kind": "notes", "title": "任务说明", "content": text}] return sections def _task_section_label(value: str) -> tuple[str, str, str] | None: plain = re.sub(r"^\*\*(.+?)\*\*\s*([：:]?.*)$", r"\1\2", value).strip() patterns = ( ("method", "实施方法", r"(?i)^method\s*[：:]\s*(.*)$"), ("goal", "目标", r"^目标\s*[：:]\s*(.*)$"), ("materials", None, r"^(必用素材(?:（[^\n]*）)?)\s*[：:]\s*(.*)$"), ("context", None, r"^(前情提要(?:（[^\n]*）)?)\s*[：:]\s*(.*)$"), ( "constraints", None, r"^((?:极其重要的|统一基调|严格|重要)?约束(?:（[^\n]*）)?)\s*[：:]\s*(.*)$", ), ) for kind, fixed_title, pattern in patterns: match = re.match(pattern, plain) if not match: continue if fixed_title is not None: return kind, fixed_title, match.group(1) return kind, match.group(1), match.group(2) return None def assessment_text(value: Any) -> str: """Keep the implementer's declaration; remove thought/process and cost telemetry.""" text = clean_report(value) if not text: return "" marker = re.search(r"(?i)#{1,6}\s*完成度申报", text) if marker: text = text[marker.end() :] else: text = re.sub( r"(?is)^.*?(?=^\s*[-*]\s*\*{0,2}(?:完成|未尽|存疑)\*{0,2}\s*[：:])", "", text, ) return _tidy(text) def clean_report(value: Any) -> str: """Turn evaluator/agent Markdown into business prose; raw input remains technical.""" text = business_text(value) if not text: return "" text = re.sub(r"(?ms)^\s*```[^\n]*\n.*?^\s*```\s*$", "", text) text = re.sub(r"(?ms)^\s*```[^\n]*\n.*\Z", "", text) lines: list[str] = [] skip_stats = False for line in text.splitlines(): stripped = line.strip() if re.match(r"^#{1,6}\s*(?:委托任务完成|评估报告)\b", stripped): continue if re.match(r"^(?:\*\*)?执行统计(?:\*\*)?\s*[：:]?", stripped): skip_stats = True continue if skip_stats and ( not stripped or re.match(r"^[-*]\s*(?:消息数|Tokens?|成本)\s*[：:]", stripped, re.I) ): continue if skip_stats and stripped: skip_stats = False if re.match(r"^[-*]?\s*(?:消息数|Tokens?|Token\s*usage|成本)\s*[：:]", stripped, re.I): continue if "分支元信息" in stripped: continue line = re.sub( r"[（(]\s*段落\s*(?:ID|id)\s*[:：]\s*[\d,，、\s]+[）)]", "", line, ) line = re.sub(r"(?i)script_build_id\s*[=:]\s*\d+\s*[·|,，]?", "", line) line = re.sub(r"(?i)\bpost_id\s*[=:：]\s*[^\s,，;；)）]+", "来源记录", line) line = re.sub(r"(?i)\bround(?:_index)?\s*[=:]\s*(\d+)\b", r"第 \1 轮", line) line = re.sub(r"(?i)\b(?:ID|id)\s*[=:：]\s*\d+\b", "", line) line = re.sub(r"[（(]\s*[，,·|\s-]*[）)]", "", line) line = re.sub(r"(?<=\d)中", " 中", line) line = re.sub(r"`?\[元素:ID\]`?\s*格式", "规范格式", line, flags=re.I) lines.append(line.rstrip()) return _tidy("\n".join(lines)) def event_business_sections(event: dict[str, Any]) -> list[dict[str, str]]: name = str(event.get("event_name") or "") task = _event_value(event, "input", "task") result = _event_value(event, "output", "summary") if name == "script_multipath_evaluator": report = EvaluationReportParser().parse(result, kind="multipath") scope = "、".join(report["subjects"]) or branch_task_text(task) return _compact_sections( [ ("scope", "评审了哪些方案", scope, "summary"), ( "conclusion", "各方案结论", _evaluation_item_text(report["itemConclusions"]), "default", ), ("recommendation", "评审建议", report["recommendation"], "report"), ] ) if name == "script_evaluator": report = EvaluationReportParser().parse(result, kind="overall") return _compact_sections( [ ("conclusion", "主脚本是否达标", report["conclusion"], "summary"), ("scope", "评审范围", branch_task_text(task), "default"), ("achievements", "已达成", "\n".join(report["achievements"]), "report"), ( "problems", "仍需解决", _evaluation_problem_text(report["problems"]), "report", ), ("next", "下一步建议", report["nextGoal"], "report"), ] ) action = friendly_event_name(event) request = branch_task_text(task) outcome = event_result_text(event) return _compact_sections( [ ("action", "做了什么", action, "summary"), ("request", "处理内容", request, "default"), ("result", "得到什么", outcome, "report"), ] ) def _evaluation_item_text(items: list[Any]) -> str: lines: list[str] = [] for item in items: if not isinstance(item, dict): continue subject = business_text(item.get("subject") or item.get("label")) conclusion = business_text(item.get("conclusion") or item.get("summary")) if subject and conclusion: lines.append(f"{subject}：{conclusion}") return "\n".join(lines) def _evaluation_problem_text(items: list[Any]) -> str: values = [] for item in items: if isinstance(item, dict): value = business_text(item.get("summary") or item.get("label")) else: value = business_text(item) if value: values.append(value) return "\n".join(values) def source_markdown(sources: list[Any]) -> str: items: list[str] = [] for source in sources: if not isinstance(source, dict): continue source_type = business_text(source.get("data_type") or source.get("type")) or "参考信息" content = business_text( source.get("data_content") or source.get("content") or source.get("summary") ) if content: items.append(f"- **{source_type}：** {content}") return "\n".join(items) def decisions_markdown(values: list[Any]) -> str: items = [business_text(value) for value in values] return "\n\n".join(f"{index}. {value}" for index, value in enumerate(items, 1) if value) def branches_markdown(branches: list[dict[str, Any]], *, raw: bool = False) -> str: parts: list[str] = [] for index, branch in enumerate(branches, 1): branch_id = branch.get("branch_id") if raw else branch.get("branchId") title = ( branch.get("target") or branch.get("impl_task") if raw else branch.get("title") or (branch.get("summary") or {}).get("task") ) status = branch.get("status") output = None if raw else (branch.get("summary") or {}).get("output") disposition = ( branch.get("decision_reasoning") if raw else (branch.get("outcome") or {}).get("reasoning") ) heading = f"### 方案 {branch_id or index}" if status: heading += f" · {status_label(status)}" rows = [heading, business_text(title)] if output: rows.append(f"**候选产出：** {business_text(output)}") if disposition: label = "选择理由" if raw else "最终选择" rows.append(f"**{label}：** {business_text(disposition)}") parts.append("\n\n".join(item for item in rows if item)) return "\n\n".join(parts) def batches_markdown(batches: list[dict[str, Any]]) -> str: if not batches: return "未记录主 Agent 多路决策。" parts = [] for index, batch in enumerate(batches, 1): branch_ids = "、".join(str(value) for value in batch.get("branchIds") or []) lines = [f"### 决策 {index}"] if branch_ids: lines.append(f"**比较方案：** {branch_ids}") if batch.get("decision"): lines.append(f"**最终决定：** {business_text(batch['decision'])}") if batch.get("reasoning"): lines.append(f"**原因：** {business_text(batch['reasoning'])}") parts.append("\n\n".join(lines)) return "\n\n".join(parts) def status_label(value: Any) -> str: raw = str(value or "").strip() return _STATUS_LABELS.get(raw.lower(), business_text(raw) or "状态未记录") def friendly_event_name(event: dict[str, Any]) -> str: name = str(event.get("event_name") or "") title = str(event.get("title") or "") labels = { "script_evaluator": "对本轮主脚本进行整体评审", "script_multipath_evaluator": "比较候选方案并给出独立评审", "record_data_decision": "记录实现方案的数据取舍", "get_branch_data_decisions": "读取方案的数据取舍记录", "get_script_build_context": "读取本次构建的业务背景", } if name in labels: return labels[name] definition = TOOL_CATALOG.get(name) if definition: if definition.category == "decision-input": return f"读取{definition.business_label}" return definition.business_label cleaned = re.sub(r"\s*\([^)]*\)\s*$", "", title).strip() if cleaned and not re.search(r"[a-z]+_[a-z_]", cleaned, re.I): return business_text(cleaned) lowered = name.lower() if any(token in lowered for token in ("search", "find", "query", "retrieve")): return "查找相关参考信息" if lowered.startswith("get_"): return "读取相关业务资料" if lowered.startswith(("record_", "save_")): return "保存本步骤的业务结果" return "完成一次运行活动" def event_result_text(event: dict[str, Any]) -> str: name = str(event.get("event_name") or "") status = str(event.get("status") or "").lower() output = _event_object(event, "output") if status in {"error", "failed", "failure"} or output.get("success") is False: message = business_text(output.get("message") or output.get("error")) return f"执行失败。{message}" if message else "执行失败，具体原因请查看技术详情。" if name == "get_script_direction": direction = _event_value(event, "output", "script_direction") return clean_report(direction) or "已读取当前创作方向。" if name == "get_account_script_section_patterns": summary = _event_value(event, "output", "分段规律摘要") return clean_report(summary) or "已读取账号常用的段落结构。" if name == "get_domain_info": count = _list_count(output.get("domain_info")) return f"已读取 {count} 条已核实领域信息。" if count is not None else "已读取已核实的领域信息。" if name == "get_script_snapshot": paragraphs = _list_count(output.get("paragraphs")) elements = _list_count(output.get("elements")) if paragraphs is not None and elements is not None: return f"已读取当前主脚本，共 {paragraphs} 个段落、{elements} 个元素。" return "已读取当前主脚本内容。" if name == "get_branch_data_decisions": count = _list_count(output.get("decisions")) return f"找到 {count} 条数据取舍记录。" if count is not None else "已读取方案的数据取舍记录。" summary = output.get("summary") or output.get("message") if isinstance(summary, str): readable = clean_report(summary) if readable: return readable count = _first_count(output) if count is not None: return f"已完成，共返回 {count} 条结果。" return "已完成。具体输入输出保留在技术详情中。" def _event_value(event: dict[str, Any], side: str, key: str | None = None) -> Any: body = event.get(side) content = body.get("content") if isinstance(body, dict) else body value = _nested_value(content, key) if value not in (None, ""): return value if key and isinstance(content, str) and not content.lstrip().startswith(("{", "[")): return content preview = event.get(f"{side}_preview") if key: extracted = _json_string_preview(preview, key) if extracted not in (None, ""): return extracted return _nested_value(preview, key) def _event_object(event: dict[str, Any], side: str) -> dict[str, Any]: body = event.get(side) content = body.get("content") if isinstance(body, dict) else body for value in (content, event.get(f"{side}_preview")): if isinstance(value, dict): return value if isinstance(value, str): try: parsed = json.loads(value) except json.JSONDecodeError: continue if isinstance(parsed, dict): return parsed return {} def _nested_value(value: Any, key: str | None) -> Any: if key is None: return value if isinstance(value, dict): return value.get(key) if isinstance(value, str): try: parsed = json.loads(value) except (TypeError, json.JSONDecodeError): return None return parsed.get(key) if isinstance(parsed, dict) else None return None def _json_string_preview(value: Any, key: str) -> str | None: text = _string(value) if not text: return None match = re.search(rf'"{re.escape(key)}"\s*:\s*"', text) if not match: return None start = match.end() escaped = False end = None for index in range(start, len(text)): char = text[index] if char == '"' and not escaped: end = index break if char == "\\" and not escaped: escaped = True else: escaped = False fragment = text[start:end] if end is not None else text[start:].rstrip("…") try: return json.loads(f'"{fragment}"') except json.JSONDecodeError: return _decode_visible_escapes(fragment) def _compact_sections( values: list[tuple[str, str, Any, str]], ) -> list[dict[str, str]]: return [ {"id": section_id, "title": title, "content": content, "variant": variant} for section_id, title, raw, variant in values if (content := business_text(raw)) ] def _list_count(value: Any) -> int | None: return len(value) if isinstance(value, list) else None def _first_count(value: dict[str, Any]) -> int | None: for key in ("count", "result_count", "path_count", "source_count", "total"): count = value.get(key) if isinstance(count, int) and not isinstance(count, bool): return count return None def _decode_visible_escapes(value: str) -> str: return value.replace("\\r\\n", "\n").replace("\\n", "\n").replace('\\"', '"') def _tidy(value: str) -> str: value = re.sub(r"[ \t]+\n", "\n", value) value = re.sub(r"\n{3,}", "\n\n", value) return value.strip() def _unique(values: list[str]) -> list[str]: return list(dict.fromkeys(values)) def _string(value: Any) -> str: return value.strip() if isinstance(value, str) else ""