@@ -6,6 +6,9 @@ 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
@@ -31,23 +34,24 @@ 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 = [ COL_NAME , COL_PHONE , COL_PROJECT , COL_ISSUE_DATE ]
REQUIRED_HEADERS = TEMPLATE_HEADERS
@router.get ( " /template " )
def download_template ( _ : AdminUser = Depends ( require_roles ( " system_admin " , " certificate_admin " ) ) ) - > StreamingResponse :
workbook = W orkbook( )
sheet = workbook . active
sheet . title = " \u8bc1 \u4e66 \u5bfc \u5165 \u6a21 \u677f "
sheet . append ( TEMPLATE_HEADERS )
workbook = build_import_template_w orkbook( )
stream_path = data_path ( " exports " ) / " certificate-import-template.xlsx "
workbook . save ( stream_path )
file_handle = stream_path . open ( " rb " )
@@ -58,6 +62,16 @@ def download_template(_: AdminUser = Depends(require_roles("system_admin", "cert
)
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 ( . . . ) ,
@@ -164,6 +178,8 @@ def confirm_import_batch(
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 ( )
@@ -172,7 +188,7 @@ def confirm_import_batch(
row . error_message = f " Project code is inactive or missing: { project_code } "
failed_rows + = 1
continue
duplicate = find_duplicate_certificate ( db , learner . id , project , issue_date )
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 "
@@ -187,6 +203,8 @@ def confirm_import_batch(
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 ,
@@ -266,8 +284,14 @@ def row_errors(row_data: dict[str, object], active_codes: set[str]) -> list[str]
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 " )
if row_data . get ( COL_ISSUE_DATE ) and not date_is_valid ( row_data [ COL_ISSUE_DATE ] ) :
errors . append ( " Issue date format is invalid " )
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
@@ -295,7 +319,14 @@ def get_active_project(db: Session, project_code: str) -> ProjectCourse:
return project
def find_duplicate_certificate ( db : Session , learner_id : int , project : ProjectCourse , issue_date : date ) - > Certificate | None :
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 )
@@ -303,6 +334,8 @@ def find_duplicate_certificate(db: Session, learner_id: int, project: ProjectCou
. 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 ( )
)
@@ -329,6 +362,10 @@ def optional_text(value: object) -> str | 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 ) :
@@ -339,7 +376,7 @@ def parse_issue_date(value: object) -> date:
return datetime . strptime ( text , fmt ) . date ( )
except ValueError :
continue
raise HTTPException ( status_code = status . HTTP_400_BAD_REQUEST , detail = f " Invalid issue date: { text } " )
raise HTTPException ( status_code = status . HTTP_400_BAD_REQUEST , detail = f " { field_name } 格式错误, 请使用YYYY-MM-DD: { text } " )
def date_is_valid ( value : object ) - > bool :
@@ -361,3 +398,62 @@ def write_error_report(db: Session, batch_id: int) -> Path:
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 "