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

from __future__ import annotations

import json
import re
from dataclasses import dataclass
from typing import Any, Literal


EvaluationKind = Literal["multipath", "overall", "unknown"]


@dataclass(frozen=True)
class _Section:
    title: str
    body: str


class EvaluationReportParser:
    """Parse evaluator Markdown into business-safe, deterministic fields.

    The parser deliberately does not accept a build or round identifier. Runtime
    metadata remains the only source for association; identifiers written inside
    a report are treated as technical prose and ignored.
    """

    VERSION = 1

    def parse(
        self,
        value: Any,
        *,
        kind: EvaluationKind = "unknown",
    ) -> dict[str, Any]:
        report = _report_text(value)
        visible = _remove_code_fences(report)
        sections = _sections(visible)
        tables = _tables(visible)
        branch_evaluations = _branch_evaluations(sections)
        comparison_rows = _comparison_rows(tables)

        subjects = _subjects(sections)
        criteria = _collect_sections(
            sections,
            ("评审标准", "评估标准", "评价标准", "评审维度", "评估维度", "评价维度", "维度评估", "领域评估维度"),
        )
        achievements = _collect_positive_sections(
            sections,
            ("已达成", "达成情况", "目标达成", "目标推进", "完成情况", "通过项", "达成项", "主要优点"),
        )
        problems = _problem_sections(sections)
        if not problems:
            inline_problem = _labeled_value(
                visible,
                ("问题详情", "主要问题", "未通过项", "遗留问题", "存疑项", "问题"),
            )
            if inline_problem:
                problems.append({"summary": inline_problem})
        item_conclusions = _item_conclusions(sections)
        if branch_evaluations:
            item_conclusions = [
                {
                    "subject": item["subject"],
                    "conclusion": item.get("conclusion"),
                    "achievements": item.get("achievements") or [],
                    "problems": item.get("problems") or [],
                }
                for item in branch_evaluations
            ]

        for table in tables:
            _merge_table(
                table,
                criteria=criteria,
                item_conclusions=item_conclusions,
                achievements=achievements,
                problems=problems,
            )

        problems = [
            item
            for item in problems
            if not _means_no_problem(str(item.get("summary") or ""))
        ]
        for item in item_conclusions:
            item["problems"] = [
                problem
                for problem in item.get("problems") or []
                if not _means_no_problem(str(problem))
            ]

        conclusion = _labeled_value(
            visible,
            ("整体结论", "总结论", "评审结论", "评估结论"),
        )
        recommendation = _section_value(
            sections,
            ("评审建议", "对比与建议", "跨方案对比", "方案对比", "采纳建议", "组合建议", "建议采用", "建议合入", "推荐"),
        )
        next_goal = _section_value(
            sections,
            ("下一步引领目标", "下一轮引领目标", "下一轮目标", "下一步目标", "引领目标"),
        )

        # Some reports use bold inline labels without a Markdown heading.
        if not recommendation:
            recommendation = _labeled_value(
                visible,
                ("采纳建议", "组合建议", "评审建议", "建议采用", "建议合入", "推荐"),
            )
        if not next_goal:
            next_goal = _labeled_value(
                visible,
                ("下一步引领目标", "下一轮引领目标", "下一轮目标", "下一步目标", "引领目标"),
            )

        result = {
            "parserVersion": self.VERSION,
            "kind": kind,
            "subjects": _unique(subjects),
            "criteria": _unique(criteria),
            "itemConclusions": _unique_dicts(item_conclusions),
            "achievements": _unique(achievements),
            "problems": _unique_dicts(problems),
            "conclusion": _business_value(conclusion),
            "recommendation": _business_value(recommendation),
            "nextGoal": _business_value(next_goal),
            "notices": [],
        }
        if branch_evaluations:
            result["branchEvaluations"] = branch_evaluations
        if comparison_rows:
            result["comparisonRows"] = comparison_rows
        if not any(
            result[key]
            for key in (
                "subjects",
                "criteria",
                "itemConclusions",
                "achievements",
                "problems",
                "conclusion",
                "recommendation",
                "nextGoal",
            )
        ):
            result["notices"].append("unparsed-report")
        return result


