修复证书导入模板识别与文件重选

backend/app/api/routes/admin_imports.py

@@ -17,7 +17,11 @@ from app.db.session import get_db
 from app.models import AdminUser, ImportBatch, ImportBatchRow, Learner, ProjectCourse
 from app.schemas.import_batch import ImportBatchOut
 from app.services.certificate_issuance import CertificateIssueData, DuplicateCertificate, issue_certificate
-from app.services.certificate_templates import CertificateTemplateDefinition, get_certificate_template
+from app.services.certificate_templates import (
+    CertificateTemplateDefinition,
+    get_certificate_template,
+    list_certificate_templates,
+)
 from app.services.learner_identity import normalize_phone
 from app.services.logs import log_action
 
@@ -99,6 +103,14 @@ def upload_import_file(
     with upload_path.open("wb") as target:
         shutil.copyfileobj(file.file, target)
 
+    detected_template = detect_template_from_workbook(upload_path)
+    if detected_template and detected_template.code != template.code:
+        upload_path.unlink(missing_ok=True)
+        raise HTTPException(
+            status_code=status.HTTP_400_BAD_REQUEST,
+            detail=f"文件列属于“{detected_template.name}”，当前选择的是“{template.name}”。请切换证书模板后重新上传。",
+        )
+
     batch = ImportBatch(
         filename=file.filename,
         file_path=str(upload_path),
@@ -332,6 +344,20 @@ def required_headers(template: CertificateTemplateDefinition) -> list[str]:
     return COMMON_HEADERS + [FIELD_COLUMNS[field.key] for field in template.fields if field.key != "learner_name" and field.required]
 
 
+def detect_template_from_workbook(upload_path: Path) -> CertificateTemplateDefinition | None:
+    workbook = load_workbook(upload_path, read_only=True, data_only=True)
+    try:
+        sheet = workbook.active
+        first_row = next(sheet.iter_rows(min_row=1, max_row=1, values_only=True), None)
+        header_names = {str(value).strip() for value in (first_row or ()) if value is not None and str(value).strip()}
+        for template in list_certificate_templates():
+            if header_names == set(template_headers(template)):
+                return template
+        return None
+    finally:
+        workbook.close()
+
+
 def optional_text(value: object) -> str | None:
     if value is None:
         return None

backend/tests/test_import_rules.py

@@ -20,6 +20,7 @@ from app.api.routes.admin_imports import (
     TEMPLATE_HEADERS,
     build_import_template_workbook,
     date_is_valid,
+    detect_template_from_workbook,
     normalize_row_data,
     parse_issue_date,
     row_errors,
@@ -50,6 +51,16 @@ def test_normalize_row_data_removes_time_from_excel_dates():
     assert row[COL_ISSUE_DATE] == "2026-07-05"
 
 
+def test_detect_template_from_workbook_uses_excel_headers(tmp_path):
+    source_path = tmp_path / "practice-camp.xlsx"
+    build_import_template_workbook("practice-camp").save(source_path)
+
+    detected = detect_template_from_workbook(source_path)
+
+    assert detected is not None
+    assert detected.code == "practice-camp"
+
+
 def test_row_errors_require_project_code_to_exist():
     row = {
         COL_NAME: "张三",