replace_docx.py

#!/usr/bin/env python3
# -*- coding: utf-8 -*-

import argparse
import csv
import json
import re
import shutil
import sys
from datetime import datetime
from pathlib import Path

from docx import Document


PLACEHOLDER_LITERALS = [
    "XXX",
    "XX公司",
    "某某公司",
    "某地区某公司",
    "待补充",
    "待填写",
    "待完善",
    "填写",
    "【】",
    "[]",
    "[ ]",
    "__",
    "____",
    "202X年",
    "20XX年",
    "xxxx",
    "xxxxx",
    "TBD",
    "TODO",
]

PLACEHOLDER_REGEXES = [
    (re.compile(r"\{\{[^{}]*\}\}"), "{{...}}"),
    (re.compile(r"\$\{[^{}]*\}"), "${...}"),
]

UNSUPPORTED_RISKS = [
    "脚注、尾注、批注、文本框、文档属性未自动处理，需人工复核。",
    "复杂域代码、目录、水印、SmartArt、图片内文字未自动处理，需人工复核。",
]

CONTEXT_WINDOW = 18
MAX_SCAN_HITS_PER_TERM = 5


def build_char_map(paragraph):
    chars = []
    for run_idx, run in enumerate(paragraph.runs):
        for char_idx, ch in enumerate(run.text):
            chars.append({
                "char": ch,
                "run_idx": run_idx,
                "char_idx": char_idx,
            })
    return chars


def normalize_spaces(text):
    return text.replace("\r", " ").replace("\n", " ").strip()


def extract_context(text, start, end, window=CONTEXT_WINDOW):
    prefix_start = max(0, start - window)
    suffix_end = min(len(text), end + window)
    snippet = text[prefix_start:suffix_end]
    snippet = normalize_spaces(snippet)
    if prefix_start > 0:
        snippet = f"…{snippet}"
    if suffix_end < len(text):
        snippet = f"{snippet}…"
    return snippet


def build_preview(full_text, match):
    before = extract_context(full_text, match["start"], match["end"])
    after_text = (
        full_text[:match["start"]]
        + match["new"]
        + full_text[match["end"]:]
    )
    after = extract_context(
        after_text,
        match["start"],
        match["start"] + len(match["new"]),
    )
    return {
        "field": match["old"],
        "before": before,
        "after": after,
        "location": match["location"],
    }


def find_matches(full_text, replacements):
    matches = []
    index = 0

    while index < len(full_text):
        matched = None
        for order, (old, new) in enumerate(replacements):
            if full_text.startswith(old, index):
                matched = {
                    "start": index,
                    "end": index + len(old),
                    "old": old,
                    "new": new,
                    "order": order,
                }
                break

        if matched is None:
            index += 1
            continue

        matches.append(matched)
        index = matched["end"]

    return matches


def apply_single_match(paragraph, chars, match):
    affected = chars[match["start"]:match["end"]]
    first_run_idx = affected[0]["run_idx"]

    delete_map = {}
    for item in affected:
        delete_map.setdefault(item["run_idx"], set()).add(item["char_idx"])

    for run_idx, run in enumerate(paragraph.runs):
        if run_idx not in delete_map:
            continue

        new_text_parts = []
        inserted = False

        for char_idx, ch in enumerate(run.text):
            if char_idx in delete_map[run_idx]:
                if run_idx == first_run_idx and not inserted:
                    new_text_parts.append(match["new"])
                    inserted = True
            else:
                new_text_parts.append(ch)

        run.text = "".join(new_text_parts)


def replace_text_in_paragraph(paragraph, replacements, stats, previews, location):
    if not paragraph.runs:
        return

    chars = build_char_map(paragraph)
    if not chars:
        return

    full_text = "".join(item["char"] for item in chars)
    matches = find_matches(full_text, replacements)
    if not matches:
        return

    for match in matches:
        stats[match["old"]] = stats.get(match["old"], 0) + 1
        if len(previews) < 2:
            match["location"] = location
            previews.append(build_preview(full_text, match))

    for match in sorted(matches, key=lambda item: item["start"], reverse=True):
        apply_single_match(paragraph, chars, match)


