core/services/docx_export.py

"""HTML → 专业 Word(.docx) 报告转换器（纯 python-docx，确定性）。

设计目标（解决"大段文字 + AI 符号 * + 缺乏可视化/专业性"的问题）：
- 复用 :class:`ReportService.assemble` 产出的**品牌化 HTML**作为唯一事实源，转写为排版
  规范的 Word 文档：真正的 Word 表格（表头底纹）、分级标题、项目符号列表、干净段落。
- **剥离 AI/Markdown 噪声**：``**粗体**`` / ``*斜体*`` / ``#标题`` / 行首 ``* - +`` 列表符
  等一律去标记，保留文字。``<style>/<script>/<svg>`` 整块丢弃。
- 中文字体：在 Normal 与标题样式上设置 ``eastAsia`` 字体，保证中文在 Word 中正常显示。

借鉴专业报告（McKinsey/FDA/MiniMax 取向）的版式：封面标题 + 副标题 + 元信息行 +
分节标题 + 图表化表格 + 简洁叙述，杜绝整页无结构的大段文字。

仅依赖标准库 ``html.parser`` 与 ``python-docx``；任何异常由调用方降级（返回 None）。
"""

from __future__ import annotations

import re
from html.parser import HTMLParser
from html import unescape
from io import BytesIO
from typing import Any, Optional

# 块级标签（用于分段）。
_BLOCK_TAGS = {
    "p", "div", "section", "h1", "h2", "h3", "h4", "h5", "h6",
    "table", "thead", "tbody", "tr", "td", "th", "ul", "ol", "li",
    "br", "header", "footer", "caption",
}
# 整块忽略的标签（连同内容）。
_SKIP_TAGS = {"style", "script", "svg", "head", "title", "noscript"}
# 空（void）元素：无闭合标签，解析时绝不入栈、不计 skip 深度（否则会吞掉后续兄弟节点）。
_VOID_TAGS = {"br", "meta", "link", "img", "hr", "input", "col", "area", "base", "source", "wbr"}


class _Node:
    __slots__ = ("tag", "attrs", "children", "text")

    def __init__(self, tag: str, attrs: Optional[dict] = None) -> None:
        self.tag = tag
        self.attrs = attrs or {}
        self.children: list = []  # _Node 或 str
        self.text = ""

    def cls(self) -> str:
        return str(self.attrs.get("class", "") or "")


class _DOMBuilder(HTMLParser):
    """把 HTML 解析为轻量 DOM 树；跳过 style/script/svg 等噪声标签的内容。"""

    def __init__(self) -> None:
        super().__init__(convert_charrefs=True)
        self.root = _Node("root")
        self._stack = [self.root]
        self._skip_depth = 0

    def handle_starttag(self, tag, attrs):
        if tag in _VOID_TAGS:
            # 空元素：不入栈、不改变 skip 深度。
            if self._skip_depth:
                return
            if tag == "br":
                self._stack[-1].children.append(_Node("br"))
            elif tag == "img":
                # 捕获图片（data-uri）为节点，供 Word 嵌入。
                self._stack[-1].children.append(_Node("img", {k: (v or "") for k, v in attrs}))
            return
        if self._skip_depth:
            if tag in _SKIP_TAGS:
                self._skip_depth += 1
            return
        if tag in _SKIP_TAGS:
            self._skip_depth = 1
            return
        if tag == "br":
            self._stack[-1].children.append(_Node("br"))
            return
        node = _Node(tag, {k: (v or "") for k, v in attrs})
        self._stack[-1].children.append(node)
        self._stack.append(node)

    def handle_startendtag(self, tag, attrs):
        if self._skip_depth or tag in _SKIP_TAGS:
            return
        self._stack[-1].children.append(_Node(tag, {k: (v or "") for k, v in attrs}))

    def handle_endtag(self, tag):
        if self._skip_depth:
            if tag in _SKIP_TAGS:
                self._skip_depth -= 1
            return
        # 关闭到匹配标签（容忍不规范嵌套）。
        for i in range(len(self._stack) - 1, 0, -1):
            if self._stack[i].tag == tag:
                del self._stack[i:]
                break

    def handle_data(self, data):
        if self._skip_depth or not data:
            return
        self._stack[-1].children.append(data)


# ---------------------------------------------------------------------------
# 文本清理（去 Markdown / AI 噪声）
# ---------------------------------------------------------------------------

def _decode_data_uri_png(src: str) -> Optional[bytes]:
    """从 ``data:image/...;base64,XXXX`` 解出图片字节；非法/非内联返回 None。"""
    import base64
    s = str(src or "")
    if not s.startswith("data:image"):
        return None
    marker = ";base64,"
    idx = s.find(marker)
    if idx < 0:
        return None
    try:
        return base64.b64decode(s[idx + len(marker):])
    except Exception:  # noqa: BLE001
        return None


