script-build-and-framework-tmp/visualization/backend/app/business_detail_text.py

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"(?<![a-z0-9_]){re.escape(field)}(?![a-z0-9_])",
            label,
            text,
            flags=re.I,
        )
    text = re.sub(r"(?i)(?<![a-z0-9_])branch_id\s*[=:：]\s*(\d+)", r"方案 \1", text)
    text = re.sub(r"(?i)(?<![a-z0-9_])branch\s*#?\s*(\d+)", r"方案 \1", text)
    text = re.sub(r"(?i)(?<![a-z0-9_])base(?![a-z0-9_])", "主脚本", text)
    text = re.sub(
        r"选\s+方案\s*(\d+)\s*合入(?:\s*主脚本)?\s*[，,、]\s*弃\s+方案\s*(\d+)",
        r"采用方案 \1 并合入主脚本，方案 \2 未采用",
        text,
    )
    text = re.sub(
        r"组\s+方案\s*(\d+)\s*与\s*方案\s*(\d+)\s*合入",
        r"组合方案 \1 与方案 \2，并合入主脚本",
        text,
    )
    text = re.sub(r"(?i)(?<![a-z0-9_])target\s*[=:：]", "任务目标：", text)
    text = re.sub(r"(?i)(?<![a-z0-9_])target(?![a-z0-9_])", "任务目标", text)
    text = re.sub(r"(?i)(?<![a-z0-9_])scope\s*[=:：]", "账号：", text)
    text = re.sub(r"(?i)(?<![a-z0-9_])null(?![a-z0-9_])", "未填写", text)
    text = (
        text.replace("主脚本 脚本表", "主脚本")
        .replace("账号 账号：", "账号：")
        .replace("同一 任务目标", "同一个任务")
    )
    text = re.sub(r"(?i)patch\s*分支", "候选方案", text)
    text = re.sub(r"[ \t]+([，。；！？、：）】,.])", r"\1", text)
    return _tidy(text)


def branch_task_text(value: Any) -> 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 ""