def replace_in_table(table, replacements, stats, previews, scope):
    for row_idx, row in enumerate(table.rows, 1):
        for col_idx, cell in enumerate(row.cells, 1):
            for para_idx, paragraph in enumerate(cell.paragraphs, 1):
                location = f"{scope} / 行 {row_idx} / 列 {col_idx} / 段落 {para_idx}"
                replace_text_in_paragraph(
                    paragraph,
                    replacements,
                    stats,
                    previews,
                    location,
                )

            for nested_idx, nested_table in enumerate(cell.tables, 1):
                nested_scope = (
                    f"{scope} / 行 {row_idx} / 列 {col_idx} / 嵌套表格 {nested_idx}"
                )
                replace_in_table(
                    nested_table,
                    replacements,
                    stats,
                    previews,
                    nested_scope,
                )


def replace_in_container(container, replacements, stats, previews, scope):
    for para_idx, paragraph in enumerate(container.paragraphs, 1):
        location = f"{scope}段落 {para_idx}"
        replace_text_in_paragraph(paragraph, replacements, stats, previews, location)

    for table_idx, table in enumerate(container.tables, 1):
        replace_in_table(table, replacements, stats, previews, f"{scope}表格 {table_idx}")


def collect_container_records(container, scope, records):
    for para_idx, paragraph in enumerate(container.paragraphs, 1):
        text = normalize_spaces(paragraph.text)
        if text:
            records.append({
                "location": f"{scope}段落 {para_idx}",
                "text": text,
            })

    for table_idx, table in enumerate(container.tables, 1):
        collect_table_records(table, f"{scope}表格 {table_idx}", records)


def collect_table_records(table, scope, records):
    for row_idx, row in enumerate(table.rows, 1):
        for col_idx, cell in enumerate(row.cells, 1):
            for para_idx, paragraph in enumerate(cell.paragraphs, 1):
                text = normalize_spaces(paragraph.text)
                if text:
                    records.append({
                        "location": f"{scope} / 行 {row_idx} / 列 {col_idx} / 段落 {para_idx}",
                        "text": text,
                    })

            for nested_idx, nested_table in enumerate(cell.tables, 1):
                collect_table_records(
                    nested_table,
                    f"{scope} / 行 {row_idx} / 列 {col_idx} / 嵌套表格 {nested_idx}",
                    records,
                )


def collect_scan_records(doc):
    records = []
    collect_container_records(doc, "正文", records)

    for section_idx, section in enumerate(doc.sections, 1):
        collect_container_records(section.header, f"第 {section_idx} 节页眉", records)
        collect_container_records(section.footer, f"第 {section_idx} 节页脚", records)
        collect_container_records(
            section.first_page_header,
            f"第 {section_idx} 节首页页眉",
            records,
        )
        collect_container_records(
            section.first_page_footer,
            f"第 {section_idx} 节首页页脚",
            records,
        )
        collect_container_records(
            section.even_page_header,
            f"第 {section_idx} 节偶数页页眉",
            records,
        )
        collect_container_records(
            section.even_page_footer,
            f"第 {section_idx} 节偶数页页脚",
            records,
        )

    return records


def scan_records_for_terms(records, terms):
    findings = []
    seen = set()

    for term in terms:
        hit_count = 0
        for record in records:
            start = record["text"].find(term)
            while start != -1:
                if hit_count >= MAX_SCAN_HITS_PER_TERM:
                    break

                end = start + len(term)
                context = extract_context(record["text"], start, end)
                key = (term, record["location"], context)
                if key not in seen:
                    findings.append({
                        "term": term,
                        "location": record["location"],
                        "context": context,
                    })
                    seen.add(key)
                    hit_count += 1

                start = record["text"].find(term, start + len(term))

    return findings