def _strip_markdown(text: str) -> str:
    if not text:
        return text
    out = text
    out = re.sub(r"\*{2,3}([^\n*]+?)\*{2,3}", r"\1", out)
    out = re.sub(r"__([^\n_]+?)__", r"\1", out)
    out = re.sub(r"(?<!\*)\*(?!\s)([^\n*]+?)(?<!\s)\*(?!\*)", r"\1", out)
    out = re.sub(r"(?m)^\s{0,3}#{1,6}\s+", "", out)
    out = re.sub(r"(?m)^\s{0,3}>\s?", "", out)
    out = re.sub(r"(?m)^\s{0,3}[-*+]\s+", "", out)
    return out


def _node_text(node: "_Node") -> str:
    """收集节点所有后代文本，去 Markdown、收敛空白。"""
    parts: list[str] = []

    def walk(n):
        if isinstance(n, str):
            parts.append(n)
            return
        if n.tag == "br":
            parts.append("\n")
            return
        for c in n.children:
            walk(c)

    walk(node)
    text = unescape("".join(parts))
    text = _strip_markdown(text)
    # 收敛空白：行内多空格→单空格；保留换行。
    text = re.sub(r"[ \t\u00a0]+", " ", text)
    text = re.sub(r"\n{2,}", "\n", text)
    return text.strip()


__all__ = ["html_to_docx", "docx_available"]


def docx_available() -> bool:
    try:
        import docx  # noqa: F401
        return True
    except Exception:  # noqa: BLE001
        return False


# ---------------------------------------------------------------------------
# 块抽取：把 DOM 扁平化为有序块列表
# ---------------------------------------------------------------------------

def _is_heading_div(node: "_Node") -> Optional[int]:
    """根据 class 判定 div 是否为标题，返回标题级别；否则 None。"""
    c = node.cls()
    if "brand-name" in c:
        return 0          # 文档标题
    if "report-type" in c:
        return 1          # 报告类型（副标题级）
    if "section-title" in c:
        return 2          # 分节标题
    return None


def _extract_table(node: "_Node") -> dict:
    """把 <table> 解析为 {rows: [[cell,...]], header: bool}。"""
    rows: list[list[str]] = []
    header = False

    def collect_rows(n):
        nonlocal header
        for c in n.children:
            if isinstance(c, str):
                continue
            if c.tag == "tr":
                cells: list[str] = []
                row_is_header = False
                for cell in c.children:
                    if isinstance(cell, str):
                        continue
                    if cell.tag in ("td", "th"):
                        cells.append(_node_text(cell))
                        if cell.tag == "th":
                            row_is_header = True
                if cells:
                    if row_is_header and not rows:
                        header = True
                    rows.append(cells)
            else:
                collect_rows(c)

    collect_rows(node)
    return {"rows": rows, "header": header}


def _extract_blocks(root: "_Node") -> list[dict]:
    """深度遍历 DOM，产出有序块：heading/para/bullet/table/kpi/note。"""
    blocks: list[dict] = []

    def emit_para(text: str, *, kind: str = "para"):
        text = text.strip()
        if text:
            blocks.append({"type": kind, "text": text})

    def walk(node: "_Node"):
        for child in node.children:
            if isinstance(child, str):
                # 裸文本（位于容器直下）→ 段落。
                t = _strip_markdown(unescape(child))
                t = re.sub(r"[ \t\u00a0]+", " ", t).strip()
                if t:
                    emit_para(t)
                continue
            tag = child.tag
            if tag in _SKIP_TAGS or tag == "br":
                continue
            if tag == "img":
                png = _decode_data_uri_png(child.attrs.get("src", ""))
                if png:
                    blocks.append({"type": "image", "png": png})
                continue
            if tag == "table":
                tbl = _extract_table(child)
                if tbl["rows"]:
                    blocks.append({"type": "table", **tbl})
                continue
            if tag in ("h1", "h2", "h3", "h4", "h5", "h6"):
                blocks.append({"type": "heading", "level": int(tag[1]), "text": _node_text(child)})
                continue
            if tag in ("p", "li"):
                emit_para(_node_text(child), kind=("bullet" if tag == "li" else "para"))
                continue
            if tag == "div":
                # KPI 卡：value + label。
                c = child.cls()
                if "kpi-card" in c:
                    val = ""
                    lab = ""
                    for sub in child.children:
                        if isinstance(sub, _Node):
                            sc = sub.cls()
                            if "kpi-value" in sc:
                                val = _node_text(sub)
                            elif "kpi-label" in sc:
                                lab = _node_text(sub)
                    if val or lab:
                        blocks.append({"type": "kpi", "value": val, "label": lab})
                    continue
                if "callout" in c:
                    emit_para(_node_text(child), kind="note")
                    continue
                lvl = _is_heading_div(child)
                if lvl is not None:
                    blocks.append({"type": "heading", "level": lvl, "text": _node_text(child)})
                    continue
                if "brand-subtitle" in c or "report-date" in c:
                    emit_para(_node_text(child), kind="subtitle")
                    continue
                # 普通容器：递归。
                walk(child)
                continue
            # 其它容器标签：递归。
            if tag in ("section", "header", "footer", "ul", "ol", "thead", "tbody",
                       "tr", "span", "main", "article", "body", "html", "caption"):
                walk(child)
                continue
            # 兜底：递归。
            walk(child)

    walk(root)
    return _coalesce_kpis(blocks)


