b3a7537364
- Certificate model: serial_number (unique), status (pending/rendering/completed/failed),
pdf_path, pdf_hash (SHA-256), verification_url, render_attempts, training_record FK
- renderer.py: LaTeX template → pdflatex subprocess → PDF; _escape_latex for XSS safety;
compute_pdf_hash for immutable verification metadata
- services.py: issue_certificate() generates serial, creates record, enqueues Celery task
- tasks.py: render_certificate_task (bind=True, autoretry 3x with 60s backoff);
sets RENDERING → COMPLETED with hash; FAILED with error on all retries exhausted
- API: /certificates/ (own list), /certificates/{id}/ (own detail)
- 15 unit + integration tests: hash consistency, LaTeX escaping, mocked render, retry behavior
Co-Authored-By: Paperclip <noreply@paperclip.ing>
47 lines
1.4 KiB
Python
47 lines
1.4 KiB
Python
from __future__ import annotations
|
|
|
|
import os
|
|
import random
|
|
import string
|
|
from datetime import datetime
|
|
|
|
from django.conf import settings
|
|
from django.utils.timezone import now
|
|
|
|
from .models import Certificate, CertificateStatus
|
|
|
|
|
|
def _generate_serial() -> str:
|
|
"""Generate a unique serial number in format CERT-YYYYMM-XXXXXXXX."""
|
|
ts = datetime.now().strftime("%Y%m")
|
|
suffix = "".join(random.choices(string.ascii_uppercase + string.digits, k=8))
|
|
return f"CERT-{ts}-{suffix}"
|
|
|
|
|
|
def issue_certificate(user, course, training_record=None) -> Certificate:
|
|
"""Create a pending certificate record and enqueue the render job."""
|
|
from .tasks import render_certificate_task
|
|
|
|
serial = _generate_serial()
|
|
while Certificate.objects.filter(serial_number=serial).exists():
|
|
serial = _generate_serial()
|
|
|
|
base_url = getattr(settings, "CERTIFICATE_VERIFICATION_BASE_URL", "/api/v1/certificates/verify/")
|
|
verification_url = f"{base_url.rstrip('/')}/{serial}/"
|
|
|
|
cert = Certificate.objects.create(
|
|
user=user,
|
|
course=course,
|
|
training_record=training_record,
|
|
serial_number=serial,
|
|
verification_url=verification_url,
|
|
)
|
|
|
|
render_certificate_task.delay(str(cert.id))
|
|
return cert
|
|
|
|
|
|
def get_pdf_storage_path(certificate: Certificate) -> str:
|
|
base = getattr(settings, "CERTIFICATE_STORAGE_ROOT", "/tmp/certificates")
|
|
return os.path.join(base, f"{certificate.id}.pdf")
|