def scan_records_for_placeholders(records):
    findings = []
    seen = set()

    for record in records:
        candidates = []

        for literal in PLACEHOLDER_LITERALS:
            start = record["text"].find(literal)
            hit_count = 0
            while start != -1:
                if hit_count >= MAX_SCAN_HITS_PER_TERM:
                    break

                candidates.append({
                    "term": literal,
                    "start": start,
                    "end": start + len(literal),
                })
                hit_count += 1
                start = record["text"].find(literal, start + len(literal))

        for pattern, label in PLACEHOLDER_REGEXES:
            hit_count = 0
            for match in pattern.finditer(record["text"]):
                if hit_count >= MAX_SCAN_HITS_PER_TERM:
                    break

                candidates.append({
                    "term": label,
                    "start": match.start(),
                    "end": match.end(),
                })
                hit_count += 1

        accepted_ranges = []
        for candidate in sorted(
            candidates,
            key=lambda item: (item["start"], -(item["end"] - item["start"])),
        ):
            overlapped = False
            for accepted in accepted_ranges:
                if not (
                    candidate["end"] <= accepted["start"]
                    or candidate["start"] >= accepted["end"]
                ):
                    overlapped = True
                    break

            if overlapped:
                continue

            accepted_ranges.append(candidate)
            context = extract_context(
                record["text"],
                candidate["start"],
                candidate["end"],
            )
            key = (candidate["term"], record["location"], context)
            if key in seen:
                continue

            findings.append({
                "term": candidate["term"],
                "location": record["location"],
                "context": context,
            })
            seen.add(key)

    return findings


def make_backup(input_path, backup_dir=None):
    timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
    backup_root = Path(backup_dir) if backup_dir else input_path.parent
    backup_root.mkdir(parents=True, exist_ok=True)
    backup_path = backup_root / f"{input_path.stem}.backup_{timestamp}{input_path.suffix}"
    shutil.copy2(input_path, backup_path)
    return backup_path


def replace_in_docx(input_path, output_path, replacements, create_backup=False, backup_dir=None):
    input_path = Path(input_path)
    output_path = Path(output_path)

    if not input_path.exists():
        raise FileNotFoundError(f"输入文件不存在：{input_path}")

    if input_path.suffix.lower() != ".docx":
        raise ValueError("仅支持 .docx 文件，不支持 .doc/.docm 或其他格式。")

    if not replacements:
        raise ValueError("替换规则为空。")

    output_path.parent.mkdir(parents=True, exist_ok=True)

    backup_path = None
    if create_backup:
        backup_path = make_backup(input_path, backup_dir=backup_dir)

    doc = Document(str(input_path))
    stats = {}
    previews = []

    replace_in_container(doc, replacements, stats, previews, "正文")

    for section_idx, section in enumerate(doc.sections, 1):
        replace_in_container(
            section.header,
            replacements,
            stats,
            previews,
            f"第 {section_idx} 节页眉",
        )
        replace_in_container(
            section.footer,
            replacements,
            stats,
            previews,
            f"第 {section_idx} 节页脚",
        )
        replace_in_container(
            section.first_page_header,
            replacements,
            stats,
            previews,
            f"第 {section_idx} 节首页页眉",
        )
        replace_in_container(
            section.first_page_footer,
            replacements,
            stats,
            previews,
            f"第 {section_idx} 节首页页脚",
        )
        replace_in_container(
            section.even_page_header,
            replacements,
            stats,
            previews,
            f"第 {section_idx} 节偶数页页眉",
        )
        replace_in_container(
            section.even_page_footer,
            replacements,
            stats,
            previews,
            f"第 {section_idx} 节偶数页页脚",
        )

    doc.save(str(output_path))
    records = collect_scan_records(doc)

    return {
        "input_path": input_path,
        "output_path": output_path,
        "backup_path": backup_path,
        "stats": stats,
        "previews": previews,
        "records": records,
    }


