实现模板解析与模板管理基础链路
This commit is contained in:
zwt13703
@@ -0,0 +1,37 @@
|
||||
from minio import Minio
|
||||
from config import settings
|
||||
|
||||
# MinIO 客户端
|
||||
minio_client = Minio(
|
||||
settings.MINIO_ENDPOINT,
|
||||
access_key=settings.MINIO_ACCESS_KEY,
|
||||
secret_key=settings.MINIO_SECRET_KEY,
|
||||
secure=settings.MINIO_USE_SSL,
|
||||
)
|
||||
|
||||
|
||||
async def init_buckets():
|
||||
"""初始化 MinIO 存储桶（应用启动时调用）"""
|
||||
buckets = [
|
||||
settings.MINIO_BUCKET_TEMPLATES, # 原始模板文件
|
||||
settings.MINIO_BUCKET_UPLOADS, # 用户上传的参考文件
|
||||
settings.MINIO_BUCKET_OUTPUTS, # 生成的文档
|
||||
]
|
||||
for bucket in buckets:
|
||||
if not minio_client.bucket_exists(bucket):
|
||||
minio_client.make_bucket(bucket)
|
||||
print(f"[MinIO] 创建存储桶: {bucket}")
|
||||
|
||||
|
||||
def get_file_url(bucket: str, object_name: str) -> str:
|
||||
"""获取文件的公开访问 URL"""
|
||||
if settings.MINIO_USE_SSL:
|
||||
protocol = "https"
|
||||
else:
|
||||
protocol = "http"
|
||||
return f"{protocol}://{settings.MINIO_ENDPOINT}/{bucket}/{object_name}"
|
||||
|
||||
|
||||
def get_presigned_url(bucket: str, object_name: str, expires: int = 3600) -> str:
|
||||
"""获取预签名下载 URL（带过期时间）"""
|
||||
return minio_client.presigned_get_object(bucket, object_name, expires=expires)
|
||||
@@ -0,0 +1,216 @@
|
||||
import json
|
||||
from collections.abc import Iterator
|
||||
from dataclasses import dataclass
|
||||
|
||||
from docx import Document
|
||||
from docx.document import Document as DocumentObject
|
||||
from docx.oxml.ns import qn
|
||||
from docx.oxml.table import CT_Tbl
|
||||
from docx.oxml.text.paragraph import CT_P
|
||||
from docx.table import Table
|
||||
from docx.text.paragraph import Paragraph
|
||||
from docx.enum.text import WD_ALIGN_PARAGRAPH
|
||||
|
||||
|
||||
@dataclass
|
||||
class ParsedParagraph:
|
||||
sort_index: int
|
||||
title: str
|
||||
content: str
|
||||
style_json: str
|
||||
is_table: bool
|
||||
table_json: str
|
||||
|
||||
|
||||
def _iter_block_items(document: DocumentObject) -> Iterator[Paragraph | Table]:
|
||||
body = document.element.body
|
||||
for child in body.iterchildren():
|
||||
if isinstance(child, CT_P):
|
||||
yield Paragraph(child, document)
|
||||
elif isinstance(child, CT_Tbl):
|
||||
yield Table(child, document)
|
||||
|
||||
|
||||
def _safe_pt(value: object) -> float | None:
|
||||
if value is None:
|
||||
return None
|
||||
try:
|
||||
return round(float(value.pt), 2)
|
||||
except AttributeError:
|
||||
return None
|
||||
|
||||
|
||||
def _safe_indent(value: object) -> float | None:
|
||||
if value is None:
|
||||
return None
|
||||
try:
|
||||
return round(float(value.pt), 2)
|
||||
except AttributeError:
|
||||
return None
|
||||
|
||||
|
||||
def _alignment_name(value: WD_ALIGN_PARAGRAPH | None) -> str:
|
||||
if value is None:
|
||||
return "LEFT"
|
||||
return getattr(value, "name", "LEFT")
|
||||
|
||||
|
||||
def _heading_level(style_name: str) -> int | None:
|
||||
if not style_name:
|
||||
return None
|
||||
normalized = style_name.lower().replace(" ", "")
|
||||
if normalized.startswith("heading"):
|
||||
level = normalized.replace("heading", "")
|
||||
if level.isdigit():
|
||||
return int(level)
|
||||
return None
|
||||
|
||||
|
||||
def _get_run_font_info(paragraph: Paragraph) -> dict:
|
||||
for run in paragraph.runs:
|
||||
if not run.text.strip():
|
||||
continue
|
||||
r_fonts = getattr(run._element.rPr, "rFonts", None) if run._element.rPr is not None else None
|
||||
east_asia = r_fonts.get(qn("w:eastAsia")) if r_fonts is not None else None
|
||||
color = None
|
||||
if run.font.color is not None and run.font.color.rgb is not None:
|
||||
color = str(run.font.color.rgb)
|
||||
return {
|
||||
"name": run.font.name,
|
||||
"eastAsia": east_asia,
|
||||
"size": _safe_pt(run.font.size),
|
||||
"bold": bool(run.bold) if run.bold is not None else False,
|
||||
"italic": bool(run.italic) if run.italic is not None else False,
|
||||
"color": color or "000000",
|
||||
}
|
||||
return {
|
||||
"name": None,
|
||||
"eastAsia": None,
|
||||
"size": None,
|
||||
"bold": False,
|
||||
"italic": False,
|
||||
"color": "000000",
|
||||
}
|
||||
|
||||
|
||||
def _capture_paragraph_style(paragraph: Paragraph, level: int) -> dict:
|
||||
fmt = paragraph.paragraph_format
|
||||
return {
|
||||
"font": _get_run_font_info(paragraph),
|
||||
"paragraph": {
|
||||
"alignment": _alignment_name(paragraph.alignment),
|
||||
"spaceBefore": _safe_pt(fmt.space_before),
|
||||
"spaceAfter": _safe_pt(fmt.space_after),
|
||||
"lineSpacing": fmt.line_spacing,
|
||||
"firstLineIndent": _safe_indent(fmt.first_line_indent),
|
||||
},
|
||||
"headingLevel": level,
|
||||
}
|
||||
|
||||
|
||||
def _get_cell_style(cell) -> dict:
|
||||
paragraph = cell.paragraphs[0] if cell.paragraphs else None
|
||||
font_info = _get_run_font_info(paragraph) if paragraph is not None else {
|
||||
"name": None,
|
||||
"eastAsia": None,
|
||||
"size": None,
|
||||
"bold": False,
|
||||
"italic": False,
|
||||
"color": "000000",
|
||||
}
|
||||
return {
|
||||
"font": font_info,
|
||||
"shading": None,
|
||||
"alignment": _alignment_name(paragraph.alignment) if paragraph is not None else "LEFT",
|
||||
"borders": {"top": None, "bottom": None, "left": None, "right": None},
|
||||
}
|
||||
|
||||
|
||||
def _extract_table_data(table: Table) -> dict:
|
||||
rows = len(table.rows)
|
||||
cols = max((len(row.cells) for row in table.rows), default=0)
|
||||
grid_span: dict[str, int] = {}
|
||||
cell_styles: list[dict] = []
|
||||
matrix: list[list[str]] = []
|
||||
|
||||
for row_index, row in enumerate(table.rows):
|
||||
row_values: list[str] = []
|
||||
for col_index, cell in enumerate(row.cells):
|
||||
text = "\n".join(paragraph.text.strip() for paragraph in cell.paragraphs if paragraph.text.strip())
|
||||
row_values.append(text)
|
||||
tc_pr = cell._tc.tcPr
|
||||
grid_span_value = None
|
||||
if tc_pr is not None and tc_pr.gridSpan is not None:
|
||||
grid_span_value = tc_pr.gridSpan.val
|
||||
if grid_span_value:
|
||||
grid_span[f"{row_index}-{col_index}"] = int(grid_span_value)
|
||||
cell_styles.append(_get_cell_style(cell))
|
||||
matrix.append(row_values)
|
||||
|
||||
return {
|
||||
"rows": rows,
|
||||
"cols": cols,
|
||||
"gridSpan": grid_span,
|
||||
"cellStyles": cell_styles,
|
||||
"tableWidth": None,
|
||||
"data": matrix,
|
||||
}
|
||||
|
||||
|
||||
def parse_template(file_path: str) -> list[ParsedParagraph]:
|
||||
document = Document(file_path)
|
||||
parsed: list[ParsedParagraph] = []
|
||||
current_item: ParsedParagraph | None = None
|
||||
loose_table_count = 0
|
||||
|
||||
for block in _iter_block_items(document):
|
||||
if isinstance(block, Paragraph):
|
||||
text = block.text.strip()
|
||||
if not text:
|
||||
continue
|
||||
|
||||
level = _heading_level(block.style.name if block.style is not None else "")
|
||||
if level is not None:
|
||||
current_item = ParsedParagraph(
|
||||
sort_index=len(parsed) + 1,
|
||||
title=text,
|
||||
content="",
|
||||
style_json=json.dumps(_capture_paragraph_style(block, level), ensure_ascii=False),
|
||||
is_table=False,
|
||||
table_json="{}",
|
||||
)
|
||||
parsed.append(current_item)
|
||||
continue
|
||||
|
||||
if current_item is None:
|
||||
current_item = ParsedParagraph(
|
||||
sort_index=len(parsed) + 1,
|
||||
title="未命名段落",
|
||||
content=text,
|
||||
style_json=json.dumps(_capture_paragraph_style(block, 0), ensure_ascii=False),
|
||||
is_table=False,
|
||||
table_json="{}",
|
||||
)
|
||||
parsed.append(current_item)
|
||||
else:
|
||||
current_item.content = "\n".join(filter(None, [current_item.content, text]))
|
||||
else:
|
||||
table_data = _extract_table_data(block)
|
||||
table_text = f"[表格] {table_data['rows']} 行 {table_data['cols']} 列"
|
||||
if current_item is None:
|
||||
loose_table_count += 1
|
||||
current_item = ParsedParagraph(
|
||||
sort_index=len(parsed) + 1,
|
||||
title=f"表格_{loose_table_count}",
|
||||
content=table_text,
|
||||
style_json="{}",
|
||||
is_table=True,
|
||||
table_json=json.dumps(table_data, ensure_ascii=False),
|
||||
)
|
||||
parsed.append(current_item)
|
||||
else:
|
||||
current_item.is_table = True
|
||||
current_item.table_json = json.dumps(table_data, ensure_ascii=False)
|
||||
current_item.content = "\n".join(filter(None, [current_item.content, table_text]))
|
||||
|
||||
return parsed
|
||||
Reference in New Issue
Block a user