proposa-ai/src/step5_reviewing/policies.py

"""Step 5 的确定性审核策略。

该模块只做不依赖 LLM 的结构/文本检查与修复，供 reviewer 和 fixer 共同使用：
  - 标题层级与同级序号
  - 标题是否携带分值
  - 正文首行缩进提示
  - 无分包时的包号清理
  - 重复内容清理
  - 模板基块与评分/废标补充块的对应关系
  - 废标项是否按 Step3 绑定节点填充
"""

from __future__ import annotations

import re
from typing import Dict, Iterable, List, Optional, Tuple

from models import BidOutline, Chapter, ReviewIssue

try:
    from step3_outlining.scoring_structure import _next_prefix
except Exception:  # pragma: no cover - 仅用于极端导入环境兜底
    def _next_prefix(level: int, index: int) -> str:
        return f"{index}. "


FULLWIDTH_SPACE = "\u3000"

_PACKAGE_KEYWORDS = ("包号", "包件")
_PACKAGE_LINE_RE = re.compile(r"(包号|包件)")
_PROTECTED_SIGNATURE_RE = re.compile(
    r"^\s*(投标人授权代表签字|投标人（公章）|投标人\(公章\)|日期|法定代表人|"
    r"授权代表|签署人)[：:（(]"
)
_PLACEHOLDER_RE = re.compile(r"%{1,3}\s*([^%]+?)\s*%{1,3}")
_SCORE_SUFFIX_RE = re.compile(
    r"[（(]\s*\d+(?:\.\d+)?\s*分\s*[）)]\s*$"
)

_PREFIX_PATTERNS: Dict[int, str] = {
    1: r"^第\s*[一二三四五六七八九十百零\d]+\s*章\s*",
    2: r"^[一二三四五六七八九十百]+\s*[、，]\s*",
    3: r"^[（(][一二三四五六七八九十百\d]+[）)]\s*",
    4: r"^(?:\d+[.、．]|[（(]\d+[）)])\s*",
}

_CHINESE_DIGITS = "零一二三四五六七八九"


def _chinese_to_int(raw: str) -> int:
    """把 1-99 的中文数字或阿拉伯数字转为 int；无法识别时返回 0。"""
    text = (raw or "").strip()
    if text.isdigit():
        return int(text)
    if not text:
        return 0
    if text == "十":
        return 10
    if text.startswith("十") and len(text) == 2:
        return 10 + _chinese_to_int(text[1])
    if text.endswith("十") and len(text) == 2:
        return _chinese_to_int(text[0]) * 10
    if "十" in text:
        tens, ones = text.split("十", 1)
        return _chinese_to_int(tens) * 10 + _chinese_to_int(ones)
    if text in _CHINESE_DIGITS:
        return _CHINESE_DIGITS.index(text)
    return 0


def split_heading_number(text: str) -> Optional[Tuple[int, str]]:
    """解析标题前缀，返回 (level, 序号原文)；不是已知标题时返回 None。"""
    value = (text or "").strip()
    match = re.match(r"^第\s*([一二三四五六七八九十百零\d]+)\s*章", value)
    if match:
        return 1, match.group(1)
    match = re.match(r"^([一二三四五六七八九十百]+)[、，]", value)
    if match:
        return 2, match.group(1)
    match = re.match(r"^[（(]([一二三四五六七八九十百\d]+)[）)]", value)
    if match:
        return 3, match.group(1)
    match = re.match(r"^(\d+)[.、．]", value)
    if match:
        return 4, match.group(1)
    match = re.match(r"^[（(](\d+)[）)]", value)
    if match:
        return 4, match.group(1)
    return None


def is_heading_line(line: str) -> bool:
    """判断一行是否属于标题/编号行（不参与正文首行缩进检查）。"""
    value = (line or "").strip()
    if not value:
        return False
    if split_heading_number(value) is not None:
        return True
    return bool(re.match(r"^(?:【|#{1,6}\s)", value))


def is_list_or_table_line(line: str) -> bool:
    """判断一行是否为列表、表格或其他不应缩进的辅助行。"""
    value = (line or "").strip()
    if not value:
        return True
    if value.startswith("|"):
        return True
    if value.startswith(("-", "*", "·", "●", "◆", "■", "▍", "—")):
        return True
    if re.match(r"^\d+[.、．]\s", value) or re.match(r"^[（(]\d+[）)]\s", value):
        return True
    return False


