Add export course certificate pdfs command

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}"
+            )

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}"
+            )