GuangYu/certificate-system
4
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

重构学员证书签发与批量导入流程

Browse Source
This commit is contained in:
Certificate System
2026-08-14 11:37:09 +08:00
parent 522f529c6a
commit 5e33657268
16 changed files with 1047 additions and 370 deletions
Download Patch File Download Diff File Expand all files Collapse all files

456
backend/app/api/routes/admin_imports.py
View File

@@ -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"