131 lines
5.1 KiB
Python
131 lines
5.1 KiB
Python
import os
|
|
import tempfile
|
|
import pypandoc
|
|
from typing import Optional
|
|
from fastapi import APIRouter, HTTPException, Depends
|
|
from fastapi.responses import FileResponse
|
|
from pydantic import BaseModel
|
|
from starlette.background import BackgroundTask
|
|
import docx
|
|
from docx.shared import Pt
|
|
from docx.oxml.ns import qn
|
|
import auth
|
|
import db
|
|
from formula_processor import preprocess_formulas
|
|
|
|
router = APIRouter()
|
|
|
|
class StyleOptions(BaseModel):
|
|
base_font_size: Optional[int] = 12
|
|
base_font_family: Optional[str] = "宋体"
|
|
h1_size: Optional[int] = 18
|
|
h2_size: Optional[int] = 16
|
|
h3_size: Optional[int] = 14
|
|
h4_size: Optional[int] = 12
|
|
|
|
class DocRequest(BaseModel):
|
|
markdown: str
|
|
styleOptions: Optional[StyleOptions] = None
|
|
|
|
def remove_file(path: str):
|
|
try:
|
|
os.remove(path)
|
|
except Exception:
|
|
pass
|
|
|
|
@router.post("/api/convert")
|
|
async def convert_md_to_docx(req: DocRequest, current_user: Optional[dict] = Depends(auth.get_optional_user)):
|
|
try:
|
|
# 如果带有合法登录 token，则记录历史
|
|
if current_user:
|
|
db.add_history_record(current_user["id"], req.markdown)
|
|
|
|
try:
|
|
pypandoc.get_pandoc_version()
|
|
except OSError:
|
|
pypandoc.download_pandoc()
|
|
|
|
# ====== 公式预处理：统一各种定界符格式 ======
|
|
# 将 GPT/AI 生成内容中 \(...\)、\[...\]、缺失反斜杠的 [...] 和 (...)
|
|
# 全部转换为 Pandoc 可识别的标准 $...$ 和 $$...$$ 格式
|
|
processed_markdown = preprocess_formulas(req.markdown)
|
|
|
|
fd, tmp_path = tempfile.mkstemp(suffix=".docx")
|
|
os.close(fd)
|
|
|
|
md_fd, md_tmp_path = tempfile.mkstemp(suffix=".md")
|
|
with os.fdopen(md_fd, 'w', encoding='utf-8') as f:
|
|
f.write(processed_markdown)
|
|
|
|
pypandoc.convert_file(
|
|
md_tmp_path,
|
|
'docx',
|
|
format='markdown+tex_math_dollars',
|
|
outputfile=tmp_path
|
|
)
|
|
os.remove(md_tmp_path)
|
|
|
|
# ====== 样式后处理 ======
|
|
if req.styleOptions:
|
|
doc = docx.Document(tmp_path)
|
|
|
|
def set_font(style_name, font_name, font_size):
|
|
try:
|
|
style = doc.styles[style_name]
|
|
if font_size:
|
|
style.font.size = Pt(font_size)
|
|
if font_name:
|
|
style.font.name = font_name
|
|
style._element.rPr.rFonts.set(qn('w:eastAsia'), font_name)
|
|
except KeyError:
|
|
pass
|
|
|
|
set_font('Normal', req.styleOptions.base_font_family, req.styleOptions.base_font_size)
|
|
set_font('Heading 1', req.styleOptions.base_font_family, req.styleOptions.h1_size)
|
|
set_font('Heading 2', req.styleOptions.base_font_family, req.styleOptions.h2_size)
|
|
set_font('Heading 3', req.styleOptions.base_font_family, req.styleOptions.h3_size)
|
|
set_font('Heading 4', req.styleOptions.base_font_family, req.styleOptions.h4_size)
|
|
set_font('Heading 5', req.styleOptions.base_font_family, req.styleOptions.h4_size)
|
|
set_font('Heading 6', req.styleOptions.base_font_family, req.styleOptions.h4_size)
|
|
|
|
# 强制遍历并覆盖所有段落的 run，确保样式生效
|
|
for p in doc.paragraphs:
|
|
# 判定当前段落是不是标题
|
|
current_font_size = req.styleOptions.base_font_size
|
|
current_font_family = req.styleOptions.base_font_family
|
|
|
|
if p.style.name.startswith('Heading'):
|
|
level_str = p.style.name.replace('Heading ', '')
|
|
try:
|
|
level = int(level_str)
|
|
if level == 1: current_font_size = req.styleOptions.h1_size
|
|
elif level == 2: current_font_size = req.styleOptions.h2_size
|
|
elif level == 3: current_font_size = req.styleOptions.h3_size
|
|
elif level >= 4: current_font_size = req.styleOptions.h4_size
|
|
except:
|
|
pass
|
|
|
|
# 为该段落中每一个具体的 run 设置字体
|
|
for run in p.runs:
|
|
run.font.name = current_font_family
|
|
run._element.rPr.rFonts.set(qn('w:eastAsia'), current_font_family)
|
|
if current_font_size:
|
|
run.font.size = Pt(current_font_size)
|
|
|
|
doc.save(tmp_path)
|
|
|
|
return FileResponse(
|
|
tmp_path,
|
|
media_type='application/vnd.openxmlformats-officedocument.wordprocessingml.document',
|
|
filename='document.docx',
|
|
background=BackgroundTask(remove_file, tmp_path)
|
|
)
|
|
except Exception as e:
|
|
print(f"Error during conversion: {e}")
|
|
raise HTTPException(status_code=500, detail=str(e))
|
|
|
|
@router.get("/api/history")
|
|
async def get_history(current_user: dict = Depends(auth.get_current_user)):
|
|
history = db.get_history_records(current_user["id"])
|
|
return {"history": history}
|