def parse_replacements(args):
    items = []
    for item in args:
        if "=" not in item:
            raise ValueError(f"参数格式错误：{item}，应为 旧词=新词")

        old, new = item.split("=", 1)
        items.append({
            "old": old,
            "new": new,
        })

    cleaned, warnings, errors = validate_replacements(items)
    if warnings:
        for warning in warnings:
            print(f"警告：{warning}", file=sys.stderr)
    if errors:
        raise ValueError("；".join(errors))
    return cleaned


def split_list_value(raw):
    if raw is None:
        return []

    text = str(raw).strip()
    if not text:
        return []

    parts = re.split(r"[，,；;、|]", text)
    return [item.strip() for item in parts if item.strip()]


def normalize_replacements(raw):
    if isinstance(raw, dict):
        return [{"old": key, "new": value} for key, value in raw.items()]

    if isinstance(raw, list):
        items = []
        for entry in raw:
            if isinstance(entry, dict) and "old" in entry and "new" in entry:
                items.append({"old": entry["old"], "new": entry["new"]})
            elif isinstance(entry, (list, tuple)) and len(entry) == 2:
                items.append({"old": entry[0], "new": entry[1]})
            else:
                raise ValueError(f"无法识别的 replacements 项：{entry}")
        return items

    raise ValueError("replacements 仅支持对象或数组格式。")


def validate_replacements(items):
    cleaned = []
    warnings = []
    errors = []
    seen = {}

    for index, item in enumerate(items, 1):
        old = "" if item.get("old") is None else str(item.get("old")).strip()
        new = "" if item.get("new") is None else str(item.get("new"))

        if not old:
            errors.append(f"第 {index} 个替换项的旧值为空。")
            continue

        if old in seen:
            if seen[old] != new:
                errors.append(f"字段 {old} 存在冲突替换：{seen[old]} / {new}")
            else:
                warnings.append(f"字段 {old} 存在重复映射，已按同值去重。")
            continue

        seen[old] = new
        cleaned.append((old, new))

        if new == "":
            warnings.append(f"字段 {old} 的新值为空，需人工确认是否允许删除。")

    old_values = [old for old, _ in cleaned]
    old_set = set(old_values)

    for old, new in cleaned:
        if new and new in old_set:
            warnings.append(f"检测到链式替换风险：{old} -> {new}")

    ordered_olds = sorted(old_values, key=len, reverse=True)
    for index, old in enumerate(ordered_olds):
        for other in ordered_olds[index + 1:]:
            if other in old:
                warnings.append(f"检测到包含关系：{old} / {other}")

    cleaned.sort(key=lambda item: len(item[0]), reverse=True)
    return cleaned, dedupe_messages(warnings), dedupe_messages(errors)


def dedupe_messages(messages):
    seen = set()
    unique = []
    for message in messages:
        if message in seen:
            continue
        seen.add(message)
        unique.append(message)
    return unique


def parse_arrow_line(line):
    for separator in ("→", "->", "=>"):
        if separator in line:
            old, new = line.split(separator, 1)
            return old.strip(), new.strip()
    raise ValueError(f"无法识别的替换行：{line}")


def parse_text_manifest(text):
    manifest = {"files": []}
    current = None

    for raw_line in text.splitlines():
        line = raw_line.strip()
        if not line:
            continue

        if line.startswith("原始文件：") or line.startswith("原始文件:"):
            manifest["template"] = line.split("：", 1)[-1].split(":", 1)[-1].strip()
            continue

        if line.startswith("输出目录：") or line.startswith("输出目录:"):
            manifest["output_dir"] = line.split("：", 1)[-1].split(":", 1)[-1].strip()
            continue

        if line.startswith("检查字段：") or line.startswith("检查字段:"):
            manifest["check_fields"] = split_list_value(
                line.split("：", 1)[-1].split(":", 1)[-1].strip()
            )
            continue

        if line.startswith("文件：") or line.startswith("文件:"):
            if current:
                manifest["files"].append(current)
            current = {
                "output_filename": line.split("：", 1)[-1].split(":", 1)[-1].strip(),
                "replacements": [],
            }
            continue

        if line.startswith("重点检查：") or line.startswith("重点检查:"):
            if current is None:
                raise ValueError("重点检查字段必须写在某个“文件：”块内。")
            current["check_fields"] = split_list_value(
                line.split("：", 1)[-1].split(":", 1)[-1].strip()
            )
            continue

        old, new = parse_arrow_line(line)
        if current is None:
            raise ValueError("替换映射必须写在某个“文件：”块内。")
        current["replacements"].append({"old": old, "new": new})

    if current:
        manifest["files"].append(current)

    return manifest


