works!

Author: nhsdevws

recordprocessor/src/batch_processor.py

@@ -9,14 +9,14 @@
 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
 
 
 def process_csv_to_fhir(incoming_message_body: dict) -> None:
     """
     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.
     """
-    encoder = "utf-8"  # default encoding
     try:
         interim_message_body = file_level_validation(incoming_message_body=incoming_message_body)
     except (InvalidHeaders, NoOperationPermissions, Exception):  # pylint: disable=broad-exception-caught
@@ -32,53 +32,72 @@ def process_csv_to_fhir(incoming_message_body: dict) -> None:
     csv_reader = interim_message_body.get("csv_dict_reader")
 
     target_disease = map_target_disease(vaccine)
+    print("process csv to fhir")
     row_count = 0
+    encoder = "utf-8"  # default encoding
     try:
         row_count = process_rows(file_id, vaccine, supplier, file_key, allowed_operations,
                                  created_at_formatted_string, csv_reader, target_disease)
     except Exception as error:  # pylint: disable=broad-exception-caught
         new_encoder = "cp1252"
         print(f"Error processing: {error}.")
         # check if it's a decode error, ie error.args[0] begins with "'utf-8' codec can't decode byte"
-        if error.args[0].startswith("'utf-8' codec can't decode byte"):
+        if error.reason == "invalid continuation byte":
             print(f"Encode error at row {row_count} with {encoder}. Switch to {new_encoder}")
-            print(f"Detected decode error: {error.args[0]}")
-            # if we are here, re-read the file with correct encoding and ignore the processed rows
-            # if error.args[0] == "'utf-8' codec can't decode byte 0xe9 in position 2996: invalid continuation byte":
-            # cp1252
-            row_count += process_rows_retry(file_id, vaccine, supplier, file_key,
-                                            allowed_operations, created_at_formatted_string,
-                                            "cp1252", start_row=row_count)
+            # print(f"Detected decode error: {error.reason}")
+            encoder = new_encoder
+            # if we are here, re-read the file with alternative encoding and skip processed rows
+            row_count = process_rows_retry(file_id, vaccine, supplier, file_key,
+                                           allowed_operations, created_at_formatted_string,
+                                           encoder, row_count)
         else:
-            logger.error(f"Non-decode error: {error}. Cannot retry.")
+            logger.error(f"Non-decode error: {error}. Cannot retry. Call someone.")
             raise error from error
 
     logger.info("Total rows processed: %s", row_count)
-    update_audit_table_status(file_key, file_id, FileStatus.PREPROCESSED)
 
 
 def process_rows_retry(file_id, vaccine, supplier, file_key, allowed_operations,
-                       created_at_formatted_string, encoder, target_disease, start_row=0) -> int:
-    new_reader = get_csv_content_dict_reader(file_key, encoding=encoder)
-    return process_rows(file_id, vaccine, supplier, file_key, allowed_operations,
-                        created_at_formatted_string, new_reader, start_row)
+                       created_at_formatted_string, encoder, total_rows_processed_count=0) -> int:
+    """
+    Retry processing rows with a different encoding from a specific row number
+    """
+    print("process_rows_retry...")
+    new_reader = get_csv_content_dict_reader(file_key, encoder=encoder)
+
+    total_rows_processed_count = process_rows(
+        file_id, vaccine, supplier, file_key, allowed_operations,
+        created_at_formatted_string, new_reader, total_rows_processed_count)
+
+    return total_rows_processed_count
 
 
 def process_rows(file_id, vaccine, supplier, file_key, allowed_operations, created_at_formatted_string,
-                 csv_reader, target_disease, start_row=0) -> int:
+                 csv_reader, target_disease,
+                 total_rows_processed_count=0) -> int:
     """
     Processes each row in the csv_reader starting from start_row.
     """
-
+    print("process_rows...")
     row_count = 0
+    start_row = total_rows_processed_count
     for row in csv_reader:
-        if row_count >= start_row:
-            row_count += 1
+
+        row_count += 1
+        if row_count > start_row:
             row_id = f"{file_id}^{row_count}"
             logger.info("MESSAGE ID : %s", row_id)
 
+            # convert dict to string and print first 20 chars
+            if (total_rows_processed_count % 1000 == 0):
+                print(f"Process: {total_rows_processed_count}")
+            if (total_rows_processed_count > 19995):
+                print(f"Process: {total_rows_processed_count} - {row['PERSON_SURNAME']}")
+
+            # 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,
@@ -89,8 +108,9 @@ def process_rows(file_id, vaccine, supplier, file_key, allowed_operations, creat
             }
 
             send_to_kinesis(supplier, outgoing_message_body, vaccine)
-
-    return row_count
+            total_rows_processed_count += 1
+            logger.info("Total rows processed: %s", total_rows_processed_count)
+    return total_rows_processed_count
 
 
 def main(event: str) -> None:

recordprocessor/src/utils_for_recordprocessor.py

@@ -15,11 +15,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="|")
 
 

recordprocessor/tests/test_batch_processor.py

@@ -2,7 +2,7 @@
 import os
 from io import BytesIO
 from unittest.mock import patch
-from utils_for_recordprocessor import dict_decode
+# from utils_for_recordprocessor import dict_decode
 
 with patch("logging_decorator.file_level_validation_logging_decorator", lambda f: f):
     # from file_level_validation import file_level_validation
