zwt13703
79 lines
2.6 KiB
Python
79 lines
2.6 KiB
Python
"""模板相关业务逻辑"""
|
|
from datetime import datetime
|
|
from pathlib import Path
|
|
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
|
|
|
|
|
|
ALLOWED_EXTENSIONS = {".docx"}
|
|
MAX_FILE_SIZE = 50 * 1024 * 1024 # 50MB
|
|
TEMPLATE_BUCKET = "templates"
|
|
|
|
|
|
def validate_file(file: UploadFile):
|
|
"""校验文件格式和大小"""
|
|
if not file.filename:
|
|
raise ValueError("文件名不能为空")
|
|
|
|
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 build_template_file_path(filename: str) -> str:
|
|
"""生成模板文件存储路径。"""
|
|
suffix = Path(filename).suffix.lower()
|
|
today = datetime.now().strftime("%Y%m%d")
|
|
return f"{today}/{uuid4().hex}{suffix}"
|
|
|
|
|
|
def store_template_file(file: UploadFile) -> tuple[str, bytes]:
|
|
"""校验并保存上传的模板文件，返回存储路径和文件内容。"""
|
|
content = validate_file(file)
|
|
file_path = build_template_file_path(file.filename or "template.docx")
|
|
stored_path = upload_file(TEMPLATE_BUCKET, file_path, content)
|
|
return stored_path, 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
|
|
|
|
|
|
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
|