diff --git a/backend/app/api/routes/admin_certificates.py b/backend/app/api/routes/admin_certificates.py index 4d68027..ce3358d 100644 --- a/backend/app/api/routes/admin_certificates.py +++ b/backend/app/api/routes/admin_certificates.py @@ -8,12 +8,11 @@ from app.core.security import generate_public_token, hash_token from app.db.session import get_db from app.models import AdminUser, Certificate, CertificateAccessToken, Learner, ProjectCourse from app.schemas.certificate import CertificateCreate, CertificateOut, PdfPregenerationJobCreate, PdfPregenerationJobOut -from app.services.certificate_number import build_certificate_no +from app.services.certificate_issuance import CertificateIssueData, DuplicateCertificate, issue_certificate from app.services.logs import log_action from app.services.pdf import PdfGenerationBusy, render_certificate_pdf from app.services.pdf_pregeneration import pdf_pregeneration_manager from app.services.system_settings import get_pdf_generation_concurrency_limit -from app.services.certificate_templates import get_certificate_template router = APIRouter() @@ -51,6 +50,7 @@ def list_certificates( Certificate.course_name.like(like), Certificate.stage_name.like(like), Learner.current_name.like(like), + Learner.phone.like(like), ) ) if status_value: @@ -68,38 +68,34 @@ def create_certificate( db: Session = Depends(get_db), admin: AdminUser = Depends(require_roles("system_admin", "certificate_admin")), ) -> Certificate: - learner = db.get(Learner, payload.learner_id) - if not learner: - raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Learner not found") - project = db.query(ProjectCourse).filter(ProjectCourse.code == payload.project_code, ProjectCourse.status == "active").first() - if not project: - raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail="Project code is inactive or missing") - template_code = payload.template_code or project.default_template_code try: - template_code = get_certificate_template(template_code).code + project = db.query(ProjectCourse).filter(ProjectCourse.code == payload.project_code, ProjectCourse.status == "active").first() + if not project: + raise ValueError("项目代码不存在或已停用") + certificate, learner, learner_created = issue_certificate( + db, + CertificateIssueData( + learner_name=payload.learner_name, + learner_phone=payload.learner_phone, + project_code=payload.project_code, + class_name=payload.class_name, + course_name=payload.course_name, + stage_name=payload.stage_name, + course_start_date=payload.course_start_date, + course_end_date=payload.course_end_date, + issue_date=payload.issue_date, + template_code=payload.template_code or project.default_template_code, + remark=payload.remark, + ), + source="manual_certificate", + ) + except DuplicateCertificate as exc: + db.rollback() + raise HTTPException(status_code=status.HTTP_409_CONFLICT, detail=str(exc)) from exc except (ValueError, FileNotFoundError) as exc: + db.rollback() raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail=str(exc)) from exc - - certificate = Certificate( - learner_id=payload.learner_id, - project_code=payload.project_code, - certificate_no="PENDING", - certificate_name=project.name, - class_name=payload.class_name, - course_name=payload.course_name or project.default_course_name or project.name, - stage_name=payload.stage_name or project.default_stage_name, - course_start_date=payload.course_start_date, - course_end_date=payload.course_end_date, - issue_date=payload.issue_date, - template_code=template_code, - remark=payload.remark, - ) - db.add(certificate) - db.flush() - certificate.certificate_no = build_certificate_no(certificate.id, certificate.project_code, certificate.issue_date) - certificate.public_token_id = _create_token(db, certificate.id, "public_link") - certificate.qr_token_id = _create_token(db, certificate.id, "qr_verify") - log_action(db, admin, "create_certificate", "certificate", certificate.id, {"certificate_no": certificate.certificate_no, "learner_name": learner.current_name, "project_code": certificate.project_code, "issue_date": certificate.issue_date}) + log_action(db, admin, "create_certificate", "certificate", certificate.id, {"certificate_no": certificate.certificate_no, "learner_name": learner.current_name, "learner_created": learner_created, "project_code": certificate.project_code, "issue_date": certificate.issue_date}) db.commit() db.refresh(certificate) return certificate @@ -167,6 +163,7 @@ def _certificate_payload(certificate: Certificate, learner: Learner | None = Non "id": certificate.id, "learner_id": certificate.learner_id, "learner_name": learner.current_name if learner else None, + "learner_phone": learner.phone if learner else None, "import_batch_id": certificate.import_batch_id, "project_code": certificate.project_code, "certificate_no": certificate.certificate_no, @@ -240,6 +237,7 @@ def preview_certificate( return { "certificate_no": certificate.certificate_no, "learner_name": learner.current_name if learner else "", + "learner_phone": learner.phone if learner else "", "certificate_name": certificate.certificate_name, "project_code": certificate.project_code, "course_name": certificate.course_name, diff --git a/backend/app/api/routes/admin_imports.py b/backend/app/api/routes/admin_imports.py index 80110b7..2f5d18e 100644 --- a/backend/app/api/routes/admin_imports.py +++ b/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, Form, HTTPException, UploadFile, status +from fastapi import APIRouter, Depends, File, Form, HTTPException, Query, UploadFile, status from fastapi.responses import FileResponse, StreamingResponse from openpyxl import Workbook, load_workbook from openpyxl.comments import Comment @@ -13,36 +13,36 @@ from sqlalchemy.orm import Session from app.api.deps import require_roles from app.core.paths import data_path -from app.core.security import generate_public_token, hash_token from app.db.session import get_db -from app.models import ( - AdminUser, - Certificate, - CertificateAccessToken, - ImportBatch, - ImportBatchRow, - Learner, - LearnerNameHistory, - ProjectCourse, -) +from app.models import AdminUser, ImportBatch, ImportBatchRow, Learner, ProjectCourse from app.schemas.import_batch import ImportBatchOut -from app.services.certificate_number import build_certificate_no +from app.services.certificate_issuance import CertificateIssueData, DuplicateCertificate, issue_certificate +from app.services.certificate_templates import CertificateTemplateDefinition, get_certificate_template +from app.services.learner_identity import normalize_phone from app.services.logs import log_action -from app.services.certificate_templates import get_certificate_template router = APIRouter() -COL_NAME = "\u59d3\u540d" -COL_PHONE = "\u624b\u673a\u53f7" -COL_PROJECT = "\u9879\u76ee\u4ee3\u7801" +COL_NAME = "姓名" +COL_PHONE = "手机号" +COL_PROJECT = "项目代码" +COL_COURSE_NAME = "课程名称" +COL_STAGE_NAME = "阶段名称" COL_COURSE_START_DATE = "课程开始日期" COL_COURSE_END_DATE = "课程结束日期" -COL_ISSUE_DATE = "\u53d1\u8bc1\u65e5\u671f" +COL_ISSUE_DATE = "发证日期" -TEMPLATE_HEADERS = [ - COL_NAME, - COL_PHONE, - COL_PROJECT, +COMMON_HEADERS = [COL_NAME, COL_PHONE, COL_PROJECT] +FIELD_COLUMNS = { + "course_name": COL_COURSE_NAME, + "stage_name": COL_STAGE_NAME, + "course_start_date": COL_COURSE_START_DATE, + "course_end_date": COL_COURSE_END_DATE, + "issue_date": COL_ISSUE_DATE, +} +TEMPLATE_HEADERS = COMMON_HEADERS + [ + COL_COURSE_NAME, + COL_STAGE_NAME, COL_COURSE_START_DATE, COL_COURSE_END_DATE, COL_ISSUE_DATE, @@ -51,25 +51,33 @@ REQUIRED_HEADERS = TEMPLATE_HEADERS @router.get("/template") -def download_template(_: AdminUser = Depends(require_roles("system_admin", "certificate_admin"))) -> StreamingResponse: - workbook = build_import_template_workbook() - stream_path = data_path("exports") / "certificate-import-template.xlsx" +def download_template( + template_code: str = Query(default="classic"), + _: AdminUser = Depends(require_roles("system_admin", "certificate_admin")), +) -> StreamingResponse: + try: + template = get_certificate_template(template_code) + except (ValueError, FileNotFoundError) as exc: + raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail=str(exc)) from exc + workbook = build_import_template_workbook(template.code) + stream_path = data_path("exports") / f"certificate-import-{template.code}.xlsx" workbook.save(stream_path) - file_handle = stream_path.open("rb") return StreamingResponse( - file_handle, + stream_path.open("rb"), media_type="application/vnd.openxmlformats-officedocument.spreadsheetml.sheet", - headers={"Content-Disposition": 'attachment; filename="certificate-import-template.xlsx"'}, + headers={"Content-Disposition": f'attachment; filename="certificate-import-{template.code}.xlsx"'}, ) -def build_import_template_workbook() -> Workbook: +def build_import_template_workbook(template_code: str = "classic") -> Workbook: + template = get_certificate_template(template_code) + headers = template_headers(template) workbook = Workbook() sheet = workbook.active - sheet.title = "\u8bc1\u4e66\u5bfc\u5165\u6a21\u677f" - sheet.append(TEMPLATE_HEADERS) - _format_template_sheet(sheet) - _add_template_instructions(workbook) + sheet.title = "证书导入模板" + sheet.append(headers) + _format_template_sheet(sheet, headers, template) + _add_template_instructions(workbook, template) return workbook @@ -81,21 +89,33 @@ def upload_import_file( 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") + raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail="仅支持 .xlsx 文件") try: - template_code = get_certificate_template(template_code).code + template = get_certificate_template(template_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 + upload_path = data_path("uploads") / f"{datetime.now():%Y%m%d%H%M%S}-{Path(file.filename).name}" with upload_path.open("wb") as target: shutil.copyfileobj(file.file, target) - batch = ImportBatch(filename=file.filename, file_path=str(upload_path), template_code=template_code, 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) - log_action(db, admin, "upload_import_file", "import_batch", batch.id, {"filename": file.filename, "total_rows": batch.total_rows, "valid_rows": batch.valid_rows, "failed_rows": batch.failed_rows, "status": batch.status}) + log_action( + db, + admin, + "upload_import_file", + "import_batch", + batch.id, + {"filename": file.filename, "template_code": template.code, "total_rows": batch.total_rows, "valid_rows": batch.valid_rows, "failed_rows": batch.failed_rows, "status": batch.status}, + ) db.commit() db.refresh(batch) return batch @@ -117,12 +137,8 @@ def download_error_report( ) -> FileResponse: batch = db.get(ImportBatch, batch_id) if not batch or not batch.error_report_path: - raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Error report not found") - return FileResponse( - batch.error_report_path, - media_type="application/vnd.openxmlformats-officedocument.spreadsheetml.sheet", - filename=f"import-errors-{batch.id}.xlsx", - ) + raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="错误报告不存在") + return FileResponse(batch.error_report_path, media_type="application/vnd.openxmlformats-officedocument.spreadsheetml.sheet", filename=f"import-errors-{batch.id}.xlsx") @router.get("/{batch_id}/file") @@ -133,12 +149,8 @@ def download_source_file( ) -> FileResponse: batch = db.get(ImportBatch, batch_id) if not batch or not Path(batch.file_path).exists(): - raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Source file not found") - return FileResponse( - batch.file_path, - media_type="application/vnd.openxmlformats-officedocument.spreadsheetml.sheet", - filename=batch.filename, - ) + raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="原始文件不存在") + return FileResponse(batch.file_path, media_type="application/vnd.openxmlformats-officedocument.spreadsheetml.sheet", filename=batch.filename) @router.delete("/{batch_id}") @@ -149,14 +161,10 @@ def delete_import_batch( ) -> dict[str, bool]: batch = db.get(ImportBatch, batch_id) if not batch: - raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Import batch not found") - + raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="导入批次不存在") for file_name in [batch.file_path, batch.error_report_path]: - if file_name: - path = Path(file_name) - if path.exists(): - path.unlink() - + if file_name and Path(file_name).exists(): + Path(file_name).unlink() db.query(ImportBatchRow).filter(ImportBatchRow.batch_id == batch.id).delete() db.delete(batch) log_action(db, admin, "delete_import_batch", "import_batch", batch.id, {"filename": batch.filename, "status": batch.status, "total_rows": batch.total_rows}) @@ -172,84 +180,73 @@ def confirm_import_batch( ) -> ImportBatch: batch = db.get(ImportBatch, batch_id) if not batch: - raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Import batch not found") - if batch.status not in {"validated", "imported"}: - raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail="Import batch is not ready") + raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="导入批次不存在") if batch.status == "imported": return batch + if batch.status != "validated": + raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail="当前批次不能确认导入") rows = db.query(ImportBatchRow).filter(ImportBatchRow.batch_id == batch.id, ImportBatchRow.status == "valid").all() - ok_rows = 0 - failed_rows = 0 + imported_rows = skipped_rows = failed_rows = created_learners = 0 for row in rows: row_data = json.loads(row.raw_json or "{}") - learner = upsert_learner(db, row_data) - course_start_date = parse_date(row_data[COL_COURSE_START_DATE], COL_COURSE_START_DATE) - course_end_date = parse_date(row_data[COL_COURSE_END_DATE], COL_COURSE_END_DATE) - issue_date = parse_issue_date(row_data[COL_ISSUE_DATE]) - project_code = str(row_data[COL_PROJECT]).strip().upper() - project = db.query(ProjectCourse).filter(ProjectCourse.code == project_code, ProjectCourse.status == "active").first() - if not project: - row.status = "failed" - 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, - batch.template_code, - ) - if duplicate: + try: + _, _, learner_created = issue_certificate( + db, + CertificateIssueData( + learner_name=str(row_data[COL_NAME]), + learner_phone=str(row_data[COL_PHONE]), + project_code=str(row_data[COL_PROJECT]), + template_code=batch.template_code, + course_name=optional_text(row_data.get(COL_COURSE_NAME)), + stage_name=optional_text(row_data.get(COL_STAGE_NAME)), + course_start_date=parse_optional_date(row_data.get(COL_COURSE_START_DATE), COL_COURSE_START_DATE), + course_end_date=parse_optional_date(row_data.get(COL_COURSE_END_DATE), COL_COURSE_END_DATE), + issue_date=parse_issue_date(row_data[COL_ISSUE_DATE]), + import_batch_id=batch.id, + ), + source="certificate_import", + ) + row.status = "imported" + imported_rows += 1 + created_learners += int(learner_created) + except DuplicateCertificate as exc: row.status = "skipped" - row.error_message = "\u5df2\u5b58\u5728\uff0c\u65e0\u9700\u5904\u7406" - ok_rows += 1 - continue + row.error_message = str(exc) + skipped_rows += 1 + except (ValueError, FileNotFoundError) as exc: + row.status = "failed" + row.error_message = str(exc) + failed_rows += 1 - certificate = Certificate( - learner_id=learner.id, - import_batch_id=batch.id, - project_code=project.code, - certificate_no="PENDING", - certificate_name=project.default_certificate_name, - course_name=project.default_course_name, - stage_name=project.default_stage_name, - course_start_date=course_start_date, - course_end_date=course_end_date, - issue_date=issue_date, - template_code=batch.template_code, - remark=None, - ) - db.add(certificate) - db.flush() - certificate.certificate_no = build_certificate_no(certificate.id, certificate.project_code, certificate.issue_date) - certificate.public_token_id = create_access_token(db, certificate.id, "public_link") - certificate.qr_token_id = create_access_token(db, certificate.id, "qr_verify") - row.status = "imported" - ok_rows += 1 - - batch.status = "imported" if ok_rows else "failed" - if failed_rows: - batch.failed_rows = (batch.failed_rows or 0) + failed_rows - log_action(db, admin, "confirm_import_batch", "import_batch", batch.id, {"filename": batch.filename, "valid_rows": batch.valid_rows, "imported_rows": ok_rows, "failed_rows": failed_rows, "status": batch.status}) + batch.status = "imported" if imported_rows or skipped_rows else "failed" + batch.failed_rows = (batch.failed_rows or 0) + failed_rows + log_action( + db, + admin, + "confirm_import_batch", + "import_batch", + batch.id, + {"filename": batch.filename, "imported_rows": imported_rows, "skipped_rows": skipped_rows, "created_learners": created_learners, "failed_rows": failed_rows, "status": batch.status}, + ) db.commit() db.refresh(batch) return batch def validate_batch(db: Session, batch: ImportBatch, upload_path: Path) -> None: + template = get_certificate_template(batch.template_code) + headers = template_headers(template) workbook = load_workbook(upload_path, read_only=True, data_only=True) sheet = workbook.active - header = [cell.value for cell in next(sheet.iter_rows(min_row=1, max_row=1))] + first_row = next(sheet.iter_rows(min_row=1, max_row=1), None) + header = [cell.value for cell in first_row] if first_row else [] header_index = {name: idx for idx, name in enumerate(header)} - missing = [name for name in TEMPLATE_HEADERS if name not in header_index] + missing = [name for name in required_headers(template) if name not in header_index] if missing: batch.status = "failed" batch.failed_rows = 1 - db.add(ImportBatchRow(batch_id=batch.id, row_no=1, status="failed", error_message=f"Missing columns: {missing}")) + db.add(ImportBatchRow(batch_id=batch.id, row_no=1, status="failed", error_message=f"缺少必填列：{'、'.join(missing)}")) return active_codes = {row[0] for row in db.query(ProjectCourse.code).filter(ProjectCourse.status == "active").all()} @@ -258,116 +255,70 @@ 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]] for name in TEMPLATE_HEADERS} - errors = row_errors(row_data, active_codes) - if errors: - failed += 1 - db.add( - ImportBatchRow( - batch_id=batch.id, - row_no=row_no, - status="failed", - error_message="; ".join(errors), - raw_json=json.dumps(row_data, ensure_ascii=False, default=str), - ) + 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)) + valid += int(not errors) + db.add( + ImportBatchRow( + batch_id=batch.id, + row_no=row_no, + status=row_status, + error_message="；".join(errors) if errors else None, + raw_json=json.dumps(row_data, ensure_ascii=False, default=str), ) - else: - valid += 1 - db.add( - ImportBatchRow( - batch_id=batch.id, - row_no=row_no, - status="valid", - raw_json=json.dumps(row_data, ensure_ascii=False, default=str), - ) - ) - + ) batch.total_rows = total batch.valid_rows = valid batch.failed_rows = failed - batch.status = "validated" + batch.status = "validated" if total else "failed" if failed: batch.error_report_path = str(write_error_report(db, batch.id)) -def row_errors(row_data: dict[str, object], active_codes: set[str]) -> list[str]: - errors = [] - for name in REQUIRED_HEADERS: - if not row_data.get(name): - errors.append(f"{name} is required") +def row_errors( + row_data: dict[str, object], + active_codes: set[str], + template_code: str = "classic", + db: Session | None = None, +) -> list[str]: + template = get_certificate_template(template_code) + errors: list[str] = [] + for name in required_headers(template): + if not optional_text(row_data.get(name)): + errors.append(f"{name}不能为空") project_code = str(row_data.get(COL_PROJECT) or "").strip().upper() if project_code and project_code not in active_codes: - errors.append("Project code is inactive or missing") - for column in [COL_COURSE_START_DATE, COL_COURSE_END_DATE, COL_ISSUE_DATE]: + errors.append("项目代码不存在或已停用") + for field in template.fields: + if field.field_type != "date" or field.key == "learner_name": + continue + column = FIELD_COLUMNS[field.key] if row_data.get(column) and not date_is_valid(row_data[column]): - errors.append(f"{column}格式错误，请使用YYYY-MM-DD，例如2026-06-01") + errors.append(f"{column}格式错误，请使用YYYY-MM-DD，例如{field.example}") if all(row_data.get(column) and date_is_valid(row_data[column]) for column in [COL_COURSE_START_DATE, COL_COURSE_END_DATE]): - start_date = parse_date(row_data[COL_COURSE_START_DATE], COL_COURSE_START_DATE) - end_date = parse_date(row_data[COL_COURSE_END_DATE], COL_COURSE_END_DATE) - if end_date < start_date: + if parse_date(row_data[COL_COURSE_END_DATE], COL_COURSE_END_DATE) < parse_date(row_data[COL_COURSE_START_DATE], COL_COURSE_START_DATE): errors.append("课程结束日期不能早于课程开始日期") + if row_data.get(COL_PHONE): + try: + phone = normalize_phone(str(row_data[COL_PHONE])) + if db: + learner = db.query(Learner).filter(Learner.phone == phone, Learner.status != "deleted").first() + input_name = str(row_data.get(COL_NAME) or "").strip() + if learner and learner.current_name.strip() != input_name: + errors.append(f"手机号已属于学员“{learner.current_name}”，姓名不一致") + except ValueError as exc: + errors.append(str(exc)) return errors -def upsert_learner(db: Session, row_data: dict[str, object]) -> Learner: - phone = str(row_data[COL_PHONE]).strip() - name = str(row_data[COL_NAME]).strip() - learner = db.query(Learner).filter(Learner.phone == phone).first() - if learner: - if learner.current_name != name: - db.add(LearnerNameHistory(learner_id=learner.id, name=name, source="import")) - learner.current_name = name - return learner - - learner = Learner(phone=phone, current_name=name) - db.add(learner) - db.flush() - db.add(LearnerNameHistory(learner_id=learner.id, name=name, source="import")) - return learner +def template_headers(template: CertificateTemplateDefinition) -> list[str]: + return COMMON_HEADERS + [FIELD_COLUMNS[field.key] for field in template.fields if field.key != "learner_name"] -def get_active_project(db: Session, project_code: str) -> ProjectCourse: - project = db.query(ProjectCourse).filter(ProjectCourse.code == project_code, ProjectCourse.status == "active").first() - if not project: - raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail=f"Project code is inactive or missing: {project_code}") - return project - - -def find_duplicate_certificate( - db: Session, - learner_id: int, - project: ProjectCourse, - course_start_date: date, - course_end_date: date, - issue_date: date, - template_code: str, -) -> Certificate | None: - return ( - db.query(Certificate) - .filter(Certificate.learner_id == learner_id) - .filter(Certificate.project_code == project.code) - .filter(Certificate.certificate_name == project.default_certificate_name) - .filter(Certificate.course_name == project.default_course_name) - .filter(Certificate.stage_name == project.default_stage_name) - .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() - ) - - -def create_access_token(db: Session, certificate_id: int, token_type: str) -> int: - raw_token = generate_public_token() - access_token = CertificateAccessToken( - certificate_id=certificate_id, - token_hash=hash_token(raw_token), - token_value=raw_token, - token_type=token_type, - ) - db.add(access_token) - db.flush() - return access_token.id +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 optional_text(value: object) -> str | None: @@ -381,6 +332,10 @@ def parse_issue_date(value: object) -> date: return parse_date(value, COL_ISSUE_DATE) +def parse_optional_date(value: object, field_name: str) -> date | None: + return parse_date(value, field_name) if value not in (None, "") else None + + def parse_date(value: object, field_name: str = "日期") -> date: if isinstance(value, datetime): return value.date() @@ -392,22 +347,22 @@ def parse_date(value: object, field_name: str = "日期") -> date: return datetime.strptime(text, fmt).date() except ValueError: continue - raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail=f"{field_name}格式错误，请使用YYYY-MM-DD：{text}") + raise ValueError(f"{field_name}格式错误，请使用YYYY-MM-DD：{text}") def date_is_valid(value: object) -> bool: try: - parse_issue_date(value) + parse_date(value) return True - except HTTPException: + except ValueError: return False def write_error_report(db: Session, batch_id: int) -> Path: workbook = Workbook() sheet = workbook.active - sheet.title = "\u9519\u8bef\u62a5\u544a" - sheet.append(["\u884c\u53f7", "\u9519\u8bef\u539f\u56e0", "\u539f\u59cb\u6570\u636e"]) + sheet.title = "错误报告" + sheet.append(["行号", "错误原因", "原始数据"]) 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]) @@ -416,60 +371,61 @@ def write_error_report(db: Session, batch_id: int) -> Path: return report_path -def _format_template_sheet(sheet) -> None: +def _format_template_sheet(sheet, headers: list[str], template: CertificateTemplateDefinition) -> None: header_fill = PatternFill("solid", fgColor="208A87") for cell in sheet[1]: cell.fill = header_fill cell.font = Font(color="FFFFFF", bold=True) cell.alignment = Alignment(horizontal="center", vertical="center") sheet.freeze_panes = "A2" - sheet.auto_filter.ref = f"A1:{sheet.cell(1, len(TEMPLATE_HEADERS)).coordinate}" - widths = [16, 18, 16, 18, 18, 18] - for index, width in enumerate(widths, start=1): - sheet.column_dimensions[sheet.cell(1, index).column_letter].width = width + sheet.auto_filter.ref = f"A1:{sheet.cell(1, len(headers)).coordinate}" + widths = {COL_NAME: 16, COL_PHONE: 18, COL_PROJECT: 16, COL_COURSE_NAME: 28, COL_STAGE_NAME: 18, COL_COURSE_START_DATE: 18, COL_COURSE_END_DATE: 18, COL_ISSUE_DATE: 18} sheet.column_dimensions["B"].number_format = "@" - for column_index in [4, 5, 6]: - column_letter = sheet.cell(1, column_index).column_letter - sheet.column_dimensions[column_letter].number_format = "yyyy-mm-dd" - validation = DataValidation(type="date", operator="between", formula1="DATE(2000,1,1)", formula2="DATE(2100,12,31)", allow_blank=False) - validation.promptTitle = "日期格式" - validation.prompt = "请按 YYYY-MM-DD 填写，例如 2026-06-01" - validation.errorTitle = "日期格式错误" - validation.error = "请填写 2000-01-01 至 2100-12-31 之间的有效日期" - validation.errorStyle = "stop" - validation.showInputMessage = True - validation.showErrorMessage = True - sheet.add_data_validation(validation) - validation.add(f"{column_letter}2:{column_letter}5000") - sheet.cell(1, column_index).comment = Comment("必填。请使用 YYYY-MM-DD 格式，例如 2026-06-01。", "证书管理系统") + for index, header in enumerate(headers, start=1): + column_letter = sheet.cell(1, index).column_letter + sheet.column_dimensions[column_letter].width = widths[header] + field = next((item for item in template.fields if item.label == header), None) + required = header in required_headers(template) + sheet.cell(1, index).comment = Comment(f"{'必填' if required else '选填'}。{field.description if field else '用于识别和归档数据'}", "证书管理系统") + if field and field.field_type == "date": + sheet.column_dimensions[column_letter].number_format = "yyyy-mm-dd" + validation = DataValidation(type="date", operator="between", formula1="DATE(2000,1,1)", formula2="DATE(2100,12,31)", allow_blank=not field.required) + validation.promptTitle = "日期格式" + validation.prompt = f"请按 YYYY-MM-DD 填写，例如 {field.example}" + validation.errorTitle = "日期格式错误" + validation.error = "请填写有效日期" + validation.errorStyle = "stop" + validation.showInputMessage = True + validation.showErrorMessage = True + sheet.add_data_validation(validation) + validation.add(f"{column_letter}2:{column_letter}5000") -def _add_template_instructions(workbook: Workbook) -> None: +def _add_template_instructions(workbook: Workbook, template: CertificateTemplateDefinition) -> None: sheet = workbook.create_sheet("填写说明") sheet.append(["字段", "是否必填", "格式或示例", "填写说明"]) - rows = [ - (COL_NAME, "是", "张三", "填写学员真实姓名"), - (COL_PHONE, "是", "13800000000", "建议将单元格设为文本，避免手机号格式变化"), - (COL_PROJECT, "是", "DBY", "填写系统中已启用的项目代码"), - (COL_COURSE_START_DATE, "是", "2026-06-01", "课程实际开始日期，必须使用 YYYY-MM-DD"), - (COL_COURSE_END_DATE, "是", "2026-06-30", "不得早于课程开始日期，必须使用 YYYY-MM-DD"), - (COL_ISSUE_DATE, "是", "2026-07-05", "证书签发日期，必须使用 YYYY-MM-DD"), + instructions = [ + (COL_NAME, True, "张三", "填写学员真实姓名；与手机号共同确认学员身份"), + (COL_PHONE, True, "13800000000", "不存在时自动创建学员；已存在时姓名必须一致"), + (COL_PROJECT, True, "DBY", "填写系统中已启用的项目代码"), ] - for row in rows: - sheet.append(row) - sheet.append([]) - sheet.append(["重要提示", "请在“证书导入模板”工作表填写正式数据，不要修改第一行列名。日期统一填写为 YYYY-MM-DD，例如 2026-06-01。"]) - sheet.merge_cells(start_row=9, start_column=2, end_row=9, end_column=4) + instructions.extend((field.label, field.required, field.example, field.description) for field in template.fields if field.key != "learner_name") + for label, required, example, description in instructions: + sheet.append([label, "是" if required else "否", example, description]) + note_row = len(instructions) + 3 + sheet.cell(note_row, 1, "重要提示") + sheet.cell(note_row, 2, f"本文件仅适用于“{template.name}”。不要修改第一行列名；日期统一使用 YYYY-MM-DD。") + sheet.merge_cells(start_row=note_row, start_column=2, end_row=note_row, end_column=4) for cell in sheet[1]: cell.fill = PatternFill("solid", fgColor="208A87") cell.font = Font(color="FFFFFF", bold=True) cell.alignment = Alignment(horizontal="center") - sheet["A9"].font = Font(color="C00000", bold=True) - sheet["B9"].font = Font(color="C00000", bold=True) - sheet["B9"].alignment = Alignment(wrap_text=True, vertical="center") - sheet.row_dimensions[9].height = 34 + sheet.cell(note_row, 1).font = Font(color="C00000", bold=True) + sheet.cell(note_row, 2).font = Font(color="C00000", bold=True) + sheet.cell(note_row, 2).alignment = Alignment(wrap_text=True, vertical="center") + sheet.row_dimensions[note_row].height = 34 sheet.column_dimensions["A"].width = 20 sheet.column_dimensions["B"].width = 16 - sheet.column_dimensions["C"].width = 22 - sheet.column_dimensions["D"].width = 54 + sheet.column_dimensions["C"].width = 24 + sheet.column_dimensions["D"].width = 58 sheet.freeze_panes = "A2" diff --git a/backend/app/api/routes/admin_learners.py b/backend/app/api/routes/admin_learners.py index 0b17ed0..99fd361 100644 --- a/backend/app/api/routes/admin_learners.py +++ b/backend/app/api/routes/admin_learners.py @@ -1,27 +1,47 @@ -from fastapi import APIRouter, Depends, HTTPException, Query, status +from io import BytesIO + +from fastapi import APIRouter, Depends, File, HTTPException, Query, UploadFile, status +from fastapi.responses import StreamingResponse +from openpyxl import Workbook, load_workbook +from openpyxl.comments import Comment +from openpyxl.styles import Alignment, Font, PatternFill +from sqlalchemy import func from sqlalchemy import or_ from sqlalchemy.orm import Session from app.api.deps import require_roles from app.db.session import get_db -from app.models import AdminUser, Learner, LearnerNameHistory -from app.schemas.learner import LearnerCreate, LearnerOut, LearnerUpdate +from app.models import AdminUser, Certificate, Learner, LearnerNameHistory +from app.schemas.learner import LearnerCreate, LearnerImportResult, LearnerOut, LearnerUpdate +from app.services.learner_identity import LearnerIdentityConflict, normalize_name, normalize_phone, resolve_learner from app.services.logs import diff_values, log_action, mask_phone router = APIRouter() +LEARNER_IMPORT_HEADERS = ["姓名", "手机号", "学员编号", "备注"] + @router.get("", response_model=list[LearnerOut]) def list_learners( keyword: str | None = Query(default=None), db: Session = Depends(get_db), _: AdminUser = Depends(require_roles("system_admin", "certificate_admin", "readonly")), -) -> list[Learner]: - query = db.query(Learner).filter(Learner.status != "deleted").order_by(Learner.id.desc()) +) -> list[dict[str, object]]: + count_subquery = ( + db.query(Certificate.learner_id.label("learner_id"), func.count(Certificate.id).label("certificate_count")) + .group_by(Certificate.learner_id) + .subquery() + ) + query = ( + db.query(Learner, func.coalesce(count_subquery.c.certificate_count, 0)) + .outerjoin(count_subquery, count_subquery.c.learner_id == Learner.id) + .filter(Learner.status != "deleted") + .order_by(Learner.id.desc()) + ) if keyword: like = f"%{keyword}%" query = query.filter(or_(Learner.current_name.like(like), Learner.phone.like(like), Learner.student_no.like(like))) - return query.limit(100).all() + return [{**LearnerOut.model_validate(learner).model_dump(), "certificate_count": certificate_count} for learner, certificate_count in query.limit(100).all()] @router.post("", response_model=LearnerOut, status_code=status.HTTP_201_CREATED) @@ -30,32 +50,132 @@ def create_learner( db: Session = Depends(get_db), admin: AdminUser = Depends(require_roles("system_admin", "certificate_admin")), ) -> Learner: - existing = db.query(Learner).filter(Learner.phone == payload.phone).first() - if existing: - if existing.current_name != payload.current_name: - db.add(LearnerNameHistory(learner_id=existing.id, name=payload.current_name, source="manual")) - existing.current_name = payload.current_name - existing.student_no = payload.student_no - existing.remark = payload.remark - log_action(db, admin, "update_learner", "learner", existing.id, {"name": existing.current_name, "phone": mask_phone(existing.phone)}) - db.commit() - db.refresh(existing) - return existing - learner = Learner( - phone=payload.phone, - current_name=payload.current_name, - student_no=payload.student_no, - remark=payload.remark, - ) - db.add(learner) + try: + learner, created = resolve_learner(db, payload.current_name, payload.phone, source="manual", create_if_missing=True) + except (ValueError, LearnerIdentityConflict) as exc: + raise HTTPException(status_code=status.HTTP_409_CONFLICT, detail=str(exc)) from exc + assert learner is not None + learner.student_no = payload.student_no + learner.remark = payload.remark + action = "create_learner" if created else "update_learner" + log_action(db, admin, action, "learner", learner.id, {"name": learner.current_name, "phone": mask_phone(learner.phone), "status": learner.status}) db.commit() db.refresh(learner) - db.add(LearnerNameHistory(learner_id=learner.id, name=learner.current_name, source="manual")) - log_action(db, admin, "create_learner", "learner", learner.id, {"name": learner.current_name, "phone": mask_phone(learner.phone), "status": learner.status}) - db.commit() return learner +@router.get("/import-template") +def download_learner_import_template( + _: AdminUser = Depends(require_roles("system_admin", "certificate_admin")), +) -> StreamingResponse: + workbook = build_learner_import_template() + stream = BytesIO() + workbook.save(stream) + stream.seek(0) + return StreamingResponse( + stream, + media_type="application/vnd.openxmlformats-officedocument.spreadsheetml.sheet", + headers={"Content-Disposition": 'attachment; filename="learner-import-template.xlsx"'}, + ) + + +@router.post("/import", response_model=LearnerImportResult) +def import_learners( + file: UploadFile = File(...), + db: Session = Depends(get_db), + admin: AdminUser = Depends(require_roles("system_admin", "certificate_admin")), +) -> dict[str, object]: + if not file.filename or not file.filename.lower().endswith(".xlsx"): + raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail="仅支持 .xlsx 文件") + try: + workbook = load_workbook(file.file, read_only=True, data_only=True) + except Exception as exc: + raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail="Excel 文件无法读取") from exc + sheet = workbook.active + first_row = next(sheet.iter_rows(min_row=1, max_row=1), None) + headers = [cell.value for cell in first_row] if first_row else [] + header_index = {name: idx for idx, name in enumerate(headers)} + missing = [name for name in LEARNER_IMPORT_HEADERS[:2] if name not in header_index] + if missing: + raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail=f"缺少必填列：{'、'.join(missing)}") + + result: dict[str, object] = {"total_rows": 0, "created_rows": 0, "updated_rows": 0, "unchanged_rows": 0, "failed_rows": 0, "errors": []} + seen_phones: dict[str, str] = {} + for row_no, row in enumerate(sheet.iter_rows(min_row=2, values_only=True), start=2): + if not any(row): + continue + result["total_rows"] += 1 + values = {name: row[index] if index < len(row) else None for name, index in header_index.items()} + try: + name = normalize_name(str(values.get("姓名") or "")) + phone = normalize_phone(str(values.get("手机号") or "")) + if not name: + raise ValueError("姓名不能为空") + if phone in seen_phones and seen_phones[phone] != name: + raise LearnerIdentityConflict(f"文件内手机号重复且姓名不一致：{seen_phones[phone]} / {name}") + seen_phones[phone] = name + learner, created = resolve_learner(db, name, phone, source="learner_import", create_if_missing=True) + assert learner is not None + changed = False + student_no = _optional_cell(values.get("学员编号")) + remark = _optional_cell(values.get("备注")) + if student_no is not None and learner.student_no != student_no: + learner.student_no = student_no + changed = True + if remark is not None and learner.remark != remark: + learner.remark = remark + changed = True + if created: + result["created_rows"] += 1 + elif changed: + result["updated_rows"] += 1 + else: + result["unchanged_rows"] += 1 + except (ValueError, LearnerIdentityConflict) as exc: + result["failed_rows"] += 1 + result["errors"].append({"row_no": row_no, "message": str(exc)}) + + log_action(db, admin, "import_learners", "learner", detail={key: value for key, value in result.items() if key != "errors"}) + db.commit() + return result + + +def build_learner_import_template() -> Workbook: + workbook = Workbook() + sheet = workbook.active + sheet.title = "学员导入模板" + sheet.append(LEARNER_IMPORT_HEADERS) + examples = { + "姓名": ("张三", "必填。填写学员真实姓名"), + "手机号": ("13800000000", "必填。11位手机号；系统中已存在时姓名必须一致"), + "学员编号": ("HY20260001", "选填。内部管理编号"), + "备注": ("2026年大本营学员", "选填。内部备注"), + } + for index, header in enumerate(LEARNER_IMPORT_HEADERS, start=1): + cell = sheet.cell(1, index) + cell.fill = PatternFill("solid", fgColor="208A87") + cell.font = Font(color="FFFFFF", bold=True) + cell.alignment = Alignment(horizontal="center", vertical="center") + cell.comment = Comment(examples[header][1], "证书管理系统") + sheet.column_dimensions[cell.column_letter].width = 18 if header != "备注" else 32 + sheet.column_dimensions["B"].number_format = "@" + instruction = workbook.create_sheet("填写说明") + instruction.append(["字段", "是否必填", "示例", "填写说明"]) + for header in LEARNER_IMPORT_HEADERS: + instruction.append([header, "是" if header in LEARNER_IMPORT_HEADERS[:2] else "否", examples[header][0], examples[header][1]]) + instruction.column_dimensions["A"].width = 18 + instruction.column_dimensions["B"].width = 14 + instruction.column_dimensions["C"].width = 22 + instruction.column_dimensions["D"].width = 52 + sheet.freeze_panes = "A2" + return workbook + + +def _optional_cell(value: object) -> str | None: + text = str(value).strip() if value is not None else "" + return text or None + + @router.get("/{learner_id}", response_model=LearnerOut) def get_learner( learner_id: int, @@ -78,7 +198,12 @@ def update_learner( learner = db.get(Learner, learner_id) if not learner: raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Learner not found") - duplicate = db.query(Learner).filter(Learner.phone == payload.phone, Learner.id != learner_id).first() + try: + normalized_phone = normalize_phone(payload.phone) + normalized_name = normalize_name(payload.current_name) + except ValueError as exc: + raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail=str(exc)) from exc + duplicate = db.query(Learner).filter(Learner.phone == normalized_phone, Learner.id != learner_id).first() if duplicate: raise HTTPException(status_code=status.HTTP_409_CONFLICT, detail="Phone already exists") before = { @@ -88,10 +213,10 @@ def update_learner( "status": learner.status, "remark": learner.remark, } - if learner.current_name != payload.current_name: - db.add(LearnerNameHistory(learner_id=learner.id, name=payload.current_name, source="manual")) - learner.phone = payload.phone - learner.current_name = payload.current_name + if learner.current_name != normalized_name: + db.add(LearnerNameHistory(learner_id=learner.id, name=normalized_name, source="manual")) + learner.phone = normalized_phone + learner.current_name = normalized_name learner.student_no = payload.student_no learner.status = payload.status learner.remark = payload.remark diff --git a/backend/app/schemas/certificate.py b/backend/app/schemas/certificate.py index c0750ab..ae619d0 100644 --- a/backend/app/schemas/certificate.py +++ b/backend/app/schemas/certificate.py @@ -4,7 +4,8 @@ from pydantic import BaseModel, Field, ValidationInfo, field_validator class CertificateCreate(BaseModel): - learner_id: int + learner_name: str = Field(min_length=1, max_length=64) + learner_phone: str = Field(min_length=6, max_length=32) project_code: str = Field(min_length=2, max_length=16) certificate_name: str | None = Field(default=None, max_length=128) class_name: str | None = Field(default=None, max_length=128) @@ -21,6 +22,11 @@ class CertificateCreate(BaseModel): def normalize_project_code(cls, value: str) -> str: return value.strip().upper() + @field_validator("learner_name", "learner_phone") + @classmethod + def strip_identity(cls, value: str) -> str: + return value.strip() + @field_validator("course_end_date") @classmethod def validate_course_period(cls, value: date, info: ValidationInfo) -> date: @@ -34,6 +40,7 @@ class CertificateOut(BaseModel): id: int learner_id: int learner_name: str | None = None + learner_phone: str | None = None import_batch_id: int | None = None project_code: str certificate_no: str diff --git a/backend/app/schemas/certificate_template.py b/backend/app/schemas/certificate_template.py index 2ac3aa0..48e2877 100644 --- a/backend/app/schemas/certificate_template.py +++ b/backend/app/schemas/certificate_template.py @@ -1,10 +1,20 @@ from pydantic import BaseModel +class CertificateTemplateFieldOut(BaseModel): + key: str + label: str + field_type: str + required: bool + example: str + description: str + + class CertificateTemplateOut(BaseModel): code: str name: str description: str dynamic_fields: list[str] + fields: list[CertificateTemplateFieldOut] preview_url: str status: str diff --git a/backend/app/schemas/learner.py b/backend/app/schemas/learner.py index 85721f3..430de5f 100644 --- a/backend/app/schemas/learner.py +++ b/backend/app/schemas/learner.py @@ -25,7 +25,22 @@ class LearnerOut(BaseModel): student_no: str | None status: str remark: str | None + certificate_count: int = 0 created_at: datetime updated_at: datetime model_config = {"from_attributes": True} + + +class LearnerImportError(BaseModel): + row_no: int + message: str + + +class LearnerImportResult(BaseModel): + total_rows: int + created_rows: int + updated_rows: int + unchanged_rows: int + failed_rows: int + errors: list[LearnerImportError] diff --git a/backend/app/services/certificate_issuance.py b/backend/app/services/certificate_issuance.py new file mode 100644 index 0000000..d1c1360 --- /dev/null +++ b/backend/app/services/certificate_issuance.py @@ -0,0 +1,135 @@ +from dataclasses import dataclass +from datetime import date + +from sqlalchemy.orm import Session + +from app.core.security import generate_public_token, hash_token +from app.models import Certificate, CertificateAccessToken, Learner, ProjectCourse +from app.services.certificate_number import build_certificate_no +from app.services.certificate_templates import certificate_field_keys, get_certificate_template +from app.services.learner_identity import resolve_learner + + +class DuplicateCertificate(ValueError): + def __init__(self, certificate: Certificate): + self.certificate = certificate + super().__init__(f"已存在相同证书：{certificate.certificate_no}") + + +@dataclass(frozen=True) +class CertificateIssueData: + learner_name: str + learner_phone: str + project_code: str + template_code: str + issue_date: date + course_start_date: date | None = None + course_end_date: date | None = None + course_name: str | None = None + stage_name: str | None = None + class_name: str | None = None + remark: str | None = None + import_batch_id: int | None = None + + +def issue_certificate(db: Session, data: CertificateIssueData, *, source: str) -> tuple[Certificate, Learner, bool]: + template = get_certificate_template(data.template_code) + project_code = data.project_code.strip().upper() + project = db.query(ProjectCourse).filter(ProjectCourse.code == project_code, ProjectCourse.status == "active").first() + if not project: + raise ValueError(f"项目代码不存在或已停用：{project_code}") + + values = _template_values(data, project) + _validate_template_values(template, values) + learner, learner_created = resolve_learner( + db, + data.learner_name, + data.learner_phone, + source=source, + create_if_missing=True, + ) + assert learner is not None + duplicate = find_duplicate_certificate(db, learner.id, project_code, template.code, values) + if duplicate: + raise DuplicateCertificate(duplicate) + + certificate = Certificate( + learner_id=learner.id, + import_batch_id=data.import_batch_id, + project_code=project_code, + certificate_no="PENDING", + certificate_name=project.default_certificate_name or project.name, + class_name=data.class_name, + course_name=values["course_name"], + stage_name=values["stage_name"], + course_start_date=values["course_start_date"], + course_end_date=values["course_end_date"], + issue_date=data.issue_date, + template_code=template.code, + remark=data.remark, + ) + db.add(certificate) + db.flush() + certificate.certificate_no = build_certificate_no(certificate.id, certificate.project_code, certificate.issue_date) + certificate.public_token_id = _create_access_token(db, certificate.id, "public_link") + certificate.qr_token_id = _create_access_token(db, certificate.id, "qr_verify") + return certificate, learner, learner_created + + +def find_duplicate_certificate( + db: Session, + learner_id: int, + project_code: str, + template_code: str, + values: dict[str, object], +) -> Certificate | None: + return ( + db.query(Certificate) + .filter(Certificate.learner_id == learner_id) + .filter(Certificate.project_code == project_code) + .filter(Certificate.template_code == template_code) + .filter(Certificate.course_name == values["course_name"]) + .filter(Certificate.stage_name == values["stage_name"]) + .filter(Certificate.course_start_date == values["course_start_date"]) + .filter(Certificate.course_end_date == values["course_end_date"]) + .filter(Certificate.issue_date == values["issue_date"]) + .filter(Certificate.status == "valid") + .first() + ) + + +def _template_values(data: CertificateIssueData, project: ProjectCourse) -> dict[str, object]: + template = get_certificate_template(data.template_code) + keys = certificate_field_keys(template) + return { + "course_name": (data.course_name or project.default_course_name) if "course_name" in keys else None, + "stage_name": (data.stage_name or project.default_stage_name) if "stage_name" in keys else None, + "course_start_date": data.course_start_date if "course_start_date" in keys else None, + "course_end_date": data.course_end_date if "course_end_date" in keys else None, + "issue_date": data.issue_date, + } + + +def _validate_template_values(template, values: dict[str, object]) -> None: + for field in template.fields: + if field.key == "learner_name": + continue + if field.required and not values.get(field.key): + raise ValueError(f"{template.name}需要填写{field.label}") + start_date = values.get("course_start_date") + end_date = values.get("course_end_date") + if start_date and end_date and end_date < start_date: + raise ValueError("课程结束日期不能早于课程开始日期") + + +def _create_access_token(db: Session, certificate_id: int, token_type: str) -> int: + raw_token = generate_public_token() + token = CertificateAccessToken( + certificate_id=certificate_id, + token_hash=hash_token(raw_token), + token_value=raw_token, + token_type=token_type, + ) + db.add(token) + db.flush() + return token.id diff --git a/backend/app/services/certificate_templates.py b/backend/app/services/certificate_templates.py index 6ec47a2..6f32384 100644 --- a/backend/app/services/certificate_templates.py +++ b/backend/app/services/certificate_templates.py @@ -2,15 +2,39 @@ from dataclasses import dataclass from pathlib import Path +@dataclass(frozen=True) +class CertificateFieldDefinition: + key: str + label: str + field_type: str + required: bool + example: str + description: str + + def to_dict(self) -> dict[str, object]: + return { + "key": self.key, + "label": self.label, + "field_type": self.field_type, + "required": self.required, + "example": self.example, + "description": self.description, + } + + @dataclass(frozen=True) class CertificateTemplateDefinition: code: str name: str description: str asset_filename: str - dynamic_fields: tuple[str, ...] + fields: tuple[CertificateFieldDefinition, ...] status: str = "active" + @property + def dynamic_fields(self) -> tuple[str, ...]: + return tuple(field.label for field in self.fields) + @property def asset_path(self) -> Path: return Path(__file__).resolve().parents[1] / "assets" / self.asset_filename @@ -20,7 +44,8 @@ class CertificateTemplateDefinition: "code": self.code, "name": self.name, "description": self.description, - "dynamic_fields": list(self.dynamic_fields), + "dynamic_fields": [field.label for field in self.fields], + "fields": [field.to_dict() for field in self.fields], "preview_url": f"/api/admin/certificate-templates/{self.code}/preview", "status": self.status, } @@ -35,14 +60,26 @@ CERTIFICATE_TEMPLATES = { name="经典结业证书", description="通用课程结业证书，正文包含课程名称、阶段、课程时间和发证日期。", asset_filename="certificate-template.png", - dynamic_fields=("姓名", "课程名称", "阶段名称", "课程开始日期", "课程结束日期", "发证日期"), + fields=( + CertificateFieldDefinition("learner_name", "姓名", "text", True, "张三", "证书持有人真实姓名"), + CertificateFieldDefinition("course_name", "课程名称", "text", True, "智慧赋能疗愈师", "证书正文中的课程名称"), + CertificateFieldDefinition("stage_name", "阶段名称", "text", True, "初级", "证书正文中的课程阶段"), + CertificateFieldDefinition("course_start_date", "课程开始日期", "date", True, "2026-06-01", "课程实际开始日期"), + CertificateFieldDefinition("course_end_date", "课程结束日期", "date", True, "2026-06-30", "课程实际结束日期"), + CertificateFieldDefinition("issue_date", "发证日期", "date", True, "2026-07-05", "证书签发日期"), + ), ), PRACTICE_CAMP_TEMPLATE_CODE: CertificateTemplateDefinition( code=PRACTICE_CAMP_TEMPLATE_CODE, name="实修大本营结业证书", description="人本智慧五个月线上实修大本营专用版，只填写姓名、课程开始日期、课程结束日期和发证日期。", asset_filename="certificate-template-practice-camp.png", - dynamic_fields=("姓名", "课程开始日期", "课程结束日期", "发证日期"), + fields=( + CertificateFieldDefinition("learner_name", "姓名", "text", True, "张三", "证书持有人真实姓名"), + CertificateFieldDefinition("course_start_date", "课程开始日期", "date", True, "2026-03-01", "课程实际开始日期"), + CertificateFieldDefinition("course_end_date", "课程结束日期", "date", True, "2026-08-31", "课程实际结束日期"), + CertificateFieldDefinition("issue_date", "发证日期", "date", True, "2026-09-05", "证书签发日期"), + ), ), } @@ -59,3 +96,7 @@ def get_certificate_template(code: str | None) -> CertificateTemplateDefinition: if not template.asset_path.exists(): raise FileNotFoundError(f"证书模板图片不存在：{template.asset_filename}") return template + + +def certificate_field_keys(template: CertificateTemplateDefinition) -> set[str]: + return {field.key for field in template.fields} diff --git a/backend/app/services/learner_identity.py b/backend/app/services/learner_identity.py new file mode 100644 index 0000000..6a5ff4e --- /dev/null +++ b/backend/app/services/learner_identity.py @@ -0,0 +1,57 @@ +import re + +from sqlalchemy.orm import Session + +from app.models import Learner, LearnerNameHistory + + +class LearnerIdentityConflict(ValueError): + pass + + +def normalize_name(value: str) -> str: + return " ".join(value.strip().split()) + + +def normalize_phone(value: str) -> str: + phone = re.sub(r"[\s-]", "", value.strip()) + if phone.startswith("+86"): + phone = phone[3:] + elif phone.startswith("0086"): + phone = phone[4:] + if not re.fullmatch(r"1\d{10}", phone): + raise ValueError("手机号格式错误，请填写11位中国大陆手机号") + return phone + + +def resolve_learner( + db: Session, + name: str, + phone: str, + *, + source: str, + create_if_missing: bool = True, +) -> tuple[Learner | None, bool]: + normalized_name = normalize_name(name) + normalized_phone = normalize_phone(phone) + if not normalized_name: + raise ValueError("姓名不能为空") + + learner = db.query(Learner).filter(Learner.phone == normalized_phone, Learner.status != "deleted").first() + if learner: + if normalize_name(learner.current_name) != normalized_name: + raise LearnerIdentityConflict( + f"手机号 {normalized_phone} 已属于学员“{learner.current_name}”，与本次姓名“{normalized_name}”不一致" + ) + if learner.status != "active": + raise LearnerIdentityConflict(f"学员“{normalized_name}”当前已停用，请先在学员管理中启用") + return learner, False + + if not create_if_missing: + return None, False + + learner = Learner(phone=normalized_phone, current_name=normalized_name) + db.add(learner) + db.flush() + db.add(LearnerNameHistory(learner_id=learner.id, name=normalized_name, source=source)) + return learner, True diff --git a/backend/docs/learner-certificate-domain.md b/backend/docs/learner-certificate-domain.md new file mode 100644 index 0000000..80dfa1a --- /dev/null +++ b/backend/docs/learner-certificate-domain.md @@ -0,0 +1,62 @@ +# 学员与证书业务模型 + +## 1. 核心关系 + +- 学员是长期主体，证书是某一次学习成果的签发记录。 +- 一个学员可以拥有多张证书，证书可以来自不同项目、不同课程、不同模板和不同时间段。 +- 证书签发后保存当次课程、阶段、日期和模板代码，不依赖项目默认值继续变化。 +- 项目只负责归档和提供默认值，不能代替证书模板定义字段。 + +## 2. 学员身份规则 + +- 系统使用标准化后的手机号定位学员，姓名用于一致性校验和业务确认。 +- 输入手机号会去除空格、短横线及中国大陆区号，最终保存为 11 位手机号。 +- 手机号不存在时，手工签发和证书批量导入都会自动创建学员。 +- 手机号已存在且姓名一致时，复用原学员。 +- 手机号已存在但姓名不一致时，必须报冲突，不能自动改名或把证书挂到错误学员。 +- 确需更名时，应在学员管理中人工修改，并保留姓名历史。 + +## 3. 模板字段规则 + +`app/services/certificate_templates.py` 是模板字段的唯一规则源。每个字段定义： + +- `key`：后端和前端使用的稳定字段键。 +- `label`：页面和 Excel 使用的中文名称。 +- `field_type`：文本或日期。 +- `required`：该模板签发证书时是否必填。 +- `example`：Excel 填写示例。 +- `description`：后台提示和 Excel 填写说明。 + +新增模板时，应先在这里声明字段。手工创建表单、Excel 表头、Excel 校验和后端签发校验都会读取同一份定义。 + +## 4. 导入流程 + +### 证书导入 + +1. 先选择证书模板。 +2. 下载该模板专用 Excel，不能跨模板混用。 +3. 上传后只做校验，不立即写入正式证书。 +4. 管理员确认后调用统一签发服务。 +5. 学员不存在时自动创建；重复证书跳过；身份冲突或字段错误进入失败记录。 + +证书导入的公共业务列是姓名、手机号和项目代码。其余列由所选证书模板决定。 + +### 学员导入 + +- 姓名和手机号必填，学员编号和备注选填。 +- 只维护学员主数据，不创建证书。 +- 已有学员姓名手机号一致时可补充学员编号和备注。 +- 冲突行单独返回，不修改已有学员姓名。 + +## 5. 签发入口 + +手工创建和证书导入统一调用 `app/services/certificate_issuance.py`，由该服务负责： + +- 校验项目和模板； +- 解析或创建学员； +- 校验模板字段和课程日期； +- 判断重复证书； +- 创建证书编号； +- 创建公开链接和二维码令牌。 + +路由层只处理权限、HTTP 参数、事务提交和操作日志，不应再次复制签发规则。 diff --git a/backend/tests/test_certificate_issuance.py b/backend/tests/test_certificate_issuance.py new file mode 100644 index 0000000..4d262a3 --- /dev/null +++ b/backend/tests/test_certificate_issuance.py @@ -0,0 +1,109 @@ +from datetime import date + +import pytest +from sqlalchemy import create_engine +from sqlalchemy.orm import Session + +from app.api.routes.admin_imports import COL_COURSE_NAME, COL_STAGE_NAME, build_import_template_workbook +from app.db.base import Base +from app.models import Certificate, Learner, ProjectCourse +from app.services.certificate_issuance import CertificateIssueData, DuplicateCertificate, issue_certificate +from app.services.learner_identity import LearnerIdentityConflict, resolve_learner + + +@pytest.fixture +def db() -> Session: + engine = create_engine("sqlite+pysqlite:///:memory:") + Base.metadata.create_all(engine) + with Session(engine) as session: + yield session + + +def add_project(db: Session) -> None: + db.add( + ProjectCourse( + code="DBY", + name="大本营", + default_certificate_name="培训结业证书", + default_course_name="智慧赋能疗愈师", + default_stage_name="初级", + default_template_code="classic", + ) + ) + db.flush() + + +def test_practice_template_has_only_its_required_certificate_columns(): + sheet = build_import_template_workbook("practice-camp")["证书导入模板"] + headers = [cell.value for cell in sheet[1]] + + assert headers == ["姓名", "手机号", "项目代码", "课程开始日期", "课程结束日期", "发证日期"] + assert COL_COURSE_NAME not in headers + assert COL_STAGE_NAME not in headers + + +def test_learner_identity_reuses_exact_name_and_phone_but_rejects_name_conflict(db: Session): + learner, created = resolve_learner(db, "张三", "138-0000-0000", source="test") + reused, reused_created = resolve_learner(db, "张三", "+86 13800000000", source="test") + + assert created is True + assert reused_created is False + assert reused.id == learner.id + with pytest.raises(LearnerIdentityConflict, match="姓名.*不一致"): + resolve_learner(db, "李四", "13800000000", source="test") + + +def test_one_learner_can_hold_multiple_template_certificates(db: Session): + add_project(db) + common = dict( + learner_name="张三", + learner_phone="13800000000", + project_code="DBY", + course_start_date=date(2026, 3, 1), + course_end_date=date(2026, 8, 31), + ) + classic, _, classic_created = issue_certificate( + db, + CertificateIssueData( + **common, + template_code="classic", + issue_date=date(2026, 9, 5), + course_name="智慧赋能疗愈师", + stage_name="初级", + ), + source="test", + ) + practice, _, practice_created = issue_certificate( + db, + CertificateIssueData( + **common, + template_code="practice-camp", + issue_date=date(2026, 9, 6), + ), + source="test", + ) + + assert classic_created is True + assert practice_created is False + assert classic.learner_id == practice.learner_id + assert db.query(Learner).count() == 1 + assert db.query(Certificate).count() == 2 + assert practice.course_name is None + assert practice.stage_name is None + + +def test_duplicate_certificate_is_rejected(db: Session): + add_project(db) + data = CertificateIssueData( + learner_name="张三", + learner_phone="13800000000", + project_code="DBY", + template_code="practice-camp", + course_start_date=date(2026, 3, 1), + course_end_date=date(2026, 8, 31), + issue_date=date(2026, 9, 5), + ) + issue_certificate(db, data, source="test") + + with pytest.raises(DuplicateCertificate): + issue_certificate(db, data, source="test") diff --git a/backend/tests/test_import_rules.py b/backend/tests/test_import_rules.py index bec870d..0cd1fe4 100644 --- a/backend/tests/test_import_rules.py +++ b/backend/tests/test_import_rules.py @@ -6,10 +6,12 @@ from pydantic import ValidationError from app.api.routes.admin_imports import ( COL_ISSUE_DATE, COL_COURSE_END_DATE, + COL_COURSE_NAME, COL_COURSE_START_DATE, COL_NAME, COL_PHONE, COL_PROJECT, + COL_STAGE_NAME, TEMPLATE_HEADERS, build_import_template_workbook, date_is_valid, @@ -33,7 +35,7 @@ def test_row_errors_require_project_code_to_exist(): COL_COURSE_END_DATE: "2026-05-31", COL_ISSUE_DATE: "2026-06-01", } - assert "Project code is inactive or missing" in row_errors(row, {"DBY"}) + assert "项目代码不存在或已停用" in row_errors(row, {"DBY"}) def test_date_is_valid_rejects_bad_text(): @@ -70,16 +72,20 @@ def test_import_template_contains_date_examples_and_validation(): instruction_sheet = workbook["填写说明"] assert [cell.value for cell in data_sheet[1]] == TEMPLATE_HEADERS - assert data_sheet.column_dimensions["D"].number_format == "yyyy-mm-dd" + assert data_sheet.column_dimensions["F"].number_format == "yyyy-mm-dd" assert len(data_sheet.data_validations.dataValidation) == 3 - assert instruction_sheet["C5"].value == "2026-06-01" - assert "YYYY-MM-DD" in instruction_sheet["D5"].value + instruction_rows = {row[0].value: row for row in instruction_sheet.iter_rows(min_row=2)} + assert instruction_rows[COL_COURSE_START_DATE][2].value == "2026-06-01" + assert "实际开始日期" in instruction_rows[COL_COURSE_START_DATE][3].value + assert COL_COURSE_NAME in TEMPLATE_HEADERS + assert COL_STAGE_NAME in TEMPLATE_HEADERS def test_certificate_create_rejects_reversed_course_period(): with pytest.raises(ValidationError, match="课程结束日期不能早于课程开始日期"): CertificateCreate( - learner_id=1, + learner_name="张三", + learner_phone="13800000000", project_code="DBY", course_start_date="2026-06-30", course_end_date="2026-06-01", @@ -89,7 +95,8 @@ def test_certificate_create_rejects_reversed_course_period(): def test_certificate_create_has_no_issuer_field(): certificate = CertificateCreate( - learner_id=1, + learner_name="张三", + learner_phone="13800000000", project_code="DBY", course_start_date="2026-06-01", course_end_date="2026-06-30", diff --git a/frontend/src/api.ts b/frontend/src/api.ts index 7ae9110..8d7850c 100644 --- a/frontend/src/api.ts +++ b/frontend/src/api.ts @@ -57,6 +57,7 @@ export interface Learner { student_no: string | null; status: string; remark: string | null; + certificate_count: number; created_at: string; updated_at: string; } @@ -65,6 +66,7 @@ export interface AdminCertificate { id: number; learner_id: number; learner_name: string | null; + learner_phone: string | null; import_batch_id: number | null; project_code: string; certificate_no: string; @@ -101,10 +103,29 @@ export interface CertificateTemplate { name: string; description: string; dynamic_fields: string[]; + fields: CertificateTemplateField[]; preview_url: string; status: string; } +export interface CertificateTemplateField { + key: string; + label: string; + field_type: "text" | "date"; + required: boolean; + example: string; + description: string; +} + +export interface LearnerImportResult { + total_rows: number; + created_rows: number; + updated_rows: number; + unchanged_rows: number; + failed_rows: number; + errors: Array<{ row_no: number; message: string }>; +} + export interface OperationLog { id: number; admin_user_id: number | null; diff --git a/frontend/src/views/AdminCertificates.vue b/frontend/src/views/AdminCertificates.vue index 1ddc9ca..839702b 100644 --- a/frontend/src/views/AdminCertificates.vue +++ b/frontend/src/views/AdminCertificates.vue @@ -30,38 +30,32 @@
- - - - + + + + + - + - + - + - + + + + + + @@ -72,6 +66,7 @@ :closable="false" :title="`${selectedTemplate.name}将填写：${selectedTemplate.dynamic_fields.join('、')}`" /> +

手机号已存在时会关联原学员；不存在时将按本次姓名和手机号自动创建学员。