AiDOPWarehouse/doc/_delivery_docx_polish.py

# -*- coding: utf-8 -*-
"""正式交付 Word：黑色正文、去重版本、静态目录、去空白页与免责/性能散落。"""
from __future__ import annotations

import re
import shutil
import sys
import tempfile
from pathlib import Path

HERE = Path(__file__).resolve().parent
if str(HERE) not in sys.path:
    sys.path.insert(0, str(HERE))

from lxml import etree
from docx import Document
from docx.enum.text import WD_BREAK
from docx.oxml import OxmlElement
from docx.oxml.ns import qn
from docx.shared import Pt, RGBColor
from docx.table import Table
from docx.text.paragraph import Paragraph

from _add_req_func_codes import set_paragraph_text
from _adjust_front_matter import remove_paragraph

BLACK = RGBColor(0, 0, 0)
HEADER_FILL = "D6E3F0"
DARK_FILLS = {"1F3864", "1F4E79", "4472C4", "2F5496", "1F4E79"}

DISCLAIMER_RE = re.compile(
    r"不包含移动端|不含移动端|不覆盖移动端|不含移动端专册|"
    r"本手册不包含手机|手机 App|手机App|"
    r"本总纲不覆盖|"
    r"不讲接口、表名和部署|遇到红色提示"
)
PERF_RE = re.compile(
    r"合同性能|页面响应遵循|不按秒级|秒级页面|秒级响应|"
    r"一般查询不超过|复杂看板不超过|"
    r"时延≤|传输\+清洗≤|1000\s*万条|"
    r"1000人同时在线|400人并发|"
    r"全年可用性|接口调用成功率|"
    r"性能：AI/ChatBI|性能目标（|不计入本系统性能|"
    r"部署与非功能|^6 非功能$|^非功能$|^性能要求"
)
DROP_HEAD_RE = re.compile(r"^(?:\d+(?:\.\d+)*\s*)?(部署与非功能|非功能|性能要求|性能与并发)\s*$")
REF_HEAD_RE = re.compile(
    r"^(?:\d+(?:\.\d+)*\s+|第[一二三四五六七八九十\d]+[章节篇]\s+|"
    r"[（(]?[一二三四五六七八九十]+[、.．)]\s*)?参考资料\s*$"
)
HEAD_NAME_RE = re.compile(r"^(?:Heading|标题)\s*(\d+)$")
HEAD_NUM_RE = re.compile(r"^(\d+(?:\.\d+)*)([.、．]\s*|\s+)(.+)$")
FRONT_TITLES = {"版本记录", "更改记录", "目录"}

FORMAL_KEYS = {
    "S0": ["V0.8", "V1.2"],
    "S1": ["V0.4", "V2.2", "V1.2"],
    "S2": ["V0.4", "V1.2"],
    "S3": ["V0.4", "V1.2"],
    "S4": ["V0.4", "V1.2"],
    "S5": ["V1.2", "V0.3"],
    "S6": ["V1.2", "V0.3"],
    "S7": ["V1.2", "V0.3"],
    "S8": ["V1.2", "V0.3"],
    "00-系统总纲": ["V0.4"],
    "S9": ["V0.4"],
    "运营诊断": ["V0.4"],
    "运营改善": ["V0.4"],
    "ChatBI": ["V0.4"],
    "系统集成": ["V0.4"],
}


def latest_formal_files(base: Path) -> list[Path]:
    out = []
    for folder, keys in FORMAL_KEYS.items():
        d = base / folder
        if not d.exists():
            continue
        for p in d.glob("*.docx"):
            if p.name.startswith("~$"):
                continue
            if any(k in p.name for k in keys):
                out.append(p)
    return out


def open_via_temp(path: Path) -> Document:
    try:
        return Document(str(path))
    except Exception:
        tmp = Path(tempfile.gettempdir()) / "aidop_polish.docx"
        shutil.copy2(path, tmp)
        return Document(str(tmp))


def save_via_temp(doc: Document, dest: Path) -> None:
    tmp = Path(tempfile.gettempdir()) / "aidop_unify_ab"
    tmp.mkdir(parents=True, exist_ok=True)
    out = tmp / f"polish_{dest.stem[:40]}.docx"
    doc.save(str(out))
    shutil.copy2(out, dest)


def _force_run_black(run) -> None:
    rPr = run._element.get_or_add_rPr()
    color_el = rPr.find(qn("w:color"))
    if color_el is not None:
        rPr.remove(color_el)
    highlight = rPr.find(qn("w:highlight"))
    if highlight is not None:
        rPr.remove(highlight)
    run.font.color.rgb = BLACK


