wip

Author: nhsdevws

recordprocessor/src/batch_processor.py

@@ -12,18 +12,24 @@
 from clients import logger
 from file_level_validation import file_level_validation
 from errors import NoOperationPermissions, InvalidHeaders
+from utils_for_recordprocessor import get_csv_content_dict_reader
+from typing import Optional
 
 
-def process_csv_to_fhir(incoming_message_body: dict) -> None:
+def process_csv_to_fhir(incoming_message_body: dict) -> int:
     """
     For each row of the csv, attempts to transform into FHIR format, sends a message to kinesis,
     and documents the outcome for each row in the ack file.
+    Returns the number of rows processed. While this is not used by the handler, the number of rows
+    processed must be correct and therefore is returned for logging and test purposes.
     """
+    encoder = "utf-8"  # default encoding
     try:
+        incoming_message_body["encoder"] = encoder
         interim_message_body = file_level_validation(incoming_message_body=incoming_message_body)
-    except (InvalidHeaders, NoOperationPermissions, Exception):  # pylint: disable=broad-exception-caught
-        # If the file is invalid, processing should cease immediately
-        return
+    except (InvalidHeaders, NoOperationPermissions, Exception) as e:  # pylint: disable=broad-exception-caught
+        logger.error(f"File level validation failed: {e}")        # If the file is invalid, processing should cease immediately
+        return 0
 
     file_id = interim_message_body.get("message_id")
     vaccine = interim_message_body.get("vaccine")
@@ -36,40 +42,81 @@ def process_csv_to_fhir(incoming_message_body: dict) -> None:
     target_disease = map_target_disease(vaccine)
 
     row_count = 0
-    for row in csv_reader:
-        row_count += 1
-        row_id = f"{file_id}^{row_count}"
-        logger.info("MESSAGE ID : %s", row_id)
+    row_count, err = process_rows(file_id, vaccine, supplier, file_key, allowed_operations,
+                                  created_at_formatted_string, csv_reader, target_disease)
+
+    if err:
+        if isinstance(err, UnicodeDecodeError):
+            """ resolves encoding issue VED-754 """
+            logger.warning(f"Encoding Error: {err}.")
+            new_encoder = "cp1252"
+            logger.info(f"Encode error at row {row_count} with {encoder}. Switch to {new_encoder}")
+            encoder = new_encoder
+
+            # load alternative encoder
+            csv_reader = get_csv_content_dict_reader(file_key, encoder=encoder)
+            # re-read the file and skip processed rows
+            row_count, = process_rows(file_id, vaccine, supplier, file_key, allowed_operations,
+                                      created_at_formatted_string, csv_reader, target_disease, row_count)
+        else:
+            logger.error(f"Row Processing error: {err}")
+            raise err
 
-        # Process the row to obtain the details needed for the message_body and ack file
-        details_from_processing = process_row(target_disease, allowed_operations, row)
-
-        # Create the message body for sending
-        outgoing_message_body = {
-            "row_id": row_id,
-            "file_key": file_key,
-            "supplier": supplier,
-            "vax_type": vaccine,
-            "created_at_formatted_string": created_at_formatted_string,
-            **details_from_processing,
-        }
-
-        send_to_kinesis(supplier, outgoing_message_body, vaccine)
+    update_audit_table_status(file_key, file_id, FileStatus.PREPROCESSED)
+    return row_count
 
-        logger.info("Total rows processed: %s", row_count)
 
-    update_audit_table_status(file_key, file_id, FileStatus.PREPROCESSED)
+# Process the row to obtain the details needed for the message_body and ack file
+def process_rows(file_id, vaccine, supplier, file_key, allowed_operations, created_at_formatted_string,
+                 csv_reader, target_disease,
+                 total_rows_processed_count=0) -> tuple[int, Optional[Exception]]:
+    """
+    Processes each row in the csv_reader starting from start_row.
+    """
+    row_count = 0
+    start_row = total_rows_processed_count
+    try:
+        for row in csv_reader:
+            row_count += 1
+            if row_count > start_row:
+                row_id = f"{file_id}^{row_count}"
+                logger.info("MESSAGE ID : %s", row_id)
+                # Log progress every 1000 rows and the first 10 rows after a restart
+                if total_rows_processed_count % 1000 == 0:
+                    logger.info(f"Process: {total_rows_processed_count+1}")
+                if start_row > 0 and row_count <= start_row+10:
+                    logger.info(f"Restarted Process (log up to first 10): {total_rows_processed_count+1}")
+                # Process the row to obtain the details needed for the message_body and ack file
+                details_from_processing = process_row(target_disease, allowed_operations, row)
+                # Create the message body for sending
+                outgoing_message_body = {
+                    "row_id": row_id,
+                    "file_key": file_key,
+                    "supplier": supplier,
+                    "vax_type": vaccine,
+                    "created_at_formatted_string": created_at_formatted_string,
+                    **details_from_processing,
+                }
+                send_to_kinesis(supplier, outgoing_message_body, vaccine)
+                total_rows_processed_count += 1
+    except UnicodeDecodeError as error:  # pylint: disable=broad-exception-caught
+        logger.error("Error processing row %s: %s", row_count, error)
+        return total_rows_processed_count, error
+
+    return total_rows_processed_count, None
 
 
 def main(event: str) -> None:
     """Process each row of the file"""
     logger.info("task started")
     start = time.time()
+    n_rows_processed = 0
     try:
-        process_csv_to_fhir(incoming_message_body=json.loads(event))
+        n_rows_processed = process_csv_to_fhir(incoming_message_body=json.loads(event))
     except Exception as error:  # pylint: disable=broad-exception-caught
         logger.error("Error processing message: %s", error)
     end = time.time()
+    logger.info("Total rows processed: %s", n_rows_processed)
     logger.info("Total time for completion: %ss", round(end - start, 5))
 
 

recordprocessor/src/file_level_validation.py

@@ -76,9 +76,17 @@ def file_level_validation(incoming_message_body: dict) -> dict:
         file_key = incoming_message_body.get("filename")
         permission = incoming_message_body.get("permission")
         created_at_formatted_string = incoming_message_body.get("created_at_formatted_string")
+        encoder = incoming_message_body.get("encoder", "utf-8")
 
         # Fetch the data
-        csv_reader = get_csv_content_dict_reader(file_key)
+        try:
+            csv_reader = get_csv_content_dict_reader(file_key, encoder=encoder)
+            validate_content_headers(csv_reader)
+        except UnicodeDecodeError as e:
+            logger.warning("Invalid Encoding detected: %s", e)
+            # retry with cp1252 encoding
+            csv_reader = get_csv_content_dict_reader(file_key, encoder="cp1252")
+            validate_content_headers(csv_reader)
 
         validate_content_headers(csv_reader)
 

recordprocessor/src/utils_for_recordprocessor.py

@@ -13,11 +13,11 @@ def get_environment() -> str:
     return _env if _env in ["internal-dev", "int", "ref", "sandbox", "prod"] else "internal-dev"
 
 
-def get_csv_content_dict_reader(file_key: str) -> DictReader:
+def get_csv_content_dict_reader(file_key: str, encoder="utf-8") -> DictReader:
     """Returns the requested file contents from the source bucket in the form of a DictReader"""
     response = s3_client.get_object(Bucket=os.getenv("SOURCE_BUCKET_NAME"), Key=file_key)
     binary_io = response["Body"]
-    text_io = TextIOWrapper(binary_io, encoding="utf-8", newline="")
+    text_io = TextIOWrapper(binary_io, encoding=encoder, newline="")
     return DictReader(text_io, delimiter="|")