@@ -14,8 +14,8 @@ class TestProcessCsvToFhir(unittest.TestCase):
     def setUp(self):
         self.logger_info_patcher = patch("logging.Logger.info")
         self.mock_logger_info = self.logger_info_patcher.start()
-        self.update_audit_table_status_patcher = patch("batch_processor.update_audit_table_status")
-        self.mock_update_audit_table_status = self.update_audit_table_status_patcher.start()
+        # self.update_audit_table_status_patcher = patch("batch_processor.update_audit_table_status")
+        # self.mock_update_audit_table_status = self.update_audit_table_status_patcher.start()
         self.send_to_kinesis_patcher = patch("batch_processor.send_to_kinesis")
         self.mock_send_to_kinesis = self.send_to_kinesis_patcher.start()
         self.map_target_disease_patcher = patch("batch_processor.map_target_disease")
@@ -36,7 +36,7 @@ def setUp(self):
     def tearDown(self):
         patch.stopall()
 
-    def create_large_test_data(self, data: list[bytes], num_rows: int) -> list[bytes]:
+    def expand_test_data(self, data: list[bytes], num_rows: int) -> list[bytes]:
         n_rows = len(data) - 1  # Exclude header
 
         if n_rows < num_rows:
@@ -45,53 +45,38 @@ def create_large_test_data(self, data: list[bytes], num_rows: int) -> list[bytes
             body = data[1:] * multiplier
             data = header + body
             data = data[:num_rows + 1]
-
+        print(f"Expanded test data to {len(data)-1} rows")
         return data
 
-    def create_test_data_from_file(self, file_name: str, num_rows: int) -> list[bytes]:
+    def create_test_data_from_file(self, file_name: str) -> list[bytes]:
         test_csv_path = os.path.join(
             os.path.dirname(__file__), "test_data", file_name
         )
-
         with open(test_csv_path, "rb") as f:
             data = f.readlines()
+        return data
 
-        n_rows = len(data) - 1  # Exclude header
-
-        if n_rows < num_rows:
-            multiplier = (num_rows // n_rows) + 1
-            header = data[0:1]
-            body = data[1:] * multiplier
-            data = header + body
-            data = data[:num_rows + 1]
-
+    def insert_cp1252_at_end(self, data: list[bytes], new_text: bytes, field: int) -> list[bytes]:
+        for i in reversed(range(len(data))):
+            line = data[i]
+            # Split fields by pipe
+            fields = line.strip().split(b"|")
+            print(f"replace field: {fields[field]}")
+            fields[field] = new_text
+            print(f"replaced field: {fields[field]}")
+
+            # Reconstruct the line
+            data[i] = b"|".join(fields) + b"\n"
+            break
         return data
 
     def test_process_csv_to_fhir_success(self):
         # Setup mocks
         print("test_process_csv_to_fhir_success")
         try:
-            data = self.create_test_data_from_file("test_batch_processor.csv", 20_000)
-            test_csv_path = os.path.join(
-                os.path.dirname(__file__), "test_data", "test_batch_processor.csv"
-            )
-            with open(test_csv_path, "rb") as f:
-                data = f.readlines()
-
-            # insert source_text into last row of cp1252_bytes
-            for i in reversed(range(len(data))):
-                line = data[i]
-                # Split fields by pipe
-                fields = line.strip().split(b"|")
-                print(f"replace field: {fields[2]}")
-                fields[2] = b'D\xe9cembre'
-                print(f"replaced field: {fields[2]}")
-
-                # Reconstruct the line
-                data[i] = b"|".join(fields) + b"\n"
-                break
-
-            # manually add
+            data = self.create_test_data_from_file("test-batch-data.csv")
+            data = self.expand_test_data(data, 20_000)
+            data = self.insert_cp1252_at_end(data, b'D\xe9cembre', 2)
 
             # Read CSV from test_csv_path as utf-8
             ret1 = {"Body": BytesIO(b"".join(data))}
@@ -107,7 +92,6 @@ def test_process_csv_to_fhir_success(self):
                         "permission": ["COVID.R", "COVID.U", "COVID.D"],
                         "allowed_operations": ["CREATE", "UPDATE", "DELETE"],
                         "created_at_formatted_string": "2024-09-05T12:00:00Z"
-                        # "csv_dict_reader": csv_rows
                     }
             # self.mock_file_level_validation.return_value = message_body
             self.mock_get_permitted_operations.return_value = {"CREATE", "UPDATE", "DELETE"}
@@ -127,3 +111,25 @@ def test_fix_cp1252(self):
         utf8_dict = dict_decode(test_dict, "cp1252")
         self.assertEqual(utf8_dict["date"], "Décembre")
         self.assertEqual(utf8_dict["name"], "Test Name")
+
+    def dict_decode(self):
+        source_text = b'D\xe9cembre'
+        test_dict = {
+            "date": source_text,
+            "name": "Test Name"}
+        utf8_dict = dict_decode(test_dict, "cp1252")
+        self.assertEqual(utf8_dict["date"], "Décembre")
+        self.assertEqual(utf8_dict["name"], "Test Name")
+
+
+def dict_decode(input_dict: dict, encoding: str) -> dict:
+    """
+    Decode all byte strings in a dictionary to UTF-8 strings using the specified encoding.
+    """
+    decoded_dict = {}
+    for key, value in input_dict.items():
+        if isinstance(value, bytes):
+            decoded_dict[key] = value.decode(encoding)
+        else:
+            decoded_dict[key] = value
+    return decoded_dict