修复证书导入日期解析与错误报告

backend/app/api/routes/admin_imports.py

@@ -136,8 +136,13 @@ def download_error_report(
     _: AdminUser = Depends(require_roles("system_admin", "certificate_admin")),
 ) -> FileResponse:
     batch = db.get(ImportBatch, batch_id)
-    if not batch or not batch.error_report_path:
+    if not batch:
+        raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="导入批次不存在")
+    failed_count = db.query(ImportBatchRow).filter(ImportBatchRow.batch_id == batch.id, ImportBatchRow.status == "failed").count()
+    if not failed_count:
         raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="错误报告不存在")
+    batch.error_report_path = str(write_error_report(db, batch.id))
+    db.commit()
     return FileResponse(batch.error_report_path, media_type="application/vnd.openxmlformats-officedocument.spreadsheetml.sheet", filename=f"import-errors-{batch.id}.xlsx")
 
 
@@ -221,6 +226,9 @@ def confirm_import_batch(
 
     batch.status = "imported" if imported_rows or skipped_rows else "failed"
     batch.failed_rows = (batch.failed_rows or 0) + failed_rows
+    if failed_rows:
+        db.flush()
+        batch.error_report_path = str(write_error_report(db, batch.id))
     log_action(
         db,
         admin,
@@ -255,7 +263,9 @@ def validate_batch(db: Session, batch: ImportBatch, upload_path: Path) -> None:
         if not any(row):
             continue
         total += 1
-        row_data = {name: row[header_index[name]] if name in header_index and header_index[name] < len(row) else None for name in headers}
+        row_data = normalize_row_data(
+            {name: row[header_index[name]] if name in header_index and header_index[name] < len(row) else None for name in headers}
+        )
         errors = row_errors(row_data, active_codes, template.code, db)
         row_status = "failed" if errors else "valid"
         failed += int(bool(errors))
@@ -274,6 +284,7 @@ def validate_batch(db: Session, batch: ImportBatch, upload_path: Path) -> None:
     batch.failed_rows = failed
     batch.status = "validated" if total else "failed"
     if failed:
+        db.flush()
         batch.error_report_path = str(write_error_report(db, batch.id))
 
 
@@ -328,6 +339,18 @@ def optional_text(value: object) -> str | None:
     return text or None
 
 
+def normalize_row_data(row_data: dict[str, object]) -> dict[str, object]:
+    normalized: dict[str, object] = {}
+    for key, value in row_data.items():
+        if isinstance(value, datetime):
+            normalized[key] = value.date().isoformat()
+        elif isinstance(value, date):
+            normalized[key] = value.isoformat()
+        else:
+            normalized[key] = value
+    return normalized
+
+
 def parse_issue_date(value: object) -> date:
     return parse_date(value, COL_ISSUE_DATE)
 
@@ -342,6 +365,14 @@ def parse_date(value: object, field_name: str = "日期") -> date:
     if isinstance(value, date):
         return value
     text = str(value).strip()
+    try:
+        return date.fromisoformat(text)
+    except ValueError:
+        pass
+    try:
+        return datetime.fromisoformat(text.replace("Z", "+00:00")).date()
+    except ValueError:
+        pass
     for fmt in ["%Y-%m-%d", "%Y/%m/%d", "%Y.%m.%d"]:
         try:
             return datetime.strptime(text, fmt).date()
@@ -359,18 +390,42 @@ def date_is_valid(value: object) -> bool:
 
 
 def write_error_report(db: Session, batch_id: int) -> Path:
+    db.flush()
+    batch = db.get(ImportBatch, batch_id)
+    template = get_certificate_template(batch.template_code) if batch else get_certificate_template("classic")
+    headers = template_headers(template)
     workbook = Workbook()
     sheet = workbook.active
     sheet.title = "错误报告"
-    sheet.append(["行号", "错误原因", "原始数据"])
+    sheet.append(["行号", "错误原因", *headers])
     rows = db.query(ImportBatchRow).filter(ImportBatchRow.batch_id == batch_id, ImportBatchRow.status == "failed").all()
     for row in rows:
-        sheet.append([row.row_no, row.error_message, row.raw_json])
+        raw_data = json.loads(row.raw_json or "{}")
+        sheet.append([row.row_no, row.error_message or "未知错误", *(report_cell_value(header, raw_data.get(header)) for header in headers)])
+    sheet.freeze_panes = "A2"
+    sheet.auto_filter.ref = f"A1:{sheet.cell(1, len(headers) + 2).coordinate}"
+    sheet.column_dimensions["A"].width = 10
+    sheet.column_dimensions["B"].width = 48
+    for cell in sheet[1]:
+        cell.fill = PatternFill("solid", fgColor="C0392B")
+        cell.font = Font(color="FFFFFF", bold=True)
+        cell.alignment = Alignment(horizontal="center", vertical="center")
+    for row in sheet.iter_rows(min_row=2):
+        row[1].alignment = Alignment(wrap_text=True, vertical="top")
     report_path = data_path("error-reports") / f"import-errors-{batch_id}.xlsx"
     workbook.save(report_path)
     return report_path
 
 
+def report_cell_value(header: str, value: object) -> object:
+    if header not in {COL_COURSE_START_DATE, COL_COURSE_END_DATE, COL_ISSUE_DATE} or value in (None, ""):
+        return value
+    try:
+        return parse_date(value, header).isoformat()
+    except ValueError:
+        return value
+
+
 def _format_template_sheet(sheet, headers: list[str], template: CertificateTemplateDefinition) -> None:
     header_fill = PatternFill("solid", fgColor="208A87")
     for cell in sheet[1]:

backend/tests/test_import_rules.py

@@ -1,8 +1,13 @@
-from datetime import date
+import json
+from datetime import date, datetime
 
 import pytest
+from openpyxl import load_workbook
 from pydantic import ValidationError
+from sqlalchemy import create_engine
+from sqlalchemy.orm import Session
 
+from app.api.routes import admin_imports
 from app.api.routes.admin_imports import (
     COL_ISSUE_DATE,
     COL_COURSE_END_DATE,
@@ -15,15 +20,34 @@ from app.api.routes.admin_imports import (
     TEMPLATE_HEADERS,
     build_import_template_workbook,
     date_is_valid,
+    normalize_row_data,
     parse_issue_date,
     row_errors,
+    write_error_report,
 )
+from app.db.base import Base
+from app.models import ImportBatch, ImportBatchRow
 from app.schemas.certificate import CertificateCreate
 
 
 def test_parse_issue_date_accepts_common_formats():
     assert parse_issue_date("2026-06-01") == date(2026, 6, 1)
     assert parse_issue_date("2026/06/01") == date(2026, 6, 1)
+    assert parse_issue_date("2026-06-01 00:00:00") == date(2026, 6, 1)
+    assert parse_issue_date("2026-06-01T08:30:00") == date(2026, 6, 1)
+
+
+def test_normalize_row_data_removes_time_from_excel_dates():
+    row = normalize_row_data(
+        {
+            COL_NAME: "张三",
+            COL_COURSE_START_DATE: datetime(2026, 6, 1, 0, 0, 0),
+            COL_ISSUE_DATE: date(2026, 7, 5),
+        }
+    )
+
+    assert row[COL_COURSE_START_DATE] == "2026-06-01"
+    assert row[COL_ISSUE_DATE] == "2026-07-05"
 
 
 def test_row_errors_require_project_code_to_exist():
@@ -81,6 +105,49 @@ def test_import_template_contains_date_examples_and_validation():
     assert COL_STAGE_NAME in TEMPLATE_HEADERS
 
 
+def test_error_report_flushes_pending_rows_and_exports_reason(tmp_path, monkeypatch):
+    engine = create_engine("sqlite+pysqlite:///:memory:")
+    Base.metadata.create_all(engine)
+
+    def temporary_data_path(name: str):
+        folder = tmp_path / name
+        folder.mkdir(parents=True, exist_ok=True)
+        return folder
+
+    monkeypatch.setattr(admin_imports, "data_path", temporary_data_path)
+    with Session(engine) as db:
+        batch = ImportBatch(filename="bad.xlsx", file_path="/tmp/bad.xlsx", template_code="practice-camp")
+        db.add(batch)
+        db.flush()
+        db.add(
+            ImportBatchRow(
+                batch_id=batch.id,
+                row_no=2,
+                status="failed",
+                error_message="发证日期格式错误",
+                raw_json=json.dumps(
+                    {
+                        COL_NAME: "张三",
+                        COL_PHONE: "13800000000",
+                        COL_PROJECT: "DBY",
+                        COL_COURSE_START_DATE: "2026-06-01 00:00:00",
+                        COL_ISSUE_DATE: "错误日期",
+                    },
+                    ensure_ascii=False,
+                ),
+            )
+        )
+
+        report_path = write_error_report(db, batch.id)
+        rows = list(load_workbook(report_path, data_only=True).active.iter_rows(values_only=True))
+
+    assert rows[1][0] == 2
+    assert rows[1][1] == "发证日期格式错误"
+    assert rows[1][2] == "张三"
+    assert rows[1][5] == "2026-06-01"
+    assert rows[1][-1] == "错误日期"
+
+
 def test_certificate_create_rejects_reversed_course_period():
     with pytest.raises(ValidationError, match="课程结束日期不能早于课程开始日期"):
         CertificateCreate(