""" 证书管理 API 端点 提供证书相关的 RESTful API： - 获取证书列表 - 获取证书详情 - 下载证书 - 验证证书 """ from typing import Optional, List from fastapi import APIRouter, Depends, HTTPException, status, Response, Query from fastapi.responses import StreamingResponse from sqlalchemy.ext.asyncio import AsyncSession import io from app.core.deps import get_db from app.core.security import get_current_user from app.models.user import User from app.services.certificate_service import CertificateService router = APIRouter() @router.get("/templates") async def get_certificate_templates( cert_type: Optional[str] = Query(None, description="证书类型: course/exam/achievement"), db: AsyncSession = Depends(get_db), current_user: User = Depends(get_current_user) ): """获取证书模板列表""" service = CertificateService(db) templates = await service.get_templates(cert_type) return { "code": 200, "message": "success", "data": templates } @router.get("/me") async def get_my_certificates( cert_type: Optional[str] = Query(None, description="证书类型过滤"), offset: int = Query(0, ge=0), limit: int = Query(20, ge=1, le=100), db: AsyncSession = Depends(get_db), current_user: User = Depends(get_current_user) ): """获取当前用户的证书列表""" service = CertificateService(db) result = await service.get_user_certificates( user_id=current_user.id, cert_type=cert_type, offset=offset, limit=limit ) return { "code": 200, "message": "success", "data": result } @router.get("/user/{user_id}") async def get_user_certificates( user_id: int, cert_type: Optional[str] = Query(None), offset: int = Query(0, ge=0), limit: int = Query(20, ge=1, le=100), db: AsyncSession = Depends(get_db), current_user: User = Depends(get_current_user) ): """获取指定用户的证书列表（需要管理员权限）""" # 只允许查看自己的证书或管理员查看 if current_user.id != user_id and current_user.role not in ["admin", "enterprise_admin"]: raise HTTPException( status_code=status.HTTP_403_FORBIDDEN, detail="无权查看其他用户的证书" ) service = CertificateService(db) result = await service.get_user_certificates( user_id=user_id, cert_type=cert_type, offset=offset, limit=limit ) return { "code": 200, "message": "success", "data": result } @router.get("/{cert_id}") async def get_certificate_detail( cert_id: int, db: AsyncSession = Depends(get_db), current_user: User = Depends(get_current_user) ): """获取证书详情""" service = CertificateService(db) cert = await service.get_certificate_by_id(cert_id) if not cert: raise HTTPException( status_code=status.HTTP_404_NOT_FOUND, detail="证书不存在" ) return { "code": 200, "message": "success", "data": cert } @router.get("/{cert_id}/image") async def get_certificate_image( cert_id: int, db: AsyncSession = Depends(get_db), current_user: User = Depends(get_current_user) ): """获取证书分享图片""" service = CertificateService(db) try: # 获取基础URL base_url = "https://kpl.example.com/certificates" # 可从配置读取 image_bytes = await service.generate_certificate_image(cert_id, base_url) return StreamingResponse( io.BytesIO(image_bytes), media_type="image/png", headers={ "Content-Disposition": f"inline; filename=certificate_{cert_id}.png" } ) except ValueError as e: raise HTTPException( status_code=status.HTTP_404_NOT_FOUND, detail=str(e) ) except Exception as e: raise HTTPException( status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, detail=f"生成证书图片失败: {str(e)}" ) @router.get("/{cert_id}/download") async def download_certificate_pdf( cert_id: int, db: AsyncSession = Depends(get_db), current_user: User = Depends(get_current_user) ): """下载证书PDF""" service = CertificateService(db) cert = await service.get_certificate_by_id(cert_id) if not cert: raise HTTPException( status_code=status.HTTP_404_NOT_FOUND, detail="证书不存在" ) # 如果已有PDF URL则重定向 if cert.get("pdf_url"): return { "code": 200, "message": "success", "data": { "download_url": cert["pdf_url"] } } # 否则返回图片作为替代 try: base_url = "https://kpl.example.com/certificates" image_bytes = await service.generate_certificate_image(cert_id, base_url) return StreamingResponse( io.BytesIO(image_bytes), media_type="image/png", headers={ "Content-Disposition": f"attachment; filename=certificate_{cert['certificate_no']}.png" } ) except Exception as e: raise HTTPException( status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, detail=f"下载失败: {str(e)}" ) @router.get("/verify/{cert_no}") async def verify_certificate( cert_no: str, db: AsyncSession = Depends(get_db) ): """ 验证证书真伪 此接口无需登录，可用于公开验证证书 """ service = CertificateService(db) cert = await service.get_certificate_by_no(cert_no) if not cert: return { "code": 404, "message": "证书不存在或编号错误", "data": { "valid": False, "certificate_no": cert_no } } return { "code": 200, "message": "证书验证通过", "data": { "valid": True, "certificate_no": cert_no, "title": cert.get("title"), "type_name": cert.get("type_name"), "issued_at": cert.get("issued_at"), "user": cert.get("user", {}), } } @router.post("/issue/course") async def issue_course_certificate( course_id: int, course_name: str, completion_rate: float = 100.0, db: AsyncSession = Depends(get_db), current_user: User = Depends(get_current_user) ): """ 颁发课程结业证书 通常由系统在用户完成课程时自动调用 """ service = CertificateService(db) try: cert = await service.issue_course_certificate( user_id=current_user.id, course_id=course_id, course_name=course_name, completion_rate=completion_rate, user_name=current_user.full_name or current_user.username ) await db.commit() return { "code": 200, "message": "证书颁发成功", "data": cert } except ValueError as e: raise HTTPException( status_code=status.HTTP_400_BAD_REQUEST, detail=str(e) ) @router.post("/issue/exam") async def issue_exam_certificate( exam_id: int, exam_name: str, score: float, db: AsyncSession = Depends(get_db), current_user: User = Depends(get_current_user) ): """ 颁发考试合格证书 通常由系统在用户考试通过时自动调用 """ service = CertificateService(db) try: cert = await service.issue_exam_certificate( user_id=current_user.id, exam_id=exam_id, exam_name=exam_name, score=score, user_name=current_user.full_name or current_user.username ) await db.commit() return { "code": 200, "message": "证书颁发成功", "data": cert } except ValueError as e: raise HTTPException( status_code=status.HTTP_400_BAD_REQUEST, detail=str(e) )