Skip to content

Commit

Permalink
Add export course certificate pdfs command
Browse files Browse the repository at this point in the history
Also review the export course certificates command that exports to CSV.
GN-1286
  • Loading branch information
igobranco committed Nov 10, 2023
1 parent cb4b0b9 commit 2011e68
Show file tree
Hide file tree
Showing 2 changed files with 233 additions and 50 deletions.
Original file line number Diff line number Diff line change
Expand Up @@ -12,10 +12,15 @@

import logging
from datetime import datetime
from common.djangoapps.util.query import (
use_read_replica_if_available,
) # lint-amnesty, pylint: disable=import-error

from django.conf import settings
from django.core.management.base import BaseCommand
from lms.djangoapps.certificates.models import GeneratedCertificate # lint-amnesty, pylint: disable=import-error
from lms.djangoapps.certificates.models import (
GeneratedCertificate,
) # lint-amnesty, pylint: disable=import-error
from lms.djangoapps.instructor_task.tasks_helper.utils import ( # lint-amnesty, pylint: disable=import-error
upload_csv_to_report_store,
)
Expand All @@ -35,74 +40,89 @@ class Command(BaseCommand):
"""

def add_arguments(self, parser):
parser.add_argument(
"--certificate_download_domain",
default="course-certificate.nau.edu.pt",
help="The domain to use to download the certificates",
)
parser.add_argument("course_ids", nargs="*", metavar="course_id")
parser.add_argument("--all", action="store_true", help="Reindex all courses")

def log_msg(self, msg):
self.stdout.write(msg)
self.stdout.flush()

def handle(self, *args, **options):
"""
Execute the command
"""
certificate_download_domain = options["certificate_download_domain"]
course_ids = options["course_ids"]
certificate_download_pdf_url = getattr(
settings,
"NAU_CERTIFICATE_DOWNLOAD_PDF_URL",
f"https://{certificate_download_domain}/attachment/certificates/",
)

for course_id in course_ids:
course_key = CourseKey.from_string(course_id)

start_date = datetime.now(UTC)
rows = self._certificates(course_id)

course_generated_certificates = use_read_replica_if_available(
GeneratedCertificate.objects.filter(course_id=course_id)
)

course_key = CourseKey.from_string(course_id)
lms_base = SiteConfiguration.get_value_for_org(
course_key.org, "LMS_BASE", settings.LMS_BASE
)

# prepare output
rows = []

# append header
rows.append(
[
"course_id",
"student email",
"student name",
"certificate verify_uuid",
"certificate_web_link_url",
"certificate_download_pdf_link",
]
)

# iterate each certificate and append each certificate as a row
for certificate in course_generated_certificates:
certificate_web_link_url = (
"https://" + lms_base + "/certificates/" + certificate.verify_uuid
)
certificate_download_pdf_link = (
certificate_download_pdf_url + certificate.verify_uuid
)

rows.append(
[
course_id,
certificate.user.email,
certificate.name,
certificate.verify_uuid,
certificate_web_link_url,
certificate_download_pdf_link,
]
)

upload_csv_to_report_store(
rows,
"export_course_certificates",
course_key,
start_date,
)

def _certificates(self, course_id):
"""
Iterate the course certificates and return each line.
"""
course_generated_certificates = GeneratedCertificate.objects.filter(
course_id=course_id
)

course_key = CourseKey.from_string(course_id)
lms_base = SiteConfiguration.get_value_for_org(
course_key.org, "LMS_BASE", settings.LMS_BASE
)

# prepare output
rows = []

# append header
rows.append(
[
"course_id",
"student email",
"student name",
"certificate verify_uuid",
"certificate_web_link_url",
"certificate_download_pdf_link",
]
)

# iterate each certificate and append each certificate as a row
for certificate in course_generated_certificates:
certificate_web_link_url = (
"https://" + lms_base + "/certificates/" + certificate.verify_uuid
lms_instructor_data_download_url = (
f"https://{lms_base}/courses/{course_id}/instructor#view-data_download"
)
certificate_download_pdf_link = (
"https://course-certificate.nau.edu.pt/attachment/certificates/"
+ certificate.verify_uuid
)

rows.append(
[
course_id,
certificate.user.email,
certificate.name,
certificate.verify_uuid,
certificate_web_link_url,
certificate_download_pdf_link,
]
self.log_msg(
f"You can confirm the existence of the file on: {lms_instructor_data_download_url}"
)

return rows
Original file line number Diff line number Diff line change
@@ -0,0 +1,163 @@
"""
Export all PDF course certificates to a zip file and upload it to the `GRADES_DOWNLOAD` storage.
You can skip the `certificate_download_domain` parameter on production environment.
To manually develop the script you can edit it on the fly and execute it.
docker cp export_course_certificates_pdfs.py \
openedx_lms:/openedx/venv/lib/python3.8/site-packages/nau_openedx_extensions/management/\
commands/export_course_certificates_pdfs.py && docker exec -i openedx_lms python \
manage.py lms export_course_certificates_pdfs \
--certificate_download_domain course-certificate.dev.nau.fccn.pt \
course-v1:FCT+CTC101x+2020_T2
"""
import requests
import shutil
import logging
from datetime import datetime

from common.djangoapps.util.query import (
use_read_replica_if_available,
) # lint-amnesty, pylint: disable=import-error
from django.conf import settings
from django.core.management.base import BaseCommand
from lms.djangoapps.certificates.models import (
GeneratedCertificate,
) # lint-amnesty, pylint: disable=import-error
from lms.djangoapps.instructor_task.tasks_helper.utils import ( # lint-amnesty, pylint: disable=import-error
upload_zip_to_report_store,
)
from opaque_keys.edx.keys import CourseKey
from openedx.core.djangoapps.site_configuration.models import ( # lint-amnesty, pylint: disable=import-error
SiteConfiguration,
)
from pytz import UTC

log = logging.getLogger(__name__)


def delete_recursive(folder):
# Delete existing files if exist
try:
shutil.rmtree(folder)
except FileNotFoundError:
# directory doesn't exist
pass


def create_folder(path):
import os

try:
os.makedirs(path)
except FileExistsError:
# directory already exists
pass


def save_file(filename, content):
certificates_file = open(filename, "w")
certificates_file.write(content)
certificates_file.close()


def download_file(base_folder, url):
response = requests.get(url)
filename = base_folder + "/"
if "content-disposition" in response.headers:
content_disposition = response.headers["content-disposition"]
filename += content_disposition.split("filename=")[1]
else:
filename += url.split("/")[-1]
with open(filename, mode="wb") as file:
file.write(response.content)
file.close()


class Command(BaseCommand):
"""
Export all PDF course certificates with its links to a csv file and upload it to the
`GRADES_DOWNLOAD` storage.
"""

output_base_folder = getattr(
settings,
"NAU_EXPORT_COURSE_CERTIFICATES_PDFS_TEMP_FOLDER",
"/tmp/export_certificates",
)

def add_arguments(self, parser):
parser.add_argument(
"--certificate_download_domain",
default="course-certificate.nau.edu.pt",
help="The domain to use to download the certificates",
)
parser.add_argument("course_ids", nargs="+", metavar="course_id")

def log_msg(self, msg):
self.stdout.write(msg)
self.stdout.flush()

def handle(self, *args, **options):
"""
Execute the command
"""
certificate_download_domain = options["certificate_download_domain"]
course_ids = options["course_ids"]
certificate_download_pdf_url = getattr(
settings,
"NAU_CERTIFICATE_DOWNLOAD_PDF_URL",
f"https://{certificate_download_domain}/attachment/certificates/",
)

delete_recursive(self.output_base_folder)

for course_id in course_ids:
course_key = CourseKey.from_string(course_id)

start_date = datetime.now(UTC)

course_generated_certificates = use_read_replica_if_available(
GeneratedCertificate.objects.filter(course_id=course_id)
)
course_certificate_folder = self.output_base_folder + "/" + course_id
create_folder(course_certificate_folder)

# iterate each certificate and append each certificate as a row
count = 0
certificate_links_total = len(course_generated_certificates)

for certificate in course_generated_certificates:
certificate_download_pdf_link = (
certificate_download_pdf_url + certificate.verify_uuid
)
download_file(course_certificate_folder, certificate_download_pdf_link)
count += 1
self.log_msg(f"Downloading {count}/{certificate_links_total}")

self.log_msg(
"Compressing output to a single zip file - "
+ course_certificate_folder
+ ".zip"
)
shutil.make_archive(
course_certificate_folder, "zip", course_certificate_folder
)

zip_file = open(course_certificate_folder + ".zip", "rb")
upload_zip_to_report_store(
zip_file,
"export_course_certificates_pdfs",
course_key,
start_date,
)

lms_base = SiteConfiguration.get_value_for_org(
course_key.org, "LMS_BASE", settings.LMS_BASE
)
lms_instructor_data_download_url = (
f"https://{lms_base}/courses/{course_id}/instructor#view-data_download"
)
self.log_msg(
f"You can confirm the existence of the file on: {lms_instructor_data_download_url}"
)

0 comments on commit 2011e68

Please sign in to comment.