def parse_table_rows(rows):
    files = []
    for row in rows:
        output_filename = (row.get("output_filename") or "").strip()
        if not output_filename:
            continue

        item = {
            "output_filename": output_filename,
            "replacements": [],
        }

        for key, value in row.items():
            if key in {"output_filename", "check_fields", "focus_fields"}:
                continue

            if key.startswith("old_"):
                suffix = key[4:]
                new_key = f"new_{suffix}"
                old_value = (value or "").strip()
                new_value = (row.get(new_key) or "").strip()
                if old_value or new_value:
                    item["replacements"].append({
                        "old": old_value,
                        "new": new_value,
                    })

        focus_value = row.get("check_fields") or row.get("focus_fields")
        if focus_value:
            item["check_fields"] = split_list_value(focus_value)

        files.append(item)

    return {"files": files}


def parse_markdown_table(text):
    table_lines = []
    for line in text.splitlines():
        stripped = line.strip()
        if stripped.startswith("|") and stripped.endswith("|"):
            table_lines.append(stripped)

    if len(table_lines) < 2:
        raise ValueError("Markdown 表格格式不足。")

    header = [cell.strip() for cell in table_lines[0].strip("|").split("|")]
    rows = []

    for line in table_lines[2:]:
        values = [cell.strip() for cell in line.strip("|").split("|")]
        row = {}
        for index, key in enumerate(header):
            row[key] = values[index] if index < len(values) else ""
        rows.append(row)

    return parse_table_rows(rows)


def parse_csv_manifest(manifest_path, delimiter=","):
    with manifest_path.open("r", encoding="utf-8-sig", newline="") as handle:
        reader = csv.DictReader(handle, delimiter=delimiter)
        rows = list(reader)
    return parse_table_rows(rows)


def load_manifest(manifest_path):
    suffix = manifest_path.suffix.lower()

    if suffix == ".json":
        data = json.loads(manifest_path.read_text(encoding="utf-8"))
        if isinstance(data, list):
            return {"files": data}
        if isinstance(data, dict):
            return data
        raise ValueError("JSON 清单必须是对象或数组。")

    if suffix == ".csv":
        return parse_csv_manifest(manifest_path, delimiter=",")

    if suffix == ".tsv":
        return parse_csv_manifest(manifest_path, delimiter="\t")

    text = manifest_path.read_text(encoding="utf-8")
    if suffix == ".md" and "|" in text and "output_filename" in text:
        return parse_markdown_table(text)
    return parse_text_manifest(text)


def resolve_path(base_dir, raw_value):
    path = Path(raw_value)
    if path.is_absolute():
        return path
    return (base_dir / path).resolve()


def pick_value(manifest, *keys):
    for key in keys:
        value = manifest.get(key)
        if value:
            return value
    return None


def escape_cell(value):
    text = "" if value is None else str(value)
    text = text.replace("\r", " ").replace("\n", "<br>")
    return text.replace("|", r"\|")


def markdown_table(headers, rows):
    output = []
    output.append("| " + " | ".join(headers) + " |")
    output.append("|" + "|".join(["---"] * len(headers)) + "|")

    if not rows:
        empty = ["-" for _ in headers]
        output.append("| " + " | ".join(empty) + " |")
        return "\n".join(output)

    for row in rows:
        output.append(
            "| " + " | ".join(escape_cell(row.get(header, "")) for header in headers) + " |"
        )
    return "\n".join(output)