def _coalesce_kpis(blocks: list[dict]) -> list[dict]:
    """把连续的 kpi 块合并为一个 kpi-row 块，便于渲染为单行小表。"""
    out: list[dict] = []
    buf: list[dict] = []
    for b in blocks:
        if b["type"] == "kpi":
            buf.append(b)
            continue
        if buf:
            out.append({"type": "kpi_row", "items": [(k["label"], k["value"]) for k in buf]})
            buf = []
        out.append(b)
    if buf:
        out.append({"type": "kpi_row", "items": [(k["label"], k["value"]) for k in buf]})
    return out


# ---------------------------------------------------------------------------
# docx 渲染
# ---------------------------------------------------------------------------

def _shade_cell(cell, fill: str) -> None:
    from docx.oxml.ns import qn
    from docx.oxml import OxmlElement
    tcPr = cell._tc.get_or_add_tcPr()
    shd = OxmlElement("w:shd")
    shd.set(qn("w:val"), "clear")
    shd.set(qn("w:color"), "auto")
    shd.set(qn("w:fill"), fill)
    tcPr.append(shd)


def _set_run_font(run, *, latin: Optional[str] = None, east_asian: Optional[str] = None) -> None:
    """显式设置单个 run 的拉丁字体与中文（eastAsia）字体。"""
    from docx.oxml.ns import qn
    rpr = run._element.get_or_add_rPr()
    rfonts = rpr.get_or_add_rFonts()
    if latin:
        run.font.name = latin
        rfonts.set(qn("w:ascii"), latin)
        rfonts.set(qn("w:hAnsi"), latin)
    if east_asian:
        rfonts.set(qn("w:eastAsia"), east_asian)


def _apply_document_fonts(
    doc,
    *,
    body_latin: str,
    body_cjk: str,
    heading_latin: str,
    heading_cjk: str,
) -> None:
    """按 SCI / 学术规范设置全文字体：

    - 正文（Normal）：拉丁衬线（Times New Roman）+ 中文宋体。
    - 标题（Title / Heading 1–4）：拉丁无衬线（Arial）+ 中文黑体。

    同时设置 ``ascii`` / ``hAnsi``（拉丁）与 ``eastAsia``（中文）三类字形槽，
    保证中英文混排时各自落到正确字体。字体在打开 Word 的机器上解析（Office 自带
    Times New Roman / Arial / 宋体 / 黑体）；缺失时由 Word 自动替换。
    """
    from docx.oxml.ns import qn

    def _set_style(style_name: str, latin: str, cjk: str) -> None:
        try:
            style = doc.styles[style_name]
        except KeyError:
            return
        style.font.name = latin
        rpr = style.element.get_or_add_rPr()
        rfonts = rpr.get_or_add_rFonts()
        rfonts.set(qn("w:ascii"), latin)
        rfonts.set(qn("w:hAnsi"), latin)
        rfonts.set(qn("w:eastAsia"), cjk)

    _set_style("Normal", body_latin, body_cjk)
    for hs in ("Title", "Heading 1", "Heading 2", "Heading 3", "Heading 4"):
        _set_style(hs, heading_latin, heading_cjk)