def strip_existing_prefix(text: str, level: int) -> str:
    """去掉标题开头的既有编号前缀，保留标题本体。"""
    pattern = _PREFIX_PATTERNS.get(level)
    if not pattern:
        return (text or "").strip()
    return re.sub(pattern, "", text or "", count=1).strip()


def strip_score_from_title(title: str) -> str:
    """去掉新增评分标题末尾携带的分值，例如“（8分）”。"""
    cleaned = _SCORE_SUFFIX_RE.sub("", (title or "").strip())
    return cleaned.strip()


def iter_body_lines(chapter: Chapter) -> Iterable[str]:
    """遍历章节正文中的普通正文行（不含标题、列表、表格与空白）。"""
    def _emit(content: str):
        for raw in (content or "").splitlines():
            line = raw.strip()
            if not line:
                continue
            if is_heading_line(line) or is_list_or_table_line(line):
                continue
            if _PLACEHOLDER_RE.search(line):
                continue
            yield line

    def _walk(node: Chapter):
        yield node
        for child in node.children or []:
            yield from _walk(child)

    for node in _walk(chapter):
        yield from _emit(node.generated_content or "")
        yield from _emit(node.supplement_content or "")


def _direct_bindings(chapter: Chapter) -> bool:
    return bool(
        getattr(chapter, "direct_scoring_bindings", [])
        or getattr(chapter, "direct_scoring_criteria", [])
        or getattr(chapter, "direct_rejection_bindings", [])
    )


def find_heading_score_issues(outline: BidOutline) -> List[ReviewIssue]:
    """发现新增评分/废标标题仍携带分值的问题。"""
    issues: List[ReviewIssue] = []
    for node in outline.flatten():
        if not _direct_bindings(node):
            continue
        cleaned = strip_score_from_title(node.title)
        if cleaned != (node.title or "").strip():
            issues.append(ReviewIssue(
                chapter_id=node.id,
                severity="error",
                issue_type="format_heading",
                description=f"标题 {node.id} {node.title} 携带分值，不应附加分值",
                suggestion="去掉标题末尾分值，分值保留在评分要求和索引中",
            ))
    return issues


def renumber_sibling_children(parent: Chapter) -> int:
    """按顺序重排直接子标题的同一级序号，返回修改数量。"""
    children = list(parent.children or [])
    if not children:
        return 0
    level = parent.level + 1
    if level not in (2, 3, 4):
        return 0
    changed = 0
    position = 1
    for child in children:
        if child.level != level:
            continue
        if split_heading_number(child.title) is None:
            # 未带标准编号的标题可能是模板原标题，保持不动。
            continue
        prefix = _next_prefix(level, position)
        body = strip_existing_prefix(child.title, level)
        new_title = prefix + body
        if new_title != (child.title or "").strip():
            child.title = new_title
            changed += 1
        position += 1
    return changed


def find_sibling_numbering_issues(outline: BidOutline) -> List[ReviewIssue]:
    """发现同级子标题序号重复、跳号或乱序的问题。"""
    issues: List[ReviewIssue] = []
    for parent in outline.flatten():
        children = [
            child for child in (parent.children or [])
            if child.level == parent.level + 1 and child.level in (2, 3, 4)
        ]
        if not children:
            continue
        problems = []
        seen: Dict[int, str] = {}
        position = 1
        for child in children:
            parsed = split_heading_number(child.title)
            if parsed is None:
                continue
            number = _chinese_to_int(parsed[1])
            if number == 0 or number != position:
                problems.append(f"{child.id} {child.title}")
            if number in seen:
                problems.append(f"{child.id} {child.title} 与 {seen[number]} 重复序号")
            seen[number] = child.id
            position += 1
        if problems:
            issues.append(ReviewIssue(
                chapter_id=parent.id,
                severity="error",
                issue_type="format_numbering",
                description=(
                    f"父标题 {parent.id} {parent.title} 下同级子标题序号异常: "
                    + "；".join(problems[:6])
                ),
                suggestion="按顺序重新生成同级中文序号",
            ))
    return issues