def build_report(report_data):
    lines = [
        "# 批量 Word 替换检查报告",
        "",
        "## 一、输入信息",
        "",
        f"- 原始文件：{report_data['template_path']}",
        f"- 输出目录：{report_data['output_dir']}",
        f"- 输出文件数量：{report_data['file_count']}",
        f"- 替换映射来源：{report_data['mapping_source']}",
        f"- 执行时间：{report_data['executed_at']}",
        "",
        "## 二、文件生成结果",
        "",
        markdown_table(
            ["序号", "输出文件名", "是否生成", "备注"],
            report_data["generation_rows"],
        ),
        "",
        "## 三、替换统计",
        "",
        markdown_table(
            ["输出文件名", "替换前", "替换后", "替换次数", "备注"],
            report_data["replacement_rows"],
        ),
        "",
        "## 四、旧信息残留检查",
        "",
        markdown_table(
            ["输出文件名", "残留字段", "出现位置", "说明"],
            report_data["residual_rows"],
        ),
        "",
        "## 五、通用占位符检查",
        "",
        markdown_table(
            ["输出文件名", "疑似占位符", "出现位置", "说明"],
            report_data["placeholder_rows"],
        ),
        "",
        "## 六、替换实例预览",
        "",
        markdown_table(
            ["输出文件名", "字段", "替换前上下文", "替换后上下文"],
            report_data["preview_rows"],
        ),
        "",
        "## 七、高风险文档提示",
        "",
        markdown_table(
            ["输出文件名", "风险项", "说明"],
            report_data["risk_rows"],
        ),
        "",
        "## 八、异常与待确认项",
        "",
        markdown_table(
            ["输出文件名", "问题", "建议处理"],
            report_data["issue_rows"],
        ),
        "",
    ]
    return "\n".join(lines)


def prepare_batch_item(item, global_check_fields):
    output_filename = (item.get("output_filename") or item.get("filename") or "").strip()
    if not output_filename:
        raise ValueError("缺少 output_filename。")

    replacements = normalize_replacements(item.get("replacements") or item.get("mapping") or {})
    cleaned, warnings, errors = validate_replacements(replacements)
    check_fields = split_list_value(item.get("check_fields") or item.get("focus_fields"))
    if not check_fields:
        check_fields = list(global_check_fields)

    return {
        "output_filename": output_filename,
        "replacements": cleaned,
        "check_fields": check_fields,
        "warnings": warnings,
        "errors": errors,
    }


