Merge pull request #2419 from alphagov/split-task

Split task for updating letters

app/celery/tasks.py

@@ -408,39 +408,49 @@ def get_template_class(template_type):
 @notify_celery.task(bind=True, name='update-letter-notifications-statuses')
 @statsd(namespace="tasks")
 def update_letter_notifications_statuses(self, filename):
+    notification_updates = parse_dvla_file(filename)
+
+    temporary_failures = []
+
+    for update in notification_updates:
+        check_billable_units(update)
+        update_letter_notification(filename, temporary_failures, update)
+    if temporary_failures:
+        # This will alert Notify that DVLA was unable to deliver the letters, we need to investigate
+        message = "DVLA response file: {filename} has failed letters with notification.reference {failures}" \
+            .format(filename=filename, failures=temporary_failures)
+        raise DVLAException(message)
+
+
+@notify_celery.task(bind=True, name="record-daily-sorted-counts")
+@statsd(namespace="tasks")
+def record_daily_sorted_counts(self, filename):
+    sorted_letter_counts = defaultdict(int)
+    notification_updates = parse_dvla_file(filename)
+    for update in notification_updates:
+        sorted_letter_counts[update.cost_threshold.lower()] += 1
+
+    unknown_status = sorted_letter_counts.keys() - {'unsorted', 'sorted'}
+    if unknown_status:
+        message = 'DVLA response file: {} contains unknown Sorted status {}'.format(
+            filename, unknown_status.__repr__()
+        )
+        raise DVLAException(message)
+
+    billing_date = get_billing_date_in_bst_from_filename(filename)
+    persist_daily_sorted_letter_counts(day=billing_date,
+                                       file_name=filename,
+                                       sorted_letter_counts=sorted_letter_counts)
+
+
+def parse_dvla_file(filename):
     bucket_location = '{}-ftp'.format(current_app.config['NOTIFY_EMAIL_DOMAIN'])
     response_file_content = s3.get_s3_file(bucket_location, filename)
-    sorted_letter_counts = defaultdict(int)
 
     try:
-        notification_updates = process_updates_from_file(response_file_content)
+        return process_updates_from_file(response_file_content)
     except TypeError:
         raise DVLAException('DVLA response file: {} has an invalid format'.format(filename))
-    else:
-        temporary_failures = []
-        for update in notification_updates:
-            check_billable_units(update)
-            update_letter_notification(filename, temporary_failures, update)
-            sorted_letter_counts[update.cost_threshold.lower()] += 1
-
-        try:
-            unknown_status = sorted_letter_counts.keys() - {'unsorted', 'sorted'}
-            if unknown_status:
-                message = 'DVLA response file: {} contains unknown Sorted status {}'.format(
-                    filename, unknown_status
-                )
-                raise DVLAException(message)
-
-            billing_date = get_billing_date_in_bst_from_filename(filename)
-            persist_daily_sorted_letter_counts(day=billing_date,
-                                               file_name=filename,
-                                               sorted_letter_counts=sorted_letter_counts)
-        finally:
-            if temporary_failures:
-                # This will alert Notify that DVLA was unable to deliver the letters, we need to investigate
-                message = "DVLA response file: {filename} has failed letters with notification.reference {failures}" \
-                    .format(filename=filename, failures=temporary_failures)
-                raise DVLAException(message)
 
 
 def get_billing_date_in_bst_from_filename(filename):

app/commands.py

@@ -11,6 +11,8 @@ from sqlalchemy.orm.exc import NoResultFound
 from notifications_utils.statsd_decorators import statsd
 
 from app import db, DATETIME_FORMAT, encryption
+from app.aws import s3
+from app.celery.tasks import record_daily_sorted_counts
 from app.celery.nightly_tasks import send_total_sent_notifications_to_performance_platform
 from app.celery.service_callback_tasks import send_delivery_status_to_service
 from app.celery.letters_pdf_tasks import create_letters_pdf
@@ -661,3 +663,15 @@ def update_emails_to_remove_gsi(service_id):
         """
         db.session.execute(update_stmt, {'user_id': str(user.user_id)})
         db.session.commit()
+
+
+@notify_command(name='replay-daily-sorted-count-files')
+@click.option('-f', '--file_extension', required=False, help="File extension to search for, defaults to rs.txt")
+@statsd(namespace="tasks")
+def replay_daily_sorted_count_files(file_extension):
+    bucket_location = '{}-ftp'.format(current_app.config['NOTIFY_EMAIL_DOMAIN'])
+    for filename in s3.get_list_of_files_by_suffix(bucket_name=bucket_location,
+                                                   subfolder='root/dispatch',
+                                                   suffix=file_extension or '.rs.txt'):
+        print("Create task to record daily sorted counts for file: ", filename)
+        record_daily_sorted_counts.apply_async([filename], queue=QueueNames.NOTIFY)

app/notifications/notifications_letter_callback.py

@@ -9,7 +9,7 @@ from flask import (
     current_app
 )
 
-from app.celery.tasks import update_letter_notifications_statuses
+from app.celery.tasks import update_letter_notifications_statuses, record_daily_sorted_counts
 from app.v2.errors import register_errors
 from app.notifications.utils import autoconfirm_subscription
 from app.schema_validation import validate
@@ -57,6 +57,7 @@ def process_letter_response():
         if filename.lower().endswith('rs.txt') or filename.lower().endswith('rsp.txt'):
             current_app.logger.info('DVLA callback: Calling task to update letter notifications')
             update_letter_notifications_statuses.apply_async([filename], queue=QueueNames.NOTIFY)
+            record_daily_sorted_counts.apply_async([filename], queue=QueueNames.NOTIFY)
 
     return jsonify(
         result="success", message="DVLA callback succeeded"