feat: parse docx blocks during upload

backend/app/services/doc_parser.py

@@ -1,56 +1,92 @@
 """Word 文档解析引擎"""
+from io import BytesIO
+from typing import Any, Dict, Iterable, List
+
 from docx import Document
-from docx.paragraph import Paragraph
+from docx.document import Document as DocxDocument
+from docx.oxml.table import CT_Tbl
+from docx.oxml.text.paragraph import CT_P
 from docx.table import Table as DocxTable
-from typing import List, Dict, Any
+from docx.table import _Cell
+from docx.text.paragraph import Paragraph
 
 
-def parse_document(file_path: str) -> List[Dict[str, Any]]:
+def parse_document(source: str | bytes) -> List[Dict[str, Any]]:
     """
-    解析 Word 文档，返回 block 列表（不含表格数据，仅标记位置）。
-    每个 block 包含：type, text, level, style
+    解析 Word 文档，返回带 block_id、parent_block_id、sort_order 的 block 列表。
     """
-    doc = Document(file_path)
-    blocks = []
+    doc = Document(BytesIO(source)) if isinstance(source, bytes) else Document(source)
+    blocks: List[Dict[str, Any]] = []
+
+    for item in iter_block_items(doc):
+        if isinstance(item, Paragraph):
+            block = classify_paragraph(item)
+        else:
+            block = classify_table(item)
 
-    for para in doc.paragraphs:
-        block = classify_paragraph(para)
         if block:
             blocks.append(block)
 
+    assign_block_ids(blocks)
+    build_tree(blocks)
     return blocks
 
 
+def iter_block_items(parent: DocxDocument | _Cell) -> Iterable[Paragraph | DocxTable]:
+    """按 Word 文档中的实际顺序遍历段落和表格。"""
+    parent_element = parent.element.body if isinstance(parent, DocxDocument) else parent._tc
+
+    for child in parent_element.iterchildren():
+        if isinstance(child, CT_P):
+            yield Paragraph(child, parent)
+        elif isinstance(child, CT_Tbl):
+            yield DocxTable(child, parent)
+
+
 def classify_paragraph(para: Paragraph) -> Dict[str, Any] | None:
     """识别段落类型"""
     text = para.text.strip()
     style_name = para.style.name if para.style else ""
 
-    if not text and "Heading" not in style_name:
+    if not text:
         return None  # 跳过空段落
 
-    if "Heading" in style_name:
-        level = int(style_name.replace("Heading", "").strip()) if style_name.replace("Heading", "").strip().isdigit() else 1
+    if is_heading_style(style_name):
+        level = parse_heading_level(style_name)
         return {"type": "heading", "text": text, "level": level, "style": style_name}
-    else:
-        return {"type": "paragraph", "text": text, "level": 0, "style": style_name}
+
+    return {"type": "paragraph", "text": text, "level": 0, "style": style_name}
 
 
-def parse_tables(doc: Document) -> List[Dict[str, Any]]:
-    """解析 Word 表格"""
-    tables = []
-    for i, table in enumerate(doc.tables):
-        headers = [cell.text.strip() for cell in table.rows[0].cells] if table.rows else []
-        tables.append({
-            "type": "table",
-            "text": " | ".join(headers),
-            "level": 0,
-            "rows": len(table.rows),
-            "cols": len(table.columns),
-            "headers": headers,
-            "table_index": i,
-        })
-    return tables
+def is_heading_style(style_name: str) -> bool:
+    return style_name.startswith("Heading") or style_name.startswith("标题")
+
+
+def parse_heading_level(style_name: str) -> int:
+    parts = style_name.replace("Heading", "").replace("标题", "").strip()
+    return int(parts) if parts.isdigit() and 1 <= int(parts) <= 6 else 1
+
+
+def classify_table(table: DocxTable) -> Dict[str, Any] | None:
+    """识别表格类型并提取行列数。"""
+    rows = len(table.rows)
+    cols = len(table.columns)
+    if rows == 0 or cols == 0:
+        return None
+
+    first_row = [cell.text.strip() for cell in table.rows[0].cells]
+    text = " | ".join(cell for cell in first_row if cell)
+    if not text:
+        text = f"表格（{rows} 行 x {cols} 列）"
+
+    return {
+        "type": "table",
+        "text": text,
+        "level": 0,
+        "rows": rows,
+        "cols": cols,
+        "headers": first_row,
+    }
 
 
 def generate_block_id(index: int) -> str:
