"""把脚本 stdout 和 MCP tool 返回值归一化为 ActionResult。""" from __future__ import annotations import json import re from typing import Any from .constants import SENSITIVE_KEYS from .models import ActionResult, BackendName PENDING_CONFIRMATION_RE = re.compile(r"PENDING_AGENT_CONFIRMATION$(?P[^)]*)$") KEY_VALUE_RE = re.compile(r"^(?P[A-Za-z_][A-Za-z0-9_]*)=(?P.*)$") def redact_text(text: str) -> str: """对文本中的敏感字段值进行脱敏。""" redacted = text for key in SENSITIVE_KEYS: redacted = re.sub( rf"({re.escape(key)}\s*[:=]\s*)([^\s&]+)", rf"\1***", redacted, flags=re.IGNORECASE, ) return redacted def parse_key_values(text: str) -> dict[str, Any]: """解析脚本输出中的 KEY=VALUE 行，重复 IP 会聚合为列表。""" values: dict[str, Any] = {} for raw_line in text.splitlines(): line = raw_line.strip() match = KEY_VALUE_RE.match(line) if not match: continue key = match.group("key") value = match.group("value") if key == "IP": values.setdefault("IP", []).append(value) else: values[key] = value return values def normalize_mcp_values(payload: Any) -> dict[str, Any]: """把 MCP 返回值归一化为脚本兼容的字段名。""" if isinstance(payload, str): try: payload = json.loads(payload) except json.JSONDecodeError: return parse_key_values(payload) if not isinstance(payload, dict): return {"RESULT": payload} values: dict[str, Any] = dict(payload) aliases = { "hashCode": "HASH_CODE", "nodeUrl": "NODE_URL", "rateOfProgress": "RATE_OF_PROGRESS", "logFile": "LOG_FILE", "action": "ACTION", "result": "RESULT", "success": "SUCCESS", "message": "MESSAGE", } for source, target in aliases.items(): if source in values and target not in values: values[target] = values[source] return values def parse_script_result( action: str, stdout: str, stderr: str, exit_code: int, backend: BackendName = "script", tool_name: str = "", ) -> ActionResult: """解析脚本执行结果，并识别待人工确认标记。""" raw_output = redact_text("\n".join(part for part in (stdout, stderr) if part)) values = parse_key_values(stdout) pending = PENDING_CONFIRMATION_RE.search(stdout) or PENDING_CONFIRMATION_RE.search(stderr) if pending: values["PENDING_AGENT_CONFIRMATION"] = pending.group(0) ok = exit_code == 0 and not pending error_summary = "" if ok else _summarize_error(stderr, stdout, pending.group(0) if pending else "") return ActionResult( action=action, backend=backend, tool_name=tool_name or action, ok=ok, values=values, exit_code=exit_code, stdout=redact_text(stdout), stderr=redact_text(stderr), raw_output=raw_output, error_summary=error_summary, ) def parse_mcp_result( action: str, payload: Any, *, ok: bool = True, tool_name: str = "", error: str = "", ) -> ActionResult: """解析 MCP tool 返回值，并包装为统一 ActionResult。""" values = normalize_mcp_values(payload) raw_output = redact_text(json.dumps(payload, ensure_ascii=False, default=str)) return ActionResult( action=action, backend="mcp", tool_name=tool_name or action, ok=ok, values=values, exit_code=0 if ok else 1, stdout=raw_output if ok else "", stderr=redact_text(error), raw_output=raw_output, error_summary="" if ok else redact_text(error or "MCP tool 执行失败"), ) def _summarize_error(stderr: str, stdout: str, pending: str) -> str: """从 stderr/stdout/确认标记中提取简短错误摘要。""" if pending: return pending for text in (stderr, stdout): for line in reversed(text.splitlines()): stripped = line.strip() if stripped: return redact_text(stripped) return "action 执行失败"