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

from __future__ import annotations

import re
from typing import Any


_HEADING = re.compile(r"(?m)^\s*#{1,6}\s+(.+?)\s*$")
_TARGET_SENTENCE = re.compile(r"(?im)^\s*[-*]?\s*目标语句\s*[：:]\s*(.+?)\s*$")
_FIELD_LINE = re.compile(r"(?im)^\s*[-*]?\s*([^：:\n]{1,24})\s*[：:]\s*(.+?)\s*$")
_BULLET = re.compile(r"^\s*(?:[-*+]\s+|[①②③④⑤⑥⑦⑧⑨⑩]\s*|\d{1,2}[.、)]\s*)")


def parse_objective_text(value: Any) -> dict[str, Any]:
    """Extract business sections from the persisted direction without inventing text."""

    text = _text(value)
    if not text:
        return {
            "headline": "创作方向未记录",
            "targets": [],
            "topicValue": None,
            "accountValue": None,
            "domainDimensions": [],
            "constraints": [],
            "constraintItems": [],
            "constraintGroups": [],
            "targetCount": 0,
            "domainDimensionCount": 0,
            "structured": False,
            "raw": "",
        }

    sections = _markdown_sections(text)
    targets = [_clean(match.group(1)) for match in _TARGET_SENTENCE.finditer(text)]
    if not targets:
        target_section = _first_section(sections, "创作总目标", "创作目标", "总目标")
        targets = _meaningful_lines(target_section)[:3]

    constraint_groups = _objective_constraint_groups(text)
    domain_dimensions = [
        str(group.get("dimensionName"))
        for group in constraint_groups
        if group.get("kind") == "domain" and group.get("dimensionName")
    ]
    if not domain_dimensions:
        domain_section = _first_section(sections, "领域评估维度", "领域维度")
        domain_dimensions = [
            line
            for line in _meaningful_lines(domain_section)
            if not re.match(
                r"^(?:评估内容|通过标准|强制性|前置条件)\s*[：:]", line
            )
        ]
    constraint_items = [
        {"group": group["title"], **item}
        for group in constraint_groups
        for item in group.get("items") or []
        if item.get("label") in {"通过标准", "强制性", "前置条件", "约束条件"}
    ]
    constraints = [item["value"] for item in constraint_items]
    topic_value = _section_preview(_first_section(sections, "选题价值主张"))
    account_value = _section_preview(_first_section(sections, "账号价值主张"))
    headline = targets[0] if targets else _first_business_sentence(text)
    return {
        "headline": headline or _clean(text),
        "targets": _unique(targets),
        "topicValue": topic_value,
        "accountValue": account_value,
        "domainDimensions": _unique(domain_dimensions),
        "constraints": _unique(constraints),
        "constraintItems": _unique_records(constraint_items),
        "constraintGroups": constraint_groups,
        "targetCount": len(_unique(targets)),
        "domainDimensionCount": len(_unique(domain_dimensions)),
        "structured": bool(sections or targets),
        "raw": text,
    }


def parse_goal_text(value: Any) -> dict[str, Any]:
    """Return the persisted round goal as one authoritative business statement."""

    text = _text(value)
    if not text:
        return {
            "headline": "本轮目标未记录",
            "focusItems": [],
            "raw": "",
        }
    return {
        "headline": text,
        "focusItems": [],
        "raw": text,
    }


def plan_path_summary(path: Any, index: int) -> str:
    if not isinstance(path, dict):
        return _clean(path) or f"方案 {index}"
    path_index = path.get("path_index") or index
    target = _clean(path.get("target"))
    action = _business_plan_term(path.get("action"))
    method = _business_plan_term(path.get("method"))
    detail = plan_path_detail(path)
    return f"方案 {path_index}：{detail}" if detail else f"方案 {path_index}"


def plan_path_detail(path: Any) -> str:
    if not isinstance(path, dict):
        return _clean(path)
    target = _clean(path.get("target"))
    action = _business_plan_term(path.get("action"))
    method = _business_plan_term(path.get("method"))
    return " · ".join(value for value in (action, target, method) if value)


def plan_mode_label(mode: Any, path_count: int) -> str:
    if path_count == 1:
        return "单路"
    value = _clean(mode)
    aliases = {"race": "竞争", "赛马": "竞争", "divide": "分工"}
    return aliases.get(value.lower(), value or "方式未记录")