def _iter_all_paragraphs(doc: Document):
    for p in doc.paragraphs:
        yield p
    for tbl in doc.tables:
        for row in tbl.rows:
            for cell in row.cells:
                for p in cell.paragraphs:
                    yield p
    for section in doc.sections:
        for part in (section.header, section.footer):
            for p in part.paragraphs:
                yield p
            for tbl in part.tables:
                for row in tbl.rows:
                    for cell in row.cells:
                        for p in cell.paragraphs:
                            yield p
    body = doc.element.body
    for txbx in body.iter(qn("w:txbxContent")):
        for p_el in txbx.iter(qn("w:p")):
            yield Paragraph(p_el, doc)


def force_all_text_black(doc: Document) -> int:
    n = 0
    for p in _iter_all_paragraphs(doc):
        for run in p.runs:
            _force_run_black(run)
            n += 1
    for tbl in doc.tables:
        for row in tbl.rows:
            for cell in row.cells:
                tc_pr = cell._tc.get_or_add_tcPr()
                shd = tc_pr.find(qn("w:shd"))
                if shd is not None:
                    fill = (shd.get(qn("w:fill")) or "").upper()
                    if fill in DARK_FILLS:
                        shd.set(qn("w:fill"), HEADER_FILL)
    return n


def heading_level(p: Paragraph) -> int | None:
    name = (p.style.name if p.style else "") or ""
    m = HEAD_NAME_RE.match(name.strip())
    if m:
        return int(m.group(1))
    pPr = p._element.find(qn("w:pPr"))
    if pPr is not None:
        ol = pPr.find(qn("w:outlineLvl"))
        if ol is not None:
            val = ol.get(qn("w:val"))
            if val is not None and val.isdigit():
                return int(val) + 1
    return None


def _xml(el) -> str:
    if hasattr(el, "xml"):
        return el.xml
    return etree.tostring(el, encoding="unicode")


def _is_page_break_only(p: Paragraph) -> bool:
    xml = _xml(p._element)
    has_br = 'w:type="page"' in xml or "w:type='page'" in xml
    return has_br and not p.text.strip()


def collect_headings(doc: Document) -> list[tuple[int, str]]:
    seen_toc = False
    items = []
    for p in doc.paragraphs:
        t = p.text.strip()
        if t == "目录":
            seen_toc = True
            continue
        if not seen_toc:
            continue
        if t in {"版本记录", "更改记录", "目录"}:
            continue
        lvl = heading_level(p)
        if lvl and lvl <= 3 and t:
            items.append((lvl, t))
    return items


def _remove_toc_fields(doc: Document) -> None:
    body = doc.element.body
    for sdt in list(body.iter(qn("w:sdt"))):
        xml = _xml(sdt)
        if "Table of Contents" in xml or "TOC" in xml or "目录" in xml:
            parent = sdt.getparent()
            if parent is not None:
                parent.remove(sdt)
    for p in list(doc.paragraphs):
        xml = _xml(p._element)
        if "w:instrText" in xml and "TOC" in xml and not heading_level(p):
            if p.text.strip() in {"", "目录"}:
                if p.text.strip() == "目录":
                    continue
                remove_paragraph(p)


def parse_heading_num(text: str) -> tuple[tuple[int, ...], str, str] | None:
    m = HEAD_NUM_RE.match(text.strip())
    if not m:
        return None
    parts = tuple(int(x) for x in m.group(1).split("."))
    return parts, m.group(2), m.group(3)


def renumber_heading_gaps(doc: Document) -> int:
    seen_toc = False
    parsed: list[tuple[Paragraph, tuple[int, ...], str, str]] = []
    for p in doc.paragraphs:
        t = p.text.strip()
        if t == "目录":
            seen_toc = True
            continue
        if not seen_toc or t in FRONT_TITLES:
            continue
        if not heading_level(p):
            continue
        got = parse_heading_num(t)
        if not got:
            continue
        num, sep, title = got
        parsed.append((p, num, sep, title))
    if not parsed:
        return 0
    from collections import defaultdict

    by_parent: dict[tuple[int, ...], list[int]] = defaultdict(list)
    for _p, num, _sep, _title in parsed:
        parent, last = num[:-1], num[-1]
        if last not in by_parent[parent]:
            by_parent[parent].append(last)
    child_map: dict[tuple[tuple[int, ...], int], int] = {}
    for parent, lasts in by_parent.items():
        ordered = sorted(set(lasts))
        for i, old in enumerate(ordered, 1):
            if old != i:
                child_map[(parent, old)] = i
    if not child_map:
        return 0

    def remap(num: tuple[int, ...]) -> tuple[int, ...]:
        out = []
        for i, part in enumerate(num):
            out.append(child_map.get((num[:i], part), part))
        return tuple(out)

    n = 0
    for p, num, sep, title in parsed:
        new = remap(num)
        if new == num:
            continue
        new_text = ".".join(str(x) for x in new) + sep + title
        set_paragraph_text(p, new_text)
        n += 1
    return n


