|
|
@@ -1,6 +1,7 @@
|
|
|
"""导入路由"""
|
|
|
|
|
|
import json
|
|
|
+import queue
|
|
|
import secrets
|
|
|
import threading
|
|
|
from pathlib import Path
|
|
|
@@ -19,6 +20,26 @@ from app.services.backup import auto_backup
|
|
|
|
|
|
router = APIRouter()
|
|
|
|
|
|
+# 单线程导入队列
|
|
|
+_import_queue = queue.Queue()
|
|
|
+
|
|
|
+def _import_worker():
|
|
|
+ """单线程消费导入任务,避免并发线程过多导致崩溃"""
|
|
|
+ while True:
|
|
|
+ task = _import_queue.get()
|
|
|
+ if task is None:
|
|
|
+ break
|
|
|
+ task_id, tmp_path, import_type, source_name, config = task
|
|
|
+ try:
|
|
|
+ _process_import(task_id, tmp_path, import_type, source_name, config)
|
|
|
+ except Exception:
|
|
|
+ pass
|
|
|
+ _import_queue.task_done()
|
|
|
+
|
|
|
+# 启动后台工作线程
|
|
|
+_worker = threading.Thread(target=_import_worker, daemon=True)
|
|
|
+_worker.start()
|
|
|
+
|
|
|
|
|
|
@router.post("/import")
|
|
|
async def import_docx(
|
|
|
@@ -27,19 +48,16 @@ async def import_docx(
|
|
|
import_type: str = Form("精选题"),
|
|
|
_=Depends(verify_token),
|
|
|
):
|
|
|
- """上传整卷 docx 进行导入(异步处理)"""
|
|
|
+ """上传整卷 docx 进行导入(队列处理)"""
|
|
|
config = request.app.state.config
|
|
|
|
|
|
- # 验证文件类型
|
|
|
if not file.filename.endswith('.docx'):
|
|
|
raise HTTPException(status_code=400, detail={
|
|
|
"error": {"code": "invalid_file", "message": "仅支持 .docx 文件"}
|
|
|
})
|
|
|
|
|
|
- # 生成任务 ID
|
|
|
task_id = secrets.token_hex(8)
|
|
|
|
|
|
- # 保存上传文件
|
|
|
upload_dir = config.data_dir / "uploads"
|
|
|
upload_dir.mkdir(parents=True, exist_ok=True)
|
|
|
tmp_path = upload_dir / f"{task_id}.docx"
|
|
|
@@ -48,19 +66,14 @@ async def import_docx(
|
|
|
with open(tmp_path, "wb") as f:
|
|
|
f.write(content)
|
|
|
|
|
|
- # 创建任务
|
|
|
conn = get_connection(config.db_path)
|
|
|
try:
|
|
|
create_import_task(conn, task_id)
|
|
|
finally:
|
|
|
conn.close()
|
|
|
|
|
|
- # 后台处理
|
|
|
- def process():
|
|
|
- _process_import(task_id, tmp_path, import_type, file.filename, config)
|
|
|
-
|
|
|
- thread = threading.Thread(target=process, daemon=True)
|
|
|
- thread.start()
|
|
|
+ # 放入队列,单线程顺序处理
|
|
|
+ _import_queue.put((task_id, tmp_path, import_type, file.filename, config))
|
|
|
|
|
|
return {"task_id": task_id, "status": "pending"}
|
|
|
|