def find_body_indent_issues(outline: BidOutline) -> List[ReviewIssue]:
    """统计正文未以两个全角空格开头的段落，供 Step6 统一格式化。"""
    issues: List[ReviewIssue] = []
    for chapter in outline.chapters:
        missing = [
            line for line in iter_body_lines(chapter)
            if not line.startswith(FULLWIDTH_SPACE * 2)
        ]
        if missing:
            issues.append(ReviewIssue(
                chapter_id=chapter.id,
                severity="warning",
                issue_type="format_body",
                description=f"第{chapter.id}章有 {len(missing)} 段正文未以两个全角空格开头",
                suggestion="正文首行缩进将由 Step6 统一应用，Step5 仅记录不改写文本",
            ))
    return issues


def remove_package_lines(content: str) -> Tuple[str, int]:
    """删除无分包项目中与包号/包件相关的整行信息。"""
    if not content:
        return content, 0
    kept = []
    removed = 0
    for line in content.splitlines():
        if _PACKAGE_LINE_RE.search(line):
            removed += 1
            continue
        kept.append(line)
    return "\n".join(kept), removed


def find_package_info_issues(
    outline: BidOutline,
    has_packages: bool,
) -> List[ReviewIssue]:
    """无分包时发现仍残留包号/包件信息的章节。"""
    if has_packages:
        return []
    issues: List[ReviewIssue] = []
    for chapter in outline.flatten():
        content = chapter.generated_content or ""
        if not _PACKAGE_LINE_RE.search(content):
            continue
        count = sum(1 for line in content.splitlines() if _PACKAGE_LINE_RE.search(line))
        issues.append(ReviewIssue(
            chapter_id=chapter.id,
            severity="error",
            issue_type="package_info",
            description=f"项目无分包，但章节 {chapter.id} 仍残留 {count} 行包号/包件信息",
            suggestion="删除章节中的包号/包件字段，避免多余信息",
        ))
    return issues


def analysis_has_package_info(analysis) -> bool:
    """依据 Step2 LLM 提取结果判断项目是否真的存在包号/包件信息。

    Step2 的 _extract_packages 会严格区分“实际包号值”和“模板占位/说明性文字”；
    这里以 packages 为主，并兼容 project_fields 中出现包号/包件键的情况。
    """
    if getattr(analysis, "packages", None):
        return True
    fields = getattr(analysis, "project_fields", {}) or {}
    return any(
        key in fields
        for key in ("包号", "包件", "包件号", "包件名称", "包名")
    )


def _normalize_line(line: str) -> str:
    return re.sub(r"[\s\u3000]+", "", line or "")


def find_duplicate_info_issues(outline: BidOutline) -> List[ReviewIssue]:
    """发现章节正文中重复出现的非空正文行。"""
    issues: List[ReviewIssue] = []
    for chapter in outline.chapters:
        seen: Dict[str, str] = {}
        duplicates: List[str] = []
        for line in iter_body_lines(chapter):
            if _PROTECTED_SIGNATURE_RE.match(line.strip()):
                continue
            key = _normalize_line(line)
            if len(key) < 12:
                continue
            if key in seen:
                if key not in duplicates:
                    duplicates.append(seen[key])
            else:
                seen[key] = line
        if duplicates:
            issues.append(ReviewIssue(
                chapter_id=chapter.id,
                severity="error",
                issue_type="duplicate_content",
                description=f"第{chapter.id}章存在 {len(duplicates)} 段重复正文",
                suggestion="删除重复段落，保留首次出现的内容",
            ))
    return issues


def dedupe_chapter_content(content: str) -> Tuple[str, int]:
    """按最小标题节删除正文中重复的非空行，保留首次出现顺序。"""
    if not content:
        return content, 0
    seen = set()
    kept = []
    removed = 0
    for line in content.splitlines():
        if _PROTECTED_SIGNATURE_RE.match(line.strip()):
            kept.append(line)
            continue
        if is_heading_line(line):
            seen = set()
            kept.append(line)
            continue
        key = _normalize_line(line)
        if key and len(key) >= 12 and key in seen:
            removed += 1
            continue
        if key:
            seen.add(key)
        kept.append(line)
    return "\n".join(kept), removed


