460 lines
19 KiB
Python
460 lines
19 KiB
Python
import json
|
||
import shutil
|
||
from datetime import date, datetime
|
||
from pathlib import Path
|
||
|
||
from fastapi import APIRouter, Depends, File, HTTPException, UploadFile, status
|
||
from fastapi.responses import FileResponse, StreamingResponse
|
||
from openpyxl import Workbook, load_workbook
|
||
from openpyxl.comments import Comment
|
||
from openpyxl.styles import Alignment, Font, PatternFill
|
||
from openpyxl.worksheet.datavalidation import DataValidation
|
||
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.schemas.import_batch import ImportBatchOut
|
||
from app.services.certificate_number import build_certificate_no
|
||
from app.services.logs import log_action
|
||
|
||
router = APIRouter()
|
||
|
||
COL_NAME = "\u59d3\u540d"
|
||
COL_PHONE = "\u624b\u673a\u53f7"
|
||
COL_PROJECT = "\u9879\u76ee\u4ee3\u7801"
|
||
COL_COURSE_START_DATE = "课程开始日期"
|
||
COL_COURSE_END_DATE = "课程结束日期"
|
||
COL_ISSUE_DATE = "\u53d1\u8bc1\u65e5\u671f"
|
||
|
||
TEMPLATE_HEADERS = [
|
||
COL_NAME,
|
||
COL_PHONE,
|
||
COL_PROJECT,
|
||
COL_COURSE_START_DATE,
|
||
COL_COURSE_END_DATE,
|
||
COL_ISSUE_DATE,
|
||
]
|
||
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"
|
||
workbook.save(stream_path)
|
||
file_handle = stream_path.open("rb")
|
||
return StreamingResponse(
|
||
file_handle,
|
||
media_type="application/vnd.openxmlformats-officedocument.spreadsheetml.sheet",
|
||
headers={"Content-Disposition": 'attachment; filename="certificate-import-template.xlsx"'},
|
||
)
|
||
|
||
|
||
def build_import_template_workbook() -> Workbook:
|
||
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)
|
||
return workbook
|
||
|
||
|
||
@router.post("", response_model=ImportBatchOut, status_code=status.HTTP_201_CREATED)
|
||
def upload_import_file(
|
||
file: UploadFile = File(...),
|
||
db: Session = Depends(get_db),
|
||
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")
|
||
|
||
upload_path = data_path("uploads") / file.filename
|
||
with upload_path.open("wb") as target:
|
||
shutil.copyfileobj(file.file, target)
|
||
|
||
batch = ImportBatch(filename=file.filename, file_path=str(upload_path), 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})
|
||
db.commit()
|
||
db.refresh(batch)
|
||
return batch
|
||
|
||
|
||
@router.get("", response_model=list[ImportBatchOut])
|
||
def list_import_batches(
|
||
db: Session = Depends(get_db),
|
||
_: AdminUser = Depends(require_roles("system_admin", "certificate_admin", "readonly")),
|
||
) -> list[ImportBatch]:
|
||
return db.query(ImportBatch).order_by(ImportBatch.id.desc()).limit(50).all()
|
||
|
||
|
||
@router.get("/{batch_id}/error-report")
|
||
def download_error_report(
|
||
batch_id: int,
|
||
db: Session = Depends(get_db),
|
||
_: AdminUser = Depends(require_roles("system_admin", "certificate_admin")),
|
||
) -> 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",
|
||
)
|
||
|
||
|
||
@router.get("/{batch_id}/file")
|
||
def download_source_file(
|
||
batch_id: int,
|
||
db: Session = Depends(get_db),
|
||
_: AdminUser = Depends(require_roles("system_admin", "certificate_admin")),
|
||
) -> 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,
|
||
)
|
||
|
||
|
||
@router.delete("/{batch_id}")
|
||
def delete_import_batch(
|
||
batch_id: int,
|
||
db: Session = Depends(get_db),
|
||
admin: AdminUser = Depends(require_roles("system_admin", "certificate_admin")),
|
||
) -> 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")
|
||
|
||
for file_name in [batch.file_path, batch.error_report_path]:
|
||
if file_name:
|
||
path = Path(file_name)
|
||
if path.exists():
|
||
path.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})
|
||
db.commit()
|
||
return {"ok": True}
|
||
|
||
|
||
@router.post("/{batch_id}/confirm", response_model=ImportBatchOut)
|
||
def confirm_import_batch(
|
||
batch_id: int,
|
||
db: Session = Depends(get_db),
|
||
admin: AdminUser = Depends(require_roles("system_admin", "certificate_admin")),
|
||
) -> 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")
|
||
if batch.status == "imported":
|
||
return batch
|
||
|
||
rows = db.query(ImportBatchRow).filter(ImportBatchRow.batch_id == batch.id, ImportBatchRow.status == "valid").all()
|
||
ok_rows = 0
|
||
failed_rows = 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)
|
||
if duplicate:
|
||
row.status = "skipped"
|
||
row.error_message = "\u5df2\u5b58\u5728\uff0c\u65e0\u9700\u5904\u7406"
|
||
ok_rows += 1
|
||
continue
|
||
|
||
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,
|
||
issuer_name=project.default_issuer_name,
|
||
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})
|
||
db.commit()
|
||
db.refresh(batch)
|
||
return batch
|
||
|
||
|
||
def validate_batch(db: Session, batch: ImportBatch, upload_path: Path) -> None:
|
||
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))]
|
||
header_index = {name: idx for idx, name in enumerate(header)}
|
||
missing = [name for name in TEMPLATE_HEADERS 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}"))
|
||
return
|
||
|
||
active_codes = {row[0] for row in db.query(ProjectCourse.code).filter(ProjectCourse.status == "active").all()}
|
||
total = valid = failed = 0
|
||
for row_no, row in enumerate(sheet.iter_rows(min_row=2, values_only=True), start=2):
|
||
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),
|
||
)
|
||
)
|
||
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"
|
||
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")
|
||
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]:
|
||
if row_data.get(column) and not date_is_valid(row_data[column]):
|
||
errors.append(f"{column}格式错误，请使用YYYY-MM-DD，例如2026-06-01")
|
||
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:
|
||
errors.append("课程结束日期不能早于课程开始日期")
|
||
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 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,
|
||
) -> 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)
|
||
.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 optional_text(value: object) -> str | None:
|
||
if value is None:
|
||
return None
|
||
text = str(value).strip()
|
||
return text or None
|
||
|
||
|
||
def parse_issue_date(value: object) -> date:
|
||
return parse_date(value, COL_ISSUE_DATE)
|
||
|
||
|
||
def parse_date(value: object, field_name: str = "日期") -> date:
|
||
if isinstance(value, datetime):
|
||
return value.date()
|
||
if isinstance(value, date):
|
||
return value
|
||
text = str(value).strip()
|
||
for fmt in ["%Y-%m-%d", "%Y/%m/%d", "%Y.%m.%d"]:
|
||
try:
|
||
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}")
|
||
|
||
|
||
def date_is_valid(value: object) -> bool:
|
||
try:
|
||
parse_issue_date(value)
|
||
return True
|
||
except HTTPException:
|
||
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"])
|
||
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])
|
||
report_path = data_path("error-reports") / f"import-errors-{batch_id}.xlsx"
|
||
workbook.save(report_path)
|
||
return report_path
|
||
|
||
|
||
def _format_template_sheet(sheet) -> 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.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。", "证书管理系统")
|
||
|
||
|
||
def _add_template_instructions(workbook: Workbook) -> 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"),
|
||
]
|
||
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)
|
||
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.column_dimensions["A"].width = 20
|
||
sheet.column_dimensions["B"].width = 16
|
||
sheet.column_dimensions["C"].width = 22
|
||
sheet.column_dimensions["D"].width = 54
|
||
sheet.freeze_panes = "A2"
|