def _means_no_problem(value: str) -> bool:
    sentences = [
        re.sub(r"\s+", "", item)
        for item in re.split(r"[。！？!?；;]+", value)
        if item.strip()
    ]
    if not sentences:
        return False
    marker = re.fullmatch(
        r"(?:无|没有|无(?:显著|明显)?(?:质量)?问题|无硬伤|无未通过项|未发现问题|未发现硬伤)",
        sentences[0],
    )
    if not marker:
        return False
    remainder = "".join(sentences[1:])
    return not re.search(r"(?:但|不过|仍|需|建议|风险|问题|不足|缺失|待|未)", remainder)


def _report_text(value: Any) -> str:
    if value is None:
        return ""
    if isinstance(value, dict):
        summary = value.get("summary")
        return str(summary) if isinstance(summary, str) else ""
    text = str(value).replace("\r\n", "\n").replace("\r", "\n").strip()
    if text.startswith("{"):
        try:
            payload = json.loads(text)
        except (TypeError, ValueError):
            return text
        if isinstance(payload, dict) and isinstance(payload.get("summary"), str):
            return payload["summary"]
    return text.replace("\\n", "\n")


def _remove_code_fences(text: str) -> str:
    return re.sub(r"(?ms)^\s*```[^\n]*\n.*?^\s*```\s*$", "", text)


def _sections(text: str) -> list[_Section]:
    lines = text.splitlines()
    sections: list[_Section] = []
    current_title = ""
    body: list[str] = []

    def flush() -> None:
        nonlocal body
        if current_title or any(line.strip() for line in body):
            sections.append(_Section(_clean_heading(current_title), "\n".join(body).strip()))
        body = []

    for line in lines:
        heading = re.match(r"^\s*#{1,6}\s+(.+?)\s*#*\s*$", line)
        if heading:
            flush()
            current_title = heading.group(1)
            continue
        bold_heading = re.match(
            r"^\s*(?:[-*+]\s*)?\*\*([^*\n]{2,40})\*\*\s*(?:[（(][^）)\n]{0,40}[）)])?\s*[：:]?\s*$",
            line,
        )
        if bold_heading:
            flush()
            current_title = bold_heading.group(1)
            continue
        body.append(line)
    flush()
    return sections


def _tables(text: str) -> list[list[dict[str, str]]]:
    lines = text.splitlines()
    tables: list[list[dict[str, str]]] = []
    index = 0
    while index + 1 < len(lines):
        headers = _table_cells(lines[index])
        separator = _table_cells(lines[index + 1])
        if not headers or not separator or not all(re.fullmatch(r":?-{3,}:?", cell) for cell in separator):
            index += 1
            continue
        rows: list[dict[str, str]] = []
        index += 2
        while index < len(lines):
            cells = _table_cells(lines[index])
            if not cells:
                break
            cells += [""] * max(0, len(headers) - len(cells))
            rows.append(dict(zip((_plain(cell) for cell in headers), cells)))
            index += 1
        if rows:
            tables.append(rows)
    return tables


def _table_cells(line: str) -> list[str]:
    stripped = line.strip()
    if "|" not in stripped:
        return []
    return [cell.strip() for cell in stripped.strip("|").split("|")]


def _subjects(sections: list[_Section]) -> list[str]:
    values: list[str] = []
    for section in sections:
        title = _business_value(section.title)
        if not title:
            continue
        match = re.search(r"(?:分支评估|方案评估|候选评审)\s*(?:方案\s*)?(\d+)", title)
        if not match:
            match = re.search(r"方案\s*(\d+)\b", title)
        if match:
            values.append(f"方案 {match.group(1)}")
        elif re.search(r"主脚本|整体评审|整体评估", title):
            values.append("当前主脚本")
    return values