def demote_front_titles(doc: Document) -> None:
    for p in doc.paragraphs:
        if p.text.strip() in FRONT_TITLES and heading_level(p):
            try:
                p.style = doc.styles["Normal"]
            except (KeyError, ValueError):
                pass


def _clear_after_toc_title(doc: Document, toc: Paragraph) -> None:
    el = toc._element.getnext()
    while el is not None:
        nxt = el.getnext()
        tag = el.tag.split("}")[-1]
        if tag == "p":
            p = Paragraph(el, doc)
            t = p.text.strip()
            lvl = heading_level(p)
            if lvl and t and t not in FRONT_TITLES:
                break
            if _is_page_break_only(p) or (not t) or (not lvl):
                el.getparent().remove(el)
                el = nxt
                continue
            break
        if tag == "sdt":
            el.getparent().remove(el)
            el = nxt
            continue
        break


def insert_toc_field(doc: Document) -> int:
    toc = None
    for p in doc.paragraphs:
        if p.text.strip() == "目录":
            toc = p
            break
    if toc is None:
        return 0
    _remove_toc_fields(doc)
    _clear_after_toc_title(doc, toc)
    p_el = OxmlElement("w:p")
    for kind, payload in (
        ("begin", None),
        ("instr", ' TOC \\o "1-4" \\h \\z \\u '),
        ("separate", None),
        ("text", ""),
        ("end", None),
    ):
        r = OxmlElement("w:r")
        rPr = OxmlElement("w:rPr")
        color = OxmlElement("w:color")
        color.set(qn("w:val"), "000000")
        rPr.append(color)
        r.append(rPr)
        if kind == "instr":
            it = OxmlElement("w:instrText")
            it.set("{http://www.w3.org/XML/1998/namespace}space", "preserve")
            it.text = payload
            r.append(it)
        elif kind == "text":
            t_el = OxmlElement("w:t")
            t_el.text = payload or ""
            r.append(t_el)
        else:
            fc = OxmlElement("w:fldChar")
            fc.set(qn("w:fldCharType"), kind)
            r.append(fc)
        p_el.append(r)
    toc._element.addnext(p_el)
    body_head = None
    after = False
    for p in doc.paragraphs:
        if p.text.strip() == "目录":
            after = True
            continue
        if after and heading_level(p) and p.text.strip() not in FRONT_TITLES:
            body_head = p
            break
    if body_head is not None:
        xml = _xml(body_head._element)
        if 'w:type="page"' not in xml and "w:type='page'" not in xml:
            text = body_head.text
            body_head.text = ""
            br = body_head.add_run()
            br.add_break(WD_BREAK.PAGE)
            run = body_head.add_run(text)
            run.font.color.rgb = BLACK
    return 1


def rebuild_static_toc(doc: Document) -> int:
    demote_front_titles(doc)
    return insert_toc_field(doc)


def _version_col(table: Table) -> int:
    headers = [c.text.replace("\n", "").strip() for c in table.rows[0].cells]
    for i, h in enumerate(headers):
        if h in {"版本", "版本号"}:
            return i
    return 1 if len(headers) > 1 else 0


def dedupe_version_table(doc: Document) -> int:
    n = 0
    for p in doc.paragraphs:
        if p.text.strip() not in {"版本记录", "更改记录"}:
            continue
        el = p._element.getnext()
        while el is not None:
            tag = el.tag.split("}")[-1]
            if tag == "tbl":
                table = Table(el, doc)
                col = _version_col(table)
                seen = set()
                for row in list(table.rows)[1:]:
                    ver = row.cells[col].text.replace("\n", "").strip() if col < len(row.cells) else ""
                    if not ver:
                        continue
                    if ver in seen:
                        table._tbl.remove(row._tr)
                        n += 1
                    else:
                        seen.add(ver)
                return n
            if tag == "p" and "".join(el.itertext()).strip():
                return 0
            el = el.getnext()
    return n


def collapse_blank_pages(doc: Document) -> int:
    n = 0
    paras = list(doc.paragraphs)
    prev_break = False
    for p in paras:
        if _is_page_break_only(p):
            if prev_break:
                remove_paragraph(p)
                n += 1
                continue
            prev_break = True
        elif p.text.strip() or heading_level(p):
            prev_break = False
    paras = list(doc.paragraphs)
    while paras and _is_page_break_only(paras[-1]):
        remove_paragraph(paras[-1])
        n += 1
        paras = list(doc.paragraphs)
    return n


