From 1d8995b739da8c739d5044f18f94681ae35ed212 Mon Sep 17 00:00:00 2001 From: Ivo Branco Date: Fri, 10 Nov 2023 12:44:08 +0000 Subject: [PATCH] feat: add export course certificate pdfs command Also review the export course certificates command that exports to CSV. GN-1286 --- .../commands/export_course_certificates.py | 119 +++++++------ .../export_course_certificates_pdfs.py | 166 ++++++++++++++++++ 2 files changed, 231 insertions(+), 54 deletions(-) create mode 100644 nau_openedx_extensions/management/commands/export_course_certificates_pdfs.py diff --git a/nau_openedx_extensions/management/commands/export_course_certificates.py b/nau_openedx_extensions/management/commands/export_course_certificates.py index 2d20df0..654d776 100644 --- a/nau_openedx_extensions/management/commands/export_course_certificates.py +++ b/nau_openedx_extensions/management/commands/export_course_certificates.py @@ -9,10 +9,9 @@ docker exec -i openedx_lms python manage.py lms export_course_certificates \ course-v1:FCT+CTC101x+2020_T2 """ - -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 @@ -25,8 +24,6 @@ ) from pytz import UTC -log = logging.getLogger(__name__) - class Command(BaseCommand): """ @@ -35,74 +32,88 @@ 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) - 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 + course_generated_certificates = use_read_replica_if_available( + GeneratedCertificate.objects.filter(course_id=course_id) ) - certificate_download_pdf_link = ( - "https://course-certificate.nau.edu.pt/attachment/certificates/" - + certificate.verify_uuid + + 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, - certificate.user.email, - certificate.name, - certificate.verify_uuid, - certificate_web_link_url, - certificate_download_pdf_link, + "course_id", + "student email", + "student name", + "certificate verify_uuid", + "certificate_web_link_url", + "certificate_download_pdf_link", ] ) - return rows + # 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, + ) + + 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}" + ) diff --git a/nau_openedx_extensions/management/commands/export_course_certificates_pdfs.py b/nau_openedx_extensions/management/commands/export_course_certificates_pdfs.py new file mode 100644 index 0000000..e633641 --- /dev/null +++ b/nau_openedx_extensions/management/commands/export_course_certificates_pdfs.py @@ -0,0 +1,166 @@ +""" +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 os +import shutil +from datetime import datetime + +import requests # lint-amnesty, pylint: disable=import-error +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 + + +def delete_recursive(folder): + """ + Delete a folder recursively. + """ + try: + shutil.rmtree(folder) + except FileNotFoundError: + # directory doesn't exist + pass + + +def create_folder(path): + """ + Crete a folder using the path. + """ + try: + os.makedirs(path) + except FileExistsError: + # directory already exists + pass + + +def save_file(filename, content): + """ + Save the content to a file. + """ + with open(filename, "w") as certificates_file: + certificates_file.write(content) + + +def download_file(base_folder, url): + """ + Download a file from an URL to a folder, by default use the filename header as the name of the + file. + """ + response = requests.get(url, timeout=60) + 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/", + ) + + 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 + delete_recursive(course_certificate_folder) + 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 + ) + + with open(course_certificate_folder + ".zip", "rb") as zip_file: + upload_zip_to_report_store( + zip_file, + "export_course_certificates_pdfs", + course_key, + start_date, + ) + delete_recursive(course_certificate_folder) + + 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}" + )