def _collect_sections(sections: list[_Section], labels: tuple[str, ...]) -> list[str]:
    values: list[str] = []
    for section in sections:
        if not _matches_label(section.title, labels):
            continue
        values.extend(_business_items(section.body))
    return values


def _collect_positive_sections(sections: list[_Section], labels: tuple[str, ...]) -> list[str]:
    values: list[str] = []
    for section in sections:
        if _is_problem_title(section.title) or not _matches_label(section.title, labels):
            continue
        values.extend(_business_items(section.body))
    return values


def _branch_evaluations(sections: list[_Section]) -> list[dict[str, Any]]:
    values: list[dict[str, Any]] = []
    current: dict[str, Any] | None = None

    for section in sections:
        subject = _branch_section_subject(section.title)
        if subject:
            current = {
                "subject": subject,
                "conclusion": _business_value(
                    _labeled_value(section.body, ("该支结论", "评审结论", "结论"))
                ),
                "achievements": [],
                "problems": [],
                "evidence": _business_value(_labeled_value(section.body, ("有据扎实度",))),
            }
            inline_achievement = _labeled_value(section.body, ("目标推进", "已达成", "主要优点"))
            inline_problem = _labeled_value(section.body, ("未通过项", "部分通过项", "主要问题", "问题", "存疑项"))
            if inline_achievement:
                current["achievements"].append(inline_achievement)
            if inline_problem and not _means_no_problem(inline_problem):
                current["problems"].append(inline_problem)
            values.append(current)
            continue

        if current is None:
            continue
        if _is_comparison_section(section.title):
            current = None
            continue

        evidence = _labeled_value(section.body, ("有据扎实度",))
        if evidence:
            current["evidence"] = _business_value(evidence)
        body = _without_labeled_lines(section.body, ("有据扎实度",))
        if _is_problem_title(section.title):
            current["problems"].extend(
                item for item in _business_items(body) if not _means_no_problem(item)
            )
        elif _is_positive_section_title(section.title):
            current["achievements"].extend(_business_items(body))

    for item in values:
        item["achievements"] = _unique(item.get("achievements") or [])
        item["problems"] = _unique(item.get("problems") or [])
    return values


def _comparison_rows(tables: list[list[dict[str, str]]]) -> list[dict[str, Any]]:
    values: list[dict[str, Any]] = []
    for rows in tables:
        if not rows:
            continue
        headers = list(rows[0])
        criterion_key = _first_header(headers, ("维度", "标准"))
        candidate_keys = [
            header for header in headers
            if re.search(r"(?:branch|方案|分支)\s*#?\s*\d+", header, re.I)
        ]
        basis_key = _first_header(headers, ("差异依据", "比较依据", "依据"))
        if not criterion_key or not candidate_keys:
            continue
        for row in rows:
            criterion = _business_value(row.get(criterion_key))
            if not criterion:
                continue
            values.append(
                {
                    "criterion": criterion,
                    "candidates": [
                        {
                            "subject": _business_value(header),
                            "value": _business_value(row.get(header)),
                        }
                        for header in candidate_keys
                        if _business_value(row.get(header))
                    ],
                    "basis": _business_value(row.get(basis_key)) if basis_key else None,
                }
            )
    return values


def _is_positive_section_title(value: str) -> bool:
    return not _is_problem_title(value) and _matches_label(
        value,
        ("已达成", "达成情况", "目标达成", "目标推进", "完成情况", "通过项", "达成项", "主要优点"),
    )


def _is_problem_title(value: str) -> bool:
    normalized = _clean_heading(value)
    return any(label in normalized for label in ("未通过", "部分通过项", "问题", "存疑", "遗留", "风险"))


def _is_comparison_section(value: str) -> bool:
    normalized = _clean_heading(value)
    return any(label in normalized for label in ("对比与建议", "跨方案对比", "方案对比", "综合建议", "执行统计"))


