pj0235-eai_agentplatform/docs/04_Backend/BE03_Media_Module.md

BE03 — 素材模块设计

版本：V1.1 | 技术：分片上传 + LibreOffice + PyMuPDF

---

1. 模块职责

• 文件上传（直传 + 分片上传）
• 素材审批流（待审批 → 通过/驳回）
• 审批通过后异步文档转换管线
• 文件预览
• 转换状态查询

2. 素材状态流转

员工提交 / 管理员上传
      │
      ▼
  pending（待审批） ──┬─ approve → approved（已通过）
      │               │
      │               └─→ 触发异步转换管线
      │                       │
      │                       ├─ 文档 → LibreOffice 转 PDF → PyMuPDF 提取
      │                       │   文本 → 切片写入 knowledge_chunk
      │                       └─ 视频/图片 → 仅标记预览可用
      │
      └─ reject → rejected（已驳回，前台不可见）

3. 上传 API

直传（文档 ≤ 200MB）

POST /api/media/upload
Content-Type: multipart/form-data

Parameters:
  - file: 文件二进制
  - bind_type: company | product | course | none
  - bind_id: 绑定实体 ID（可选）

Response:
  {
    "media_id": 1,
    "status": "pending",
    "filename": "原始名称.pptx"
  }

分片上传（视频 > 100MB）

# 1. 初始化
POST /api/media/upload-init
{
  "filename": "training.mp4",
  "file_size": 524288000,
  "bind_type": "course",
  "bind_id": 1
}
Response: { "upload_id": "uuid", "chunk_size": 5242880, "chunk_count": 100 }

# 2. 上传分片（循环调用）
POST /api/media/upload-chunk
Content-Type: multipart/form-data
{
  "upload_id": "uuid",
  "chunk_index": 0,
  "file": <binary>
}

# 3. 完成合并
POST /api/media/upload-complete
{ "upload_id": "uuid" }
Response: { "media_id": 1, "status": "pending" }

4. 审批 API

# 管理员
GET  /api/media/audit-list?status=pending&page=1&size=20

POST /api/media/audit/{mediaId}
{
  "action": "approve",          # approve | reject
  "reject_reason": "..."        # 驳回时必填
}

5. 异步转换管线

# services/media_service.py
import threading
from datetime import datetime

def _async_convert_pipeline(media_id: int):
    """审批通过后的异步转换管线"""
    try:
        media = db.query(MediaFile).get(media_id)
        
        # 仅文档需要转换（PPT/Word/PDF）
        if media.file_ext in ("ppt", "pptx", "doc", "docx"):
            # Step 1: LibreOffice 转 PDF
            pdf_path = _libreoffice_to_pdf(media.stored_path)
            
            # Step 2: PyMuPDF 提取文本
            text = _pymupdf_extract(pdf_path)
            
            # Step 3: 按段落切片入库
            chunks = _split_into_chunks(text)
            for i, chunk_text in enumerate(chunks):
                db.add(KnowledgeChunk(
                    media_file_id=media.id,
                    source_type=media.file_ext,
                    chunk_index=i,
                    content=chunk_text,
                ))
        elif media.file_ext == "pdf":
            # PDF 直接 PyMuPDF 提取
            text = _pymupdf_extract(media.stored_path)
            chunks = _split_into_chunks(text)
            for i, chunk_text in enumerate(chunks):
                db.add(KnowledgeChunk(media_file_id=media.id, ...))
        
        # 视频/图片：不提取文本
        media.extracted = True
        db.commit()
        logger.info(f"转换完成: media_id={media_id}")
    except Exception as e:
        logger.error(f"转换失败: media_id={media_id}, error={e}")
        media.extracted = False  # 标记失败可重试

def approve_media(media_id: int, auditor_id: int):
    """审批通过 → 启动异步转换"""
    media = db.query(MediaFile).get(media_id)
    media.status = "approved"
    media.audit_by = auditor_id
    media.audit_at = datetime.utcnow()
    db.commit()
    
    thread = threading.Thread(target=_async_convert_pipeline, args=(media_id,))
    thread.start()

6. LibreOffice 转换接口

# utils/libreoffice.py
import subprocess
import requests

def libreoffice_convert(input_path: str, output_dir: str) -> str:
    """调用 LibreOffice 容器将文档转 PDF"""
    # 方式1：本地安装 libreoffice
    subprocess.run([
        "libreoffice", "--headless", "--convert-to", "pdf",
        "--outdir", output_dir, input_path
    ], check=True)
    
    # 方式2：Docker 容器 HTTP 接口
    # response = requests.post(
    #     f"{settings.libreoffice_url}/convert",
    #     files={"file": open(input_path, "rb")}
    # )
    # return response.json()["pdf_path"]

7. PyMuPDF 文本提取

# utils/pdf_extractor.py
import fitz  # PyMuPDF

def extract_text(pdf_path: str) -> str:
    """提取 PDF 全部文本"""
    doc = fitz.open(pdf_path)
    text = ""
    for page in doc:
        text += page.get_text()
    doc.close()
    return text

def split_into_chunks(text: str, max_chars: int = 1000) -> list[str]:
    """按段落 + 最大字符数切片"""
    paragraphs = text.split("\n\n")
    chunks = []
    current = ""
    for p in paragraphs:
        if len(current) + len(p) > max_chars:
            if current:
                chunks.append(current.strip())
            current = p
        else:
            current += "\n\n" + p if current else p
    if current:
        chunks.append(current.strip())
    return chunks

8. 预览 API

GET /api/media/preview/{mediaId}
# 仅 approved 素材可预览
Response:
{
  "preview_url": "/media/upload/uuid-filename.pdf",
  "file_ext": "pdf",
  "can_preview": true
}

GET /api/media/status/{mediaId}
# 查询素材状态（含提取进度）
Response:
{
  "status": "approved",
  "extracted": true,
  "chunk_count": 42
}

9. 安全约束

• 扩展名白名单：ppt/pptx/pdf/doc/docx/mp4/png/jpg/jpeg
• 文件名重命名为 UUID，杜绝路径穿越
• 上传目录对静态预览只读，禁止直接执行
• 文件大小：文档 ≤ 200MB，视频 ≤ 2GB
• MIME 类型校验 + 扩展名双重校验