def run_batch(args):
    manifest_path = Path(args.batch).resolve()
    manifest = load_manifest(manifest_path)

    template_value = args.template or pick_value(manifest, "template", "input_file", "source_file")
    output_dir_value = args.output_dir or pick_value(
        manifest,
        "output_dir",
        "output_directory",
    )

    if not template_value:
        raise ValueError("批量模式缺少原始文件，请通过 manifest 或 --template 提供。")
    if not output_dir_value:
        raise ValueError("批量模式缺少输出目录，请通过 manifest 或 --output-dir 提供。")

    template_path = resolve_path(manifest_path.parent, template_value)
    output_dir = resolve_path(manifest_path.parent, output_dir_value)
    output_dir.mkdir(parents=True, exist_ok=True)

    files = manifest.get("files") or manifest.get("items") or []
    if not files:
        raise ValueError("批量模式未读取到任何文件任务。")

    global_check_fields = split_list_value(manifest.get("check_fields"))
    report_data = {
        "template_path": template_path,
        "output_dir": output_dir,
        "file_count": len(files),
        "mapping_source": manifest_path,
        "executed_at": datetime.now().strftime("%Y-%m-%d %H:%M:%S"),
        "generation_rows": [],
        "replacement_rows": [],
        "residual_rows": [],
        "placeholder_rows": [],
        "preview_rows": [],
        "risk_rows": [],
        "issue_rows": [],
    }

    seen_outputs = set()

    for index, raw_item in enumerate(files, 1):
        output_filename = raw_item.get("output_filename") or raw_item.get("filename") or ""

        try:
            item = prepare_batch_item(raw_item, global_check_fields)
        except Exception as exc:
            report_data["generation_rows"].append({
                "序号": index,
                "输出文件名": output_filename or "-",
                "是否生成": "否",
                "备注": str(exc),
            })
            report_data["issue_rows"].append({
                "输出文件名": output_filename or "-",
                "问题": str(exc),
                "建议处理": "检查替换清单格式后重试。",
            })
            continue

        if item["output_filename"] in seen_outputs:
            report_data["generation_rows"].append({
                "序号": index,
                "输出文件名": item["output_filename"],
                "是否生成": "否",
                "备注": "输出文件名重复。",
            })
            report_data["issue_rows"].append({
                "输出文件名": item["output_filename"],
                "问题": "输出文件名重复。",
                "建议处理": "修改清单，确保每个输出文件名唯一。",
            })
            continue
        seen_outputs.add(item["output_filename"])

        for error in item["errors"]:
            report_data["issue_rows"].append({
                "输出文件名": item["output_filename"],
                "问题": error,
                "建议处理": "修正冲突后重新执行。",
            })

        if item["errors"]:
            report_data["generation_rows"].append({
                "序号": index,
                "输出文件名": item["output_filename"],
                "是否生成": "否",
                "备注": "存在冲突替换，已停止处理。",
            })
            continue

        output_path = output_dir / item["output_filename"]
        if output_path.resolve() == template_path.resolve():
            report_data["generation_rows"].append({
                "序号": index,
                "输出文件名": item["output_filename"],
                "是否生成": "否",
                "备注": "输出文件不能覆盖原始模板。",
            })
            report_data["issue_rows"].append({
                "输出文件名": item["output_filename"],
                "问题": "输出文件路径与原始模板相同。",
                "建议处理": "修改输出目录或输出文件名。",
            })
            continue

        overwrite_note = "覆盖已有文件。" if output_path.exists() else ""

        try:
            result = replace_in_docx(
                template_path,
                output_path,
                item["replacements"],
                create_backup=args.backup,
                backup_dir=args.backup_dir,
            )
        except Exception as exc:
            report_data["generation_rows"].append({
                "序号": index,
                "输出文件名": item["output_filename"],
                "是否生成": "否",
                "备注": str(exc),
            })
            report_data["issue_rows"].append({
                "输出文件名": item["output_filename"],
                "问题": str(exc),
                "建议处理": "检查模板文件、输出目录和替换清单。",
            })
            continue

        report_data["generation_rows"].append({
            "序号": index,
            "输出文件名": item["output_filename"],
            "是否生成": "是",
            "备注": overwrite_note or "生成成功。",
        })

        terms_to_check = []
        for old, _ in item["replacements"]:
            if old not in terms_to_check:
                terms_to_check.append(old)
        for field in item["check_fields"]:
            if field not in terms_to_check:
                terms_to_check.append(field)

        residuals = scan_records_for_terms(result["records"], terms_to_check)
        placeholders = scan_records_for_placeholders(result["records"])

        for old, new in item["replacements"]:
            count = result["stats"].get(old, 0)
            note_parts = []
            if count == 0:
                note_parts.append("未命中")
            report_data["replacement_rows"].append({
                "输出文件名": item["output_filename"],
                "替换前": old,
                "替换后": new,
                "替换次数": count,
                "备注": "；".join(note_parts) if note_parts else "",
            })

            if count == 0:
                report_data["issue_rows"].append({
                    "输出文件名": item["output_filename"],
                    "问题": f"字段 {old} 未命中。",
                    "建议处理": "确认模板文本是否完全一致，或检查是否存在未覆盖区域。",
                })

        for finding in residuals:
            report_data["residual_rows"].append({
                "输出文件名": item["output_filename"],
                "残留字段": finding["term"],
                "出现位置": finding["location"],
                "说明": finding["context"],
            })

        for finding in placeholders:
            report_data["placeholder_rows"].append({
                "输出文件名": item["output_filename"],
                "疑似占位符": finding["term"],
                "出现位置": finding["location"],
                "说明": finding["context"],
            })

        for preview in result["previews"]:
            report_data["preview_rows"].append({
                "输出文件名": item["output_filename"],
                "字段": preview["field"],
                "替换前上下文": preview["before"],
                "替换后上下文": preview["after"],
            })

        for warning in item["warnings"]:
            report_data["risk_rows"].append({
                "输出文件名": item["output_filename"],
                "风险项": "替换规则风险",
                "说明": warning,
            })

        for risk in UNSUPPORTED_RISKS:
            report_data["risk_rows"].append({
                "输出文件名": item["output_filename"],
                "风险项": "工具覆盖范围有限",
                "说明": risk,
            })

        if result["backup_path"]:
            report_data["risk_rows"].append({
                "输出文件名": item["output_filename"],
                "风险项": "已生成备份",
                "说明": f"备份文件：{result['backup_path']}",
            })

        if residuals:
            report_data["issue_rows"].append({
                "输出文件名": item["output_filename"],
                "问题": "检测到旧字段残留。",
                "建议处理": "根据残留位置人工复核，必要时补充替换规则或改用更底层处理方式。",
            })

        if placeholders:
            report_data["issue_rows"].append({
                "输出文件名": item["output_filename"],
                "问题": "检测到通用占位符或模板残留。",
                "建议处理": "人工确认是否需要补充替换，或保留为待填写项。",
            })

    report_path = output_dir / "check.md"
    report_path.write_text(build_report(report_data), encoding="utf-8")

    print("批量处理完成")
    print(f"模板文件：{template_path}")
    print(f"输出目录：{output_dir}")
    print(f"检查报告：{report_path}")