def _without_labeled_lines(text: str, labels: tuple[str, ...]) -> str:
    pattern = "|".join(re.escape(label) for label in labels)
    return "\n".join(
        line for line in text.splitlines()
        if not re.match(rf"^\s*(?:[-*+]\s*)?(?:\*\*)?(?:{pattern})(?:\*\*)?\s*[：:]", line)
    )


def _problem_sections(sections: list[_Section]) -> list[dict[str, str]]:
    values: list[dict[str, str]] = []
    for section in sections:
        if not _is_problem_title(section.title):
            continue
        body = _without_labeled_lines(section.body, ("有据扎实度",))
        for item in _business_items(body):
            problem: dict[str, str] = {"summary": item}
            severity = _severity(item)
            if severity:
                problem["severity"] = severity
            values.append(problem)
    return values


def _item_conclusions(sections: list[_Section]) -> list[dict[str, Any]]:
    values: list[dict[str, Any]] = []
    for section in sections:
        subject = _subject_from_title(section.title)
        if not subject:
            continue
        conclusion = _labeled_value(section.body, ("该支结论", "评审结论", "结论"))
        achievements = _inline_or_section_items(section.body, ("目标推进", "已达成", "主要优点"))
        problems = _inline_or_section_items(section.body, ("未通过项", "主要问题", "问题"))
        if conclusion or achievements or problems:
            values.append(
                {
                    "subject": subject,
                    "conclusion": _business_value(conclusion),
                    "achievements": _unique(achievements),
                    "problems": _unique(problems),
                }
            )
    return values


def _merge_table(
    rows: list[dict[str, str]],
    *,
    criteria: list[str],
    item_conclusions: list[dict[str, Any]],
    achievements: list[str],
    problems: list[dict[str, str]],
) -> None:
    if not rows:
        return
    headers = list(rows[0])
    subject_key = _first_header(headers, ("方案", "分支", "对象", "候选", "维度", "标准"))
    conclusion_key = _first_header(headers, ("结论", "结果", "判断"))
    achievement_key = _first_header(headers, ("已达成", "达成", "优点", "通过项"))
    problem_key = _first_header(headers, ("问题", "未通过", "遗留", "风险"))

    for row in rows:
        subject = _business_value(row.get(subject_key)) if subject_key else None
        conclusion = _business_value(row.get(conclusion_key)) if conclusion_key else None
        row_achievements = _business_items(row.get(achievement_key, "")) if achievement_key else []
        row_problems = _business_items(row.get(problem_key, "")) if problem_key else []

        if subject_key and any(term in subject_key for term in ("维度", "标准")) and subject:
            criteria.append(subject)
        if subject and conclusion:
            item_conclusions.append(
                {
                    "subject": subject,
                    "conclusion": conclusion,
                    "achievements": row_achievements,
                    "problems": row_problems,
                }
            )
        achievements.extend(row_achievements)
        for item in row_problems:
            problem: dict[str, str] = {"summary": item}
            severity = _severity(item)
            if severity:
                problem["severity"] = severity
            problems.append(problem)


def _section_value(sections: list[_Section], labels: tuple[str, ...]) -> str | None:
    for section in sections:
        if _matches_label(section.title, labels):
            items = _business_items(section.body)
            return "；".join(items) if items else None
    return None


def _labeled_value(text: str, labels: tuple[str, ...]) -> str | None:
    if not text:
        return None
    label_pattern = "|".join(re.escape(label) for label in labels)
    match = re.search(
        rf"(?im)^\s*(?:#{{1,6}}\s+)?(?:[-*+]\s*)?(?:\*\*)?(?:{label_pattern})(?:\*\*)?\s*[：:]\s*(.+?)\s*$",
        text,
    )
    return _business_value(match.group(1)) if match else None


def _inline_or_section_items(text: str, labels: tuple[str, ...]) -> list[str]:
    value = _labeled_value(text, labels)
    return [value] if value else []