def find_template_integrity_issues(outline: BidOutline) -> List[ReviewIssue]:
    """发现模板基块与评分/废标补充块的对应关系不一致。"""
    issues: List[ReviewIssue] = []
    entries = list(getattr(outline, "evaluation_index_entries", []) or [])
    entries_by_heading: Dict[str, set] = {}
    for entry in entries:
        heading_id = str(entry.get("final_heading_id", "") or "")
        if heading_id:
            entries_by_heading.setdefault(heading_id, set()).add(
                str(entry.get("entry_type", ""))
            )

    for node in outline.flatten():
        for block in (node.content_blocks or []):
            block_type = str(block.get("block_type", ""))
            if block_type not in ("scoring_supplement", "rejection_supplement"):
                continue
            kind = "scoring" if block_type == "scoring_supplement" else "rejection"
            target_id = str(block.get("target_heading_id", "") or "") or str(node.id)
            expected = entries_by_heading.get(target_id, set())
            if kind not in expected:
                issues.append(ReviewIssue(
                    chapter_id=node.id,
                    severity="error",
                    issue_type="duplicate_content",
                    description=(
                        f"节点 {node.id} {node.title} 存在未被 Step3 大纲授权的"
                        f"{'评分' if kind == 'scoring' else '废标'}补充块"
                    ),
                    suggestion="删除多余补充块，仅保留 Step3 绑定的授权内容",
                ))
    return issues


_REJECTION_CONSEQUENCE_SUFFIXES = (
    r"(?:，|；|。)?(?:其)?投标(?:将)?(?:被)?(?:认定为|作为|按)?无效(?:投标)?",
    r"(?:，|；|。)?(?:其)?投标无效",
    r"(?:，|；|。)?(?:招标人|采购人|评标委员会)?(?:将|可|有权)?(?:予以)?拒绝"
    r"(?:其)?(?:投标|接收|参加本次采购活动)?",
    r"(?:，|；|。)?不能通过(?:资格性|符合性)?审查",
)


def compact_rejection_requirement(text: str) -> str:
    """保留废标条件本体，去掉每条末尾重复出现的处理后果。"""
    value = re.sub(r"\s+", "", str(text or "")).strip("。；; ")
    previous = None
    while value and value != previous:
        previous = value
        for suffix in _REJECTION_CONSEQUENCE_SUFFIXES:
            value = re.sub(f"{suffix}[。；; ]*$", "", value).strip("，。；; ")
    return value


def find_rejection_filling_issues(
    outline: BidOutline,
    rejection_entries: List[dict],
) -> List[ReviewIssue]:
    """检查废标项是否全部填充到 Step3 绑定的承诺节点。"""
    if not rejection_entries:
        return []
    nodes = [
        node for node in outline.flatten()
        if getattr(node, "direct_rejection_bindings", [])
    ]
    # Step3 把所有废标项唯一绑定到第一章要求承诺函；未找到时给出可定位问题。
    if not nodes:
        return [ReviewIssue(
            chapter_id="*",
            severity="error",
            issue_type="rejection_filling",
            description="Step3 大纲存在废标项，但未找到直接绑定节点",
            suggestion="在第一章要求承诺函绑定废标项",
        )]
    target = nodes[0]
    content = "\n".join([
        target.supplement_content or "",
        target.generated_content or "",
    ])
    missing = []
    for entry in rejection_entries:
        requirement = compact_rejection_requirement(
            str(entry.get("requirement", ""))
        )
        if requirement and requirement not in content:
            missing.append(str(entry.get("source_id", "")))
    if missing:
        return [ReviewIssue(
            chapter_id=target.id,
            severity="error",
            issue_type="rejection_filling",
            description=(
                f"要求承诺函未完整填充 {len(missing)} 条废标项: "
                + ", ".join(missing[:8])
            ),
            suggestion="把缺失废标项写入该承诺节点，不得散落到其他章节",
        )]
    return []


__all__ = [
    "split_heading_number",
    "strip_existing_prefix",
    "strip_score_from_title",
    "is_heading_line",
    "is_list_or_table_line",
    "iter_body_lines",
    "renumber_sibling_children",
    "find_heading_score_issues",
    "find_sibling_numbering_issues",
    "find_body_indent_issues",
    "remove_package_lines",
    "find_package_info_issues",
    "dedupe_chapter_content",
    "find_duplicate_info_issues",
    "find_template_integrity_issues",
    "compact_rejection_requirement",
    "find_rejection_filling_issues",
]