import logging import string import random from django.db import transaction from django.db.utils import IntegrityError from labs.models import ( User, CertificateType, CertificateIssuance, CertificateRecipient, ) from labs.certificate.exceptions import ( CertificateIssuanceError, ) logger = logging.getLogger(__name__) # Certificate number pattern: # - 16 characters long, split into 4 groups of 4 characters, separated by '-' # - case insensitive, only use lowercase letters and digits, no special characters # - remove 'o' and 'l' to avoid confusion with 0 and 1 # Example: 0000-1111-2222-3333 CERTIFICATE_NUMBER_LENGTH = 16 CERTIFICATE_NUMBER_GROUP_LENGTH = 4 CERTIFICATE_NUMBER_CHARS = ( (string.ascii_lowercase + string.digits).replace("o", "").replace("l", "") ) def _generate_unique_certificate_number() -> str: # _generate_unique_certificate_number is used to generate a unique certificate number # it does not guarantee the uniqueness of the certificate number, # the caller should ensure the uniqueness of the certificate number by itself number = random.choices(CERTIFICATE_NUMBER_CHARS, k=CERTIFICATE_NUMBER_LENGTH) number = "-".join( "".join(number[i : i + CERTIFICATE_NUMBER_GROUP_LENGTH]) for i in range(0, CERTIFICATE_NUMBER_LENGTH, CERTIFICATE_NUMBER_GROUP_LENGTH) ) return number def issue_certificate( user: User, cert_short_name: str, source: CertificateIssuance.Source = CertificateIssuance.Source.LAB, ) -> CertificateIssuance: # issue_certificate is used to issue a certificate for a user, # it will generate a unique certificate number and create a certificate issuance record # a user can only have one certificate issuance record for a certificate type # if the user already has a certificate issuance record for the certificate type, # it will return the existing certificate issuance record, # Raise `CertificateIssuanceError` if failed to issue the certificate try: cert_type = CertificateType.objects.get(short_name=cert_short_name) except CertificateType.DoesNotExist: raise CertificateIssuanceError( f"Certificate type {cert_short_name} does not exist" ) if not user.first_name or not user.last_name: raise CertificateIssuanceError( f"User {user.email} does not have a first name or last name" ) recipient_data = { "first_name": user.first_name, "last_name": user.last_name, "email": user.email, "email_verified": True, } cert_recipient, created = CertificateRecipient.objects.get_or_create( user=user, defaults=recipient_data, ) issuance = None with transaction.atomic(): cert_recipient = CertificateRecipient.objects.select_for_update().get( pk=cert_recipient.pk ) attempts = 0 while attempts < 5: # if the certificate issuance record already exists, return it try: issuance = CertificateIssuance.objects.get( recipient=cert_recipient, cert_type=cert_type, ) logger.info( f"Certificate issuance record already exists for user {user.email} " f"and certificate type {cert_short_name}" ) return issuance except CertificateIssuance.DoesNotExist: pass certificate_number = _generate_unique_certificate_number() try: issuance = CertificateIssuance.objects.create( recipient=cert_recipient, cert_type=cert_type, cert_number=certificate_number, cert_snapshot=recipient_data, source=source, ) break except IntegrityError: attempts += 1 continue if issuance: logger.info( f"Certificate issuance record created for user {user.email} and " f"certificate type {cert_short_name} after {attempts} attempts" ) from labs.tasks.certificate import generate_certificate_issuance_pdf_task generate_certificate_issuance_pdf_task.apply_async( args=[issuance.id], countdown=2 ) return issuance else: raise CertificateIssuanceError( f"Failed to issue certificate for user {user.email} and certificate type {cert_short_name}" )