def drop_disclaimer_and_perf(doc: Document, *, keep_perf: bool) -> int:
    n = 0
    dropping_section = False
    for p in list(doc.paragraphs):
        t = p.text.strip()
        if not t:
            continue
        lvl = heading_level(p)
        if lvl:
            if DROP_HEAD_RE.match(t) and not keep_perf:
                dropping_section = True
                remove_paragraph(p)
                n += 1
                continue
            dropping_section = False
        if dropping_section and not keep_perf:
            remove_paragraph(p)
            n += 1
            continue
        if DISCLAIMER_RE.search(t):
            remove_paragraph(p)
            n += 1
            continue
        if (not keep_perf) and PERF_RE.search(t):
            if t.startswith("请使用 Chrome"):
                set_paragraph_text(p, "请使用 Chrome 或 Edge 近两个正式版。")
                n += 1
                continue
            remove_paragraph(p)
            n += 1
    for table in doc.tables:
        for row in table.rows:
            for cell in row.cells:
                for p in cell.paragraphs:
                    raw = p.text.strip()
                    if not raw:
                        continue
                    if DISCLAIMER_RE.search(raw) or ((not keep_perf) and PERF_RE.search(raw)):
                        if raw in {"文档编号", "版本", "密级", "编制单位", "编制日期"}:
                            continue
                        if re.match(r"^V\d", raw):
                            continue
                        set_paragraph_text(p, "")
                        n += 1
    return n


def is_sys_brd(path: Path) -> bool:
    return "业务需求描述-总纲" in path.name


def drop_h1_sections(doc: Document, title_re: re.Pattern[str]) -> int:
    n = 0
    dropping = False
    body = doc.element.body
    for child in list(body.iterchildren()):
        tag = child.tag.split("}")[-1]
        if tag == "p":
            p = Paragraph(child, doc)
            t = p.text.strip()
            lvl = heading_level(p)
            if lvl == 1 and title_re.search(t):
                dropping = True
                body.remove(child)
                n += 1
                continue
            if dropping:
                if t in FRONT_TITLES:
                    dropping = False
                    continue
                if lvl == 1:
                    dropping = False
                    continue
                body.remove(child)
                n += 1
        elif tag == "tbl" and dropping:
            body.remove(child)
            n += 1
    return n


SAAS_NFR_H1_RE = re.compile(r"SaaS\s*部署与运维|非功能设计")


def drop_references_section(doc: Document) -> int:
    n = 0
    dropping = False
    drop_lvl = 1
    body = doc.element.body
    for child in list(body.iterchildren()):
        tag = child.tag.split("}")[-1]
        if tag == "p":
            p = Paragraph(child, doc)
            t = p.text.strip()
            lvl = heading_level(p)
            if REF_HEAD_RE.match(t) and lvl:
                dropping = True
                drop_lvl = lvl
                body.remove(child)
                n += 1
                continue
            if dropping:
                if t in {"目录", "版本记录"}:
                    dropping = False
                    continue
                if lvl and lvl <= drop_lvl:
                    dropping = False
                    continue
                body.remove(child)
                n += 1
        elif tag == "tbl" and dropping:
            body.remove(child)
            n += 1
    return n


def polish_doc(doc: Document, path: Path) -> dict:
    keep_perf = is_sys_brd(path)
    dropped = drop_disclaimer_and_perf(doc, keep_perf=keep_perf)
    dropped += drop_references_section(doc)
    renum = renumber_heading_gaps(doc)
    demote_front_titles(doc)
    dupes = dedupe_version_table(doc)
    toc_n = insert_toc_field(doc)
    blanks = collapse_blank_pages(doc)
    colors = force_all_text_black(doc)
    return {
        "drop": dropped,
        "renum": renum,
        "dupe": dupes,
        "toc": toc_n,
        "blank": blanks,
        "runs": colors,
    }


def polish_file(path: Path) -> str:
    doc = open_via_temp(path)
    stats = polish_doc(doc, path)
    save_via_temp(doc, path)
    return (
        f"OK {path.parent.name}/{path.name} "
        f"drop={stats['drop']} renum={stats['renum']} dupe={stats['dupe']} toc={stats['toc']} blank={stats['blank']}"
    )


def main() -> None:
    from _unify_ab_delivery_vnext import BASE

    files = latest_formal_files(BASE)
    print("polish", len(files), "files", flush=True)
    for p in files:
        try:
            msg = polish_file(p)
        except Exception as exc:
            msg = f"FAIL {p.parent.name}/{p.name}: {exc}"
        print(msg, flush=True)


if __name__ == "__main__":
    main()