@@ -58,29 +94,46 @@ def generate_block_id(index: int) -> str:
     return f"block_{index:03d}"
 
 
-def build_tree(blocks: List[Dict]) -> Dict:
+def assign_block_ids(blocks: List[Dict[str, Any]]) -> None:
+    for index, block in enumerate(blocks, start=1):
+        block["block_id"] = generate_block_id(index)
+        block["sort_order"] = index
+        block["parent_block_id"] = None
+
+
+def build_tree(blocks: List[Dict[str, Any]]) -> List[Dict[str, Any]]:
     """构建层级树结构"""
-    tree = []
-    stack = []  # 存父级节点
+    tree: List[Dict[str, Any]] = []
+    heading_stack: List[Dict[str, Any]] = []
 
     for block in blocks:
-        node = {"block_id": block["block_id"], "text": block["text"][:50], "children": []}
-        level = block.get("level", 0)
+        node = {
+            "block_id": block["block_id"],
+            "block_type": block["type"],
+            "text": block["text"][:50],
+            "level": block.get("level", 0),
+            "children": [],
+        }
 
-        # 回退栈
-        while stack and stack[-1]["level"] >= level:
-            stack.pop()
+        if block["type"] == "heading":
+            level = block.get("level", 1)
 
-        if stack:
-            parent = stack[-1]["node"]
+            while heading_stack and heading_stack[-1]["level"] >= level:
+                heading_stack.pop()
+
+            if heading_stack:
+                parent = heading_stack[-1]["node"]
+                parent["children"].append(node)
+                block["parent_block_id"] = parent["block_id"]
+            else:
+                tree.append(node)
+
+            heading_stack.append({"level": level, "node": node})
+        elif heading_stack:
+            parent = heading_stack[-1]["node"]
             parent["children"].append(node)
             block["parent_block_id"] = parent["block_id"]
         else:
             tree.append(node)
-            block["parent_block_id"] = None
 
-        # 非叶子节点入栈
-        if level < 2 or block["type"] == "heading":
-            stack.append({"level": level, "node": node})
-
-    return {"blocks": tree}
+    return tree

backend/app/services/template_service.py

@@ -6,6 +6,7 @@ from uuid import uuid4
 from fastapi import UploadFile
 from sqlalchemy.orm import Session
 from ..models.template import Template
+from ..models.template_block import TemplateBlock
 from .storage import upload_file
 
 
@@ -52,3 +53,26 @@ def create_template_record(db: Session, name: str, file_path: str, type: str = "
     db.commit()
     db.refresh(tmpl)
     return tmpl
+
+
+def save_template_blocks(db: Session, template_id: int, blocks: list[dict]) -> list[TemplateBlock]:
+    """批量保存解析出的模板区域。"""
+    records = [
+        TemplateBlock(
+            template_id=template_id,
+            block_id=block["block_id"],
+            parent_block_id=block.get("parent_block_id"),
+            block_type=block["type"],
+            block_name=block.get("text", "")[:200] or None,
+            text_preview=block.get("text", "")[:500] or None,
+            level=block.get("level", 0),
+            sort_order=block.get("sort_order", index),
+            table_rows=block.get("rows"),
+            table_cols=block.get("cols"),
+        )
+        for index, block in enumerate(blocks, start=1)
+    ]
+
+    db.add_all(records)
+    db.commit()
+    return records