85 lines
2.9 KiB
Python
85 lines
2.9 KiB
Python
from fastapi import APIRouter, Depends, Query
|
|
from fastapi.responses import StreamingResponse
|
|
from openpyxl import Workbook
|
|
from sqlalchemy.orm import Session
|
|
|
|
from app.api.deps import require_roles
|
|
from app.core.config import settings
|
|
from app.core.paths import data_path
|
|
from app.db.session import get_db
|
|
from app.models import AdminUser, Certificate, CertificateAccessToken, Learner
|
|
from app.services.logs import log_action
|
|
|
|
router = APIRouter()
|
|
|
|
|
|
@router.get("/certificates")
|
|
def export_certificates(
|
|
project_code: str | None = Query(default=None),
|
|
status_value: str | None = Query(default=None, alias="status"),
|
|
db: Session = Depends(get_db),
|
|
admin: AdminUser = Depends(require_roles("system_admin", "certificate_admin")),
|
|
) -> StreamingResponse:
|
|
query = (
|
|
db.query(Certificate, Learner, CertificateAccessToken)
|
|
.join(Learner, Learner.id == Certificate.learner_id)
|
|
.join(CertificateAccessToken, CertificateAccessToken.id == Certificate.public_token_id)
|
|
.order_by(Certificate.id.desc())
|
|
)
|
|
if project_code:
|
|
query = query.filter(Certificate.project_code == project_code.strip().upper())
|
|
if status_value:
|
|
query = query.filter(Certificate.status == status_value)
|
|
|
|
workbook = Workbook()
|
|
sheet = workbook.active
|
|
sheet.title = "证书导出"
|
|
sheet.append(
|
|
[
|
|
"姓名",
|
|
"手机号",
|
|
"证书编号",
|
|
"证书对外直达链接",
|
|
"项目代码",
|
|
"证书名称",
|
|
"课程名称",
|
|
"阶段名称",
|
|
"发证日期",
|
|
"证书状态",
|
|
"PDF状态",
|
|
]
|
|
)
|
|
for certificate, learner, token in query.limit(50_000).all():
|
|
sheet.append(
|
|
[
|
|
learner.current_name,
|
|
_mask_phone(learner.phone),
|
|
certificate.certificate_no,
|
|
f"{settings.public_base_url}/cert/{token.token_value}",
|
|
certificate.project_code,
|
|
certificate.certificate_name,
|
|
certificate.course_name,
|
|
certificate.stage_name,
|
|
certificate.issue_date.isoformat(),
|
|
certificate.status,
|
|
certificate.pdf_status,
|
|
]
|
|
)
|
|
|
|
export_path = data_path("exports") / "certificates-export.xlsx"
|
|
workbook.save(export_path)
|
|
log_action(db, admin, "export_certificates", "certificate", detail={"project_code": project_code, "status": status_value})
|
|
db.commit()
|
|
file_handle = export_path.open("rb")
|
|
return StreamingResponse(
|
|
file_handle,
|
|
media_type="application/vnd.openxmlformats-officedocument.spreadsheetml.sheet",
|
|
headers={"Content-Disposition": 'attachment; filename="certificates-export.xlsx"'},
|
|
)
|
|
|
|
|
|
def _mask_phone(phone: str) -> str:
|
|
if len(phone) < 7:
|
|
return phone
|
|
return f"{phone[:3]}****{phone[-4:]}"
|