def _business_items(text: str) -> list[str]:
    values: list[str] = []
    for paragraph in re.split(r"\n\s*\n", text or ""):
        lines = paragraph.splitlines()
        for line in lines:
            if re.match(r"^\s*\|.*\|\s*$", line):
                continue
            value = re.sub(r"^\s*(?:[-*+]\s+|\d+[.)、]\s*)", "", line).strip()
            value = _business_value(value)
            if value:
                values.append(value)
    return values


def _business_value(value: Any) -> str | None:
    text = _plain(str(value or ""))
    if not text:
        return None
    # Remove only named technical assignments. Never remove bare numbers or
    # generic "ID" tokens: business content such as "领域信息 31/32" must survive.
    text = re.sub(r"(?i)\bscript_build_id\s*[=:：]\s*\d+\s*[·|,，;；]?", "", text)
    text = re.sub(r"(?i)\bround(?:_index)?\s*[=:：]\s*\d+\s*[·|,，;；]?", "", text)
    text = re.sub(r"(?i)\b(?:paragraph|element|post)_id\s*[=:：]\s*[^\s:：,，;；)）]+\s*[·|,，;；]?", "", text)
    text = re.sub(r"(?i)\bbranch_id\s*[=:：]\s*(\d+)", r"方案 \1", text)
    text = re.sub(r"(?i)\bbranch\s*#?\s*(\d+)\b", r"方案 \1", text)
    text = re.sub(r"^(?:D|P)\d+\s*[：:|·-]?\s*(?=\D)", "", text, flags=re.I)
    text = re.sub(r"(?i)\b(?:action|target|path_type|dimension_pass)\s*[=:：]\s*[^,，;；]+", "", text)
    text = re.sub(r"^\s*[：:·|,，;；-]+\s*", "", text)
    text = re.sub(r"\s+([，。；！？、：,.])", r"\1", text)
    text = re.sub(r"\s{2,}", " ", text).strip(" ：:，,;；|-_")
    return text or None


def _plain(text: str) -> str:
    text = text.replace("\\n", "\n")
    text = re.sub(r"`([^`]*)`", r"\1", text)
    text = re.sub(r"[*#]", "", text)
    return text.strip()


def _clean_heading(value: str) -> str:
    return _plain(value).strip(" ：:")


def _matches_label(value: str, labels: tuple[str, ...]) -> bool:
    normalized = _clean_heading(value)
    return any(label in normalized for label in labels)


def _subject_from_title(title: str) -> str | None:
    cleaned = _business_value(title)
    if not cleaned:
        return None
    match = re.search(r"(?:分支评估|方案评估|候选评审)?\s*方案\s*(\d+)\b", cleaned)
    if match:
        return f"方案 {match.group(1)}"
    return None


def _branch_section_subject(title: str) -> str | None:
    cleaned = _business_value(title)
    if not cleaned or not re.search(r"分支评估|方案评估|候选评审", cleaned):
        return None
    return _subject_from_title(title)


def _first_header(headers: list[str], labels: tuple[str, ...]) -> str | None:
    return next((header for header in headers if any(label in header for label in labels)), None)


def _severity(value: str) -> str | None:
    match = re.search(r"(?:严重程度|级别|等级)\s*[：:]\s*(严重|高|中|低)", value)
    if match:
        return match.group(1)
    for label in ("严重", "高", "中", "低"):
        if re.match(rf"^(?:[【\[]{label}[】\]]|{label}\s*[：:])", value):
            return label
    return None


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


def _unique_dicts(values: list[dict[str, Any]]) -> list[dict[str, Any]]:
    seen: set[str] = set()
    result: list[dict[str, Any]] = []
    for value in values:
        marker = json.dumps(value, ensure_ascii=False, sort_keys=True)
        if marker not in seen:
            seen.add(marker)
            result.append(value)
    return result