init project

2026-07-01 19:50:29 +08:00
commit 685d69ebdc
21 changed files with 1002 additions and 0 deletions
@@ -0,0 +1,86 @@
+"""Word 文档解析引擎"""
+from docx import Document
+from docx.paragraph import Paragraph
+from docx.table import Table as DocxTable
+from typing import List, Dict, Any
+
+
+def parse_document(file_path: str) -> List[Dict[str, Any]]:
+    """
+    解析 Word 文档，返回 block 列表（不含表格数据，仅标记位置）。
+    每个 block 包含：type, text, level, style
+    """
+    doc = Document(file_path)
+    blocks = []
+
+    for para in doc.paragraphs:
+        block = classify_paragraph(para)
+        if block:
+            blocks.append(block)
+
+    return blocks
+
+
+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:
+        return None  # 跳过空段落
+
+    if "Heading" in style_name:
+        level = int(style_name.replace("Heading", "").strip()) if style_name.replace("Heading", "").strip().isdigit() else 1
+        return {"type": "heading", "text": text, "level": level, "style": style_name}
+    else:
+        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 generate_block_id(index: int) -> str:
+    """生成 block_id，格式：block_001"""
+    return f"block_{index:03d}"
+
+
+def build_tree(blocks: List[Dict]) -> Dict:
+    """构建层级树结构"""
+    tree = []
+    stack = []  # 存父级节点
+
+    for block in blocks:
+        node = {"block_id": block["block_id"], "text": block["text"][:50], "children": []}
+        level = block.get("level", 0)
+
+        # 回退栈
+        while stack and stack[-1]["level"] >= level:
+            stack.pop()
+
+        if stack:
+            parent = 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}
@@ -0,0 +1,27 @@
+"""HTML 预览生成服务"""
+from typing import List, Dict, Any
+
+
+def generate_preview_html(blocks: List[Dict[str, Any]]) -> str:
+    """
+    将 block 列表转为带 data-block-id 的 HTML 预览字符串。
+    每个元素都会带上 data-block-id 属性供前端点击交互。
+    """
+    html_parts = ['<div class="doc-preview-content">']
+
+    for block in blocks:
+        block_id = block.get("block_id", "")
+        block_type = block.get("type", "")
+        text = block.get("text", "")
+
+        if block_type == "heading":
+            html = f'<h2 data-block-id="{block_id}" style="font-size:16px;font-weight:600;margin:20px 0 10px;padding-bottom:6px;border-bottom:2px solid #1a1d24">{text}</h2>'
+        elif block_type == "table":
+            html = f'<div data-block-id="{block_id}" style="margin:12px 0;padding:8px;background:#f7f8fa;border:1px dashed #ccc;border-radius:4px;color:#5b626e">📊 {text[:80]}</div>'
+        else:
+            html = f'<p data-block-id="{block_id}" style="text-indent:2em;margin:8px 0;line-height:1.8;text-align:justify">{text}</p>'
+
+        html_parts.append(html)
+
+    html_parts.append('</div>')
+    return "\n".join(html_parts)
@@ -0,0 +1,28 @@
+import os
+from io import BytesIO
+
+# 先实现本地文件存储，MinIO 作为选项
+STORAGE_BASE = os.getenv("LOCAL_STORAGE_PATH", "./data/files")
+
+
+def save_file(content: bytes, file_path: str) -> str:
+    """保存文件到本地存储，返回完整路径"""
+    full_path = os.path.join(STORAGE_BASE, file_path)
+    os.makedirs(os.path.dirname(full_path), exist_ok=True)
+    with open(full_path, "wb") as f:
+        f.write(content)
+    return full_path
+
+
+def read_file(file_path: str) -> bytes:
+    """读取文件内容"""
+    full_path = os.path.join(STORAGE_BASE, file_path)
+    with open(full_path, "rb") as f:
+        return f.read()
+
+
+def delete_file(file_path: str):
+    """删除文件"""
+    full_path = os.path.join(STORAGE_BASE, file_path)
+    if os.path.exists(full_path):
+        os.remove(full_path)
@@ -0,0 +1,30 @@
+"""模板相关业务逻辑"""
+from fastapi import UploadFile
+from sqlalchemy.orm import Session
+from ..models.template import Template
+
+
+ALLOWED_EXTENSIONS = {".docx"}
+MAX_FILE_SIZE = 50 * 1024 * 1024  # 50MB
+
+
+def validate_file(file: UploadFile):
+    """校验文件格式和大小"""
+    ext = "." + file.filename.rsplit(".", 1)[-1].lower() if "." in file.filename else ""
+    if ext not in ALLOWED_EXTENSIONS:
+        raise ValueError(f"不支持的文件格式: {ext}，仅支持 .docx")
+    # 读取文件头校验大小
+    content = file.file.read()
+    file.file.seek(0)
+    if len(content) > MAX_FILE_SIZE:
+        raise ValueError(f"文件大小超过限制 (50MB)")
+    return content
+
+
+def create_template_record(db: Session, name: str, file_path: str, type: str = "report") -> Template:
+    """创建模板记录"""
+    tmpl = Template(name=name, type=type, original_file_path=file_path)
+    db.add(tmpl)
+    db.commit()
+    db.refresh(tmpl)
+    return tmpl