feat: add nudge braze email using commands

enterprise_access/apps/content_assignments/api.py

@@ -594,7 +594,7 @@ def expire_assignment(assignment, content_metadata, modify_assignment=True):
     current_date = now()
 
     if auto_cancellation_date and current_date > auto_cancellation_date:
-        assignment_expiry_reason = AssignmentAutomaticExpiredReason.NIENTY_DAYS_PASSED
+        assignment_expiry_reason = AssignmentAutomaticExpiredReason.NINETY_DAYS_PASSED
     elif enrollment_end_date and enrollment_end_date < current_date:
         assignment_expiry_reason = AssignmentAutomaticExpiredReason.ENROLLMENT_DATE_PASSED
     elif subsidy_expiration_datetime and subsidy_expiration_datetime < current_date:
@@ -612,7 +612,7 @@ def expire_assignment(assignment, content_metadata, modify_assignment=True):
             logger.info('Modifying assignment %s to expired', assignment.uuid)
             assignment.state = LearnerContentAssignmentStateChoices.CANCELLED
 
-            if assignment_expiry_reason == AssignmentAutomaticExpiredReason.NIENTY_DAYS_PASSED:
+            if assignment_expiry_reason == AssignmentAutomaticExpiredReason.NINETY_DAYS_PASSED:
                 assignment.clear_pii()
                 assignment.clear_historical_pii()
 

enterprise_access/apps/content_assignments/constants.py

@@ -104,7 +104,7 @@ class AssignmentAutomaticExpiredReason:
     """
     Reason for assignment automatic expiry.
     """
-    NIENTY_DAYS_PASSED = 'NIENTY_DAYS_PASSED'
+    NINETY_DAYS_PASSED = 'NINETY_DAYS_PASSED'
     ENROLLMENT_DATE_PASSED = 'ENROLLMENT_DATE_PASSED'
     SUBSIDY_EXPIRED = 'SUBSIDY_EXPIRED'
 

enterprise_access/apps/content_assignments/management/commands/automatically_nudge_assignments.py

@@ -0,0 +1,135 @@
+"""
+Management command to automatically nudge learners enrolled in a course in advance
+"""
+
+import datetime
+import logging
+
+from django.core.management.base import BaseCommand
+from django.core.paginator import Paginator
+from django.utils import timezone
+
+from enterprise_access.apps.content_assignments.api import send_reminder_email_for_pending_assignment
+from enterprise_access.apps.content_assignments.constants import LearnerContentAssignmentStateChoices
+from enterprise_access.apps.content_assignments.content_metadata_api import get_content_metadata_for_assignments
+from enterprise_access.apps.content_assignments.models import AssignmentConfiguration
+from enterprise_access.apps.content_assignments.utils import are_dates_matching_with_day_offset
+
+logger = logging.getLogger(__name__)
+
+
+class Command(BaseCommand):
+    """
+    Automatically nudge learners who are 'days_before_course_start_date' days away from the course start_date
+    The default notification lead time without a provided `days_before_course_start_date` argument is 30 days
+    """
+    help = (
+        'Spin off celery tasks to automatically send a braze email to '
+        'remind learners about an upcoming accepted assignment a certain number '
+        'of days in advanced determined by the "days_before_course_start_date" argument'
+    )
+
+    def add_arguments(self, parser):
+        """
+        Entry point to add arguments.
+        """
+        parser.add_argument(
+            '--dry-run',
+            action='store_true',
+            dest='dry_run',
+            default=False,
+            help='Dry Run, print log messages without spawning the celery tasks.',
+        )
+        parser.add_argument(
+            '--days_before_course_start_date',
+            action='store_true',
+            dest='days_before_course_start_date',
+            default=30,
+            help='The amount of days before the course start date to send a nudge email through braze',
+        )
+
+    @staticmethod
+    def to_datetime(value):
+        """
+        Return a datetime object of `value` if it is a str.
+        """
+        if isinstance(value, str):
+            return datetime.datetime.strptime(
+                value,
+                "%Y-%m-%dT%H:%M:%SZ"
+            ).replace(
+                tzinfo=datetime.timezone.utc
+            )
+
+        return value
+
+    def handle(self, *args, **options):
+        dry_run = options['dry_run']
+        days_before_course_start_date = options['days_before_course_start_date']
+
+        for assignment_configuration in AssignmentConfiguration.objects.filter(active=True):
+            subsidy_access_policy = assignment_configuration.subsidy_access_policy
+            enterprise_catalog_uuid = subsidy_access_policy.catalog_uuid
+
+            message = (
+                '[AUTOMATICALLY_REMIND_ACCEPTED_ASSIGNMENTS_1] Assignment Configuration. UUID: [%s], '
+                'Policy: [%s], Catalog: [%s], Enterprise: [%s], dry_run [%s]',
+            )
+            logger.info(
+                message,
+                assignment_configuration.uuid,
+                subsidy_access_policy.uuid,
+                enterprise_catalog_uuid,
+                assignment_configuration.enterprise_customer_uuid,
+                dry_run,
+            )
+
+            accepted_assignments = assignment_configuration.assignments.filter(
+                state=LearnerContentAssignmentStateChoices.ACCEPTED
+            )
+
+            paginator = Paginator(accepted_assignments, 100)
+            for page_number in paginator.page_range:
+                assignments = paginator.page(page_number)
+
+                content_metadata_for_assignments = get_content_metadata_for_assignments(
+                    enterprise_catalog_uuid,
+                    assignments
+                )
+
+                for assignment in assignments:
+                    content_metadata = content_metadata_for_assignments.get(assignment.content_key, {})
+                    start_date = content_metadata.get('normalized_metadata', {}).get('start_date')
+                    course_type = content_metadata.get('course_type')
+
+                    is_executive_education_course_type = (course_type == 'executive-education-2u' or
+                                                          course_type == 'executive-education')
+
+                    # Determine if the date from today + days_before_course_state_date is
+                    # equal to the date of the start date
+                    # If they are equal, then send the nudge email, otherwise continue
+                    datetime_start_date = self.to_datetime(start_date)
+                    can_send_nudge_notification_in_advance = are_dates_matching_with_day_offset(
+                        days_offset=days_before_course_start_date,
+                        target_date=datetime_start_date,
+                        date_to_offset=timezone.now(),
+                    )
+
+                    if is_executive_education_course_type and can_send_nudge_notification_in_advance:
+                        message = (
+                            '[AUTOMATICALLY_REMIND_ACCEPTED_ASSIGNMENTS_2]  assignment_configuration_uuid: [%s], '
+                            'start_date: [%s], datetime_start_date: [%s], '
+                            'days_before_course_start_date: [%s], can_send_nudge_notification_in_advance: [%s], '
+                            'course_type: [%s], dry_run [%s]'
+                        )
+                        logger.info(message,
+                                    assignment_configuration.uuid,
+                                    start_date,
+                                    datetime_start_date,
+                                    days_before_course_start_date,
+                                    can_send_nudge_notification_in_advance,
+                                    course_type,
+                                    dry_run,
+                                    )
+                        if not dry_run:
+                            send_reminder_email_for_pending_assignment.delay(assignment.uuid)