def html_to_docx(
    html: str,
    *,
    brand: str = "Pharma K",
    font_name: str = "宋体",
    body_latin: str = "Times New Roman",
    body_cjk: str = "宋体",
    heading_latin: str = "Arial",
    heading_cjk: str = "黑体",
) -> Optional[bytes]:
    """把品牌化报告 HTML 转换为专业 Word(.docx) 字节流；失败返回 ``None``。

    字体遵循 SCI / 学术排版规范：正文 Times New Roman + 中文宋体；标题 / 表头
    Arial + 中文黑体。``font_name`` 保留作向后兼容（等同 ``body_cjk``）。
    """
    if not html:
        return None
    # 向后兼容：旧调用以 font_name 指定中文正文字体。
    if font_name and font_name != "宋体":
        body_cjk = font_name
    try:
        from docx import Document
        from docx.shared import Pt, RGBColor, Inches
        from docx.enum.text import WD_ALIGN_PARAGRAPH

        builder = _DOMBuilder()
        builder.feed(html)
        blocks = _extract_blocks(builder.root)

        doc = Document()
        # 基础样式：正文 10.5pt；SCI 字体（正文衬线 + 中文宋体，标题无衬线 + 中文黑体）。
        normal = doc.styles["Normal"]
        normal.font.size = Pt(10.5)
        _apply_document_fonts(
            doc,
            body_latin=body_latin, body_cjk=body_cjk,
            heading_latin=heading_latin, heading_cjk=heading_cjk,
        )

        accent = RGBColor(0x1F, 0x4E, 0x79)  # 专业深蓝

        for b in blocks:
            btype = b["type"]
            if btype == "heading":
                lvl = b.get("level", 2)
                text = b.get("text", "")
                if not text:
                    continue
                if lvl == 0:
                    p = doc.add_heading("", level=0)
                    run = p.add_run(text)
                    run.font.color.rgb = accent
                    p.alignment = WD_ALIGN_PARAGRAPH.CENTER
                else:
                    doc.add_heading(text, level=min(max(lvl, 1), 4))
            elif btype == "subtitle":
                p = doc.add_paragraph()
                run = p.add_run(b.get("text", ""))
                run.italic = True
                run.font.size = Pt(9)
                run.font.color.rgb = RGBColor(0x66, 0x66, 0x66)
                p.alignment = WD_ALIGN_PARAGRAPH.CENTER
            elif btype == "note":
                p = doc.add_paragraph()
                run = p.add_run(b.get("text", ""))
                run.italic = True
                run.font.color.rgb = RGBColor(0x57, 0x60, 0x6A)
            elif btype == "bullet":
                try:
                    doc.add_paragraph(b.get("text", ""), style="List Bullet")
                except Exception:  # noqa: BLE001
                    doc.add_paragraph("• " + b.get("text", ""))
            elif btype == "kpi_row":
                items = b.get("items", [])
                if not items:
                    continue
                tbl = doc.add_table(rows=2, cols=len(items))
                tbl.style = "Table Grid"
                for j, (label, value) in enumerate(items):
                    vc = tbl.cell(0, j)
                    vc.text = value
                    for r in vc.paragraphs[0].runs:
                        r.bold = True
                        r.font.size = Pt(13)
                        r.font.color.rgb = accent
                        _set_run_font(r, latin=heading_latin, east_asian=heading_cjk)
                    vc.paragraphs[0].alignment = WD_ALIGN_PARAGRAPH.CENTER
                    lc = tbl.cell(1, j)
                    lc.text = label
                    lc.paragraphs[0].alignment = WD_ALIGN_PARAGRAPH.CENTER
                    for r in lc.paragraphs[0].runs:
                        r.font.size = Pt(8)
                        r.font.color.rgb = RGBColor(0x66, 0x66, 0x66)
                        _set_run_font(r, latin=heading_latin, east_asian=heading_cjk)
                    _shade_cell(lc, "F0F3F5")
            elif btype == "image":
                # 内联 PNG → Word 图片（受限宽度）；解码/嵌入失败则跳过不崩溃。
                try:
                    doc.add_picture(BytesIO(b["png"]), width=Inches(6.0))
                    doc.paragraphs[-1].alignment = WD_ALIGN_PARAGRAPH.CENTER
                except Exception:  # noqa: BLE001
                    pass
            elif btype == "table":
                rows = b.get("rows", [])
                if not rows:
                    continue
                ncols = max(len(r) for r in rows)
                tbl = doc.add_table(rows=len(rows), cols=ncols)
                tbl.style = "Table Grid"
                for i, row in enumerate(rows):
                    for j in range(ncols):
                        cell = tbl.cell(i, j)
                        cell.text = row[j] if j < len(row) else ""
                        if i == 0 and b.get("header"):
                            _shade_cell(cell, "1F4E79")
                            for para in cell.paragraphs:
                                for r in para.runs:
                                    r.bold = True
                                    r.font.color.rgb = RGBColor(0xFF, 0xFF, 0xFF)
                                    r.font.size = Pt(9)
                                    _set_run_font(r, latin=heading_latin, east_asian=heading_cjk)
                        else:
                            for para in cell.paragraphs:
                                for r in para.runs:
                                    r.font.size = Pt(9)
                doc.add_paragraph()  # 表后留白
            else:  # para
                text = b.get("text", "")
                for line in text.split("\n"):
                    line = line.strip()
                    if line:
                        doc.add_paragraph(line)

        buf = BytesIO()
        doc.save(buf)
        return buf.getvalue()
    except Exception:  # noqa: BLE001 - 任何失败由调用方降级
        return None