From 3e9cfdfece4b6b3929b8ac9a4240f3d7a72059b5 Mon Sep 17 00:00:00 2001 From: Certificate System Date: Fri, 14 Aug 2026 11:51:53 +0800 Subject: [PATCH] =?UTF-8?q?=E4=BF=AE=E5=A4=8D=E8=AF=81=E4=B9=A6=E5=AF=BC?= =?UTF-8?q?=E5=85=A5=E6=97=A5=E6=9C=9F=E8=A7=A3=E6=9E=90=E4=B8=8E=E9=94=99?= =?UTF-8?q?=E8=AF=AF=E6=8A=A5=E5=91=8A?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- backend/app/api/routes/admin_imports.py | 63 ++++++++++++++++++++-- backend/tests/test_import_rules.py | 69 ++++++++++++++++++++++++- 2 files changed, 127 insertions(+), 5 deletions(-) diff --git a/backend/app/api/routes/admin_imports.py b/backend/app/api/routes/admin_imports.py index 2f5d18e..12aba4e 100644 --- a/backend/app/api/routes/admin_imports.py +++ b/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]: diff --git a/backend/tests/test_import_rules.py b/backend/tests/test_import_rules.py index 0cd1fe4..7b101fe 100644 --- a/backend/tests/test_import_rules.py +++ b/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(