def preview(value: Any, limit: int = 140) -> str | None:
    text = _clean(value)
    if not text:
        return None
    if len(text) <= limit:
        return text
    cut = text[: limit + 1]
    punctuation = max(cut.rfind(token) for token in "。！？；")
    if punctuation >= max(24, int(limit * 0.55)):
        return cut[: punctuation + 1]
    return text[: limit - 1].rstrip() + "…"


def _markdown_sections(text: str) -> dict[str, str]:
    matches = list(_HEADING.finditer(text))
    sections: dict[str, str] = {}
    for index, match in enumerate(matches):
        title = _clean(match.group(1)).replace("**", "")
        end = matches[index + 1].start() if index + 1 < len(matches) else len(text)
        body = text[match.end() : end].strip()
        if title and body:
            sections[title] = body
    return sections


def _first_section(sections: dict[str, str], *labels: str) -> str | None:
    for label in labels:
        for title, body in sections.items():
            if label in title:
                return body
    return None


def _objective_constraint_groups(text: str) -> list[dict[str, Any]]:
    headings = list(_HEADING.finditer(text))
    groups: list[dict[str, Any]] = []
    for index, heading in enumerate(headings):
        title = _clean(heading.group(1)).replace("**", "")
        if not re.match(r"^(?:目标\s*\d+|领域维度\s*\d+)", title):
            continue
        end = headings[index + 1].start() if index + 1 < len(headings) else len(text)
        body = text[heading.end() : end]
        fields = [
            {"label": label.strip(), "value": _clean(value)}
            for label, value in _FIELD_LINE.findall(body)
            if label.strip() in {
                "name",
                "维度名",
                "名称",
                "评估内容",
                "通过标准",
                "强制性",
                "前置条件",
                "约束条件",
            }
            and _clean(value)
        ]
        dimension_name = next(
            (
                item["value"]
                for item in fields
                if item["label"].lower() in {"name", "维度名", "名称"}
            ),
            None,
        )
        business_items = [
            item
            for item in fields
            if item["label"].lower() not in {"name", "维度名", "名称"}
        ]
        if not business_items and not dimension_name:
            continue
        groups.append(
            {
                "title": title,
                "kind": "domain" if title.startswith("领域维度") else "target",
                "dimensionName": dimension_name,
                "items": business_items,
            }
        )
    return groups


def _section_preview(value: str | None) -> str | None:
    lines = _meaningful_lines(value)
    return lines[0] if lines else None


def _meaningful_lines(value: Any) -> list[str]:
    result: list[str] = []
    for raw in _text(value).splitlines():
        line = _clean(_BULLET.sub("", raw))
        if not line or line.startswith("#"):
            continue
        if re.match(r"^(?:评估维度|评估内容|通过标准|强制性|前置条件|name)\s*[：:]", line, re.I):
            continue
        result.append(line)
    return _unique(result)


def _first_business_sentence(text: str) -> str:
    without_headings = _HEADING.sub("", text)
    lines = _meaningful_lines(without_headings)
    return lines[0] if lines else _clean(text)


def _clean(value: Any) -> str:
    text = str(value or "").replace("\\n", "\n")
    text = re.sub(r"[`*_]", "", text)
    text = re.sub(r"\s+", " ", text).strip(" #：:")
    return text


def _text(value: Any) -> str:
    return str(value or "").replace("\r\n", "\n").replace("\\n", "\n").strip()


def _unique(values: list[str]) -> list[str]:
    result: list[str] = []
    for value in values:
        if value and value not in result:
            result.append(value)
    return result


def _unique_records(values: list[dict[str, str]]) -> list[dict[str, str]]:
    result: list[dict[str, str]] = []
    seen: set[tuple[str, str, str]] = set()
    for value in values:
        key = (
            value.get("group") or "",
            value.get("label") or "",
            value.get("value") or "",
        )
        if key not in seen:
            seen.add(key)
            result.append(value)
    return result


def _business_plan_term(value: Any) -> str:
    text = _clean(value)
    return {
        "增": "新增",
        "改": "修改",
        "删": "删除",
        "组": "组合",
        "产生脉络": "形成段落结构",
        "产生元素": "补充脚本元素",
    }.get(text, text)