新增多证书模板管理流程

backend/app/api/routes/admin_imports.py

@@ -3,7 +3,7 @@ import shutil
 from datetime import date, datetime
 from pathlib import Path
 
-from fastapi import APIRouter, Depends, File, HTTPException, UploadFile, status
+from fastapi import APIRouter, Depends, File, Form, HTTPException, UploadFile, status
 from fastapi.responses import FileResponse, StreamingResponse
 from openpyxl import Workbook, load_workbook
 from openpyxl.comments import Comment
@@ -28,6 +28,7 @@ from app.models import (
 from app.schemas.import_batch import ImportBatchOut
 from app.services.certificate_number import build_certificate_no
 from app.services.logs import log_action
+from app.services.certificate_templates import get_certificate_template
 
 router = APIRouter()
 
@@ -74,18 +75,23 @@ def build_import_template_workbook() -> Workbook:
 
 @router.post("", response_model=ImportBatchOut, status_code=status.HTTP_201_CREATED)
 def upload_import_file(
+    template_code: str = Form(...),
     file: UploadFile = File(...),
     db: Session = Depends(get_db),
     admin: AdminUser = Depends(require_roles("system_admin", "certificate_admin")),
 ) -> ImportBatch:
     if not file.filename or not file.filename.lower().endswith(".xlsx"):
         raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail="Only .xlsx files are supported")
+    try:
+        template_code = get_certificate_template(template_code).code
+    except (ValueError, FileNotFoundError) as exc:
+        raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail=str(exc)) from exc
 
     upload_path = data_path("uploads") / file.filename
     with upload_path.open("wb") as target:
         shutil.copyfileobj(file.file, target)
 
-    batch = ImportBatch(filename=file.filename, file_path=str(upload_path), created_by=admin.id)
+    batch = ImportBatch(filename=file.filename, file_path=str(upload_path), template_code=template_code, created_by=admin.id)
     db.add(batch)
     db.flush()
     validate_batch(db, batch, upload_path)
@@ -188,7 +194,15 @@ def confirm_import_batch(
             row.error_message = f"Project code is inactive or missing: {project_code}"
             failed_rows += 1
             continue
-        duplicate = find_duplicate_certificate(db, learner.id, project, course_start_date, course_end_date, issue_date)
+        duplicate = find_duplicate_certificate(
+            db,
+            learner.id,
+            project,
+            course_start_date,
+            course_end_date,
+            issue_date,
+            batch.template_code,
+        )
         if duplicate:
             row.status = "skipped"
             row.error_message = "\u5df2\u5b58\u5728\uff0c\u65e0\u9700\u5904\u7406"
@@ -207,6 +221,7 @@ def confirm_import_batch(
             course_end_date=course_end_date,
             issue_date=issue_date,
             issuer_name=project.default_issuer_name,
+            template_code=batch.template_code,
             remark=None,
         )
         db.add(certificate)
@@ -326,6 +341,7 @@ def find_duplicate_certificate(
     course_start_date: date,
     course_end_date: date,
     issue_date: date,
+    template_code: str,
 ) -> Certificate | None:
     return (
         db.query(Certificate)
@@ -337,6 +353,7 @@ def find_duplicate_certificate(
         .filter(Certificate.course_start_date == course_start_date)
         .filter(Certificate.course_end_date == course_end_date)
         .filter(Certificate.issue_date == issue_date)
+        .filter(Certificate.template_code == template_code)
         .first()
     )