def run_single(args):
    if not args.input_file or not args.output_file or not args.replacements:
        raise ValueError(
            "单文件模式用法：python replace_docx.py 输入.docx 输出.docx 旧词=新词 [旧词=新词 ...]"
        )

    replacements = parse_replacements(args.replacements)
    result = replace_in_docx(
        args.input_file,
        args.output_file,
        replacements,
        create_backup=not args.no_backup,
        backup_dir=args.backup_dir,
    )

    print("处理完成")
    print(f"输入文件：{result['input_path']}")
    print(f"输出文件：{result['output_path']}")
    if result["backup_path"]:
        print(f"备份文件：{result['backup_path']}")

    print("\n替换统计：")
    for old, new in replacements:
        count = result["stats"].get(old, 0)
        print(f"- {old} -> {new}：{count} 处")

    missed = [old for old, _ in replacements if result["stats"].get(old, 0) == 0]
    if missed:
        print("\n未命中项：")
        for item in missed:
            print(f"- {item}")

    print("\n风险提示：")
    for risk in UNSUPPORTED_RISKS:
        print(f"- {risk}")


def build_parser():
    parser = argparse.ArgumentParser(
        description="批量或单文件替换 docx 模板中的变量。",
    )
    parser.add_argument("--batch", help="批量替换清单，支持 .json/.txt/.md/.csv/.tsv")
    parser.add_argument("--template", help="批量模式模板文件，可覆盖 manifest 中的原始文件")
    parser.add_argument("--output-dir", help="批量模式输出目录，可覆盖 manifest 中的输出目录")
    parser.add_argument("--backup", action="store_true", help="批量模式也生成模板备份")
    parser.add_argument("--no-backup", action="store_true", help="单文件模式不生成备份")
    parser.add_argument("--backup-dir", help="备份文件输出目录")
    parser.add_argument("input_file", nargs="?")
    parser.add_argument("output_file", nargs="?")
    parser.add_argument("replacements", nargs="*")
    return parser


def main():
    parser = build_parser()
    args = parser.parse_args()

    try:
        if args.batch:
            run_batch(args)
        else:
            run_single(args)
    except Exception as exc:
        print(f"错误：{exc}")
        sys.exit(2)


if __name__ == "__main__":
    main()