working WiP

Author: nhsdevws

recordprocessor/poetry.lock

@@ -10,17 +10,18 @@
 from audit_table import update_audit_table_status
 from send_to_kinesis import send_to_kinesis
 from clients import logger
-from file_level_validation import file_level_validation
+from file_level_validation import file_level_validation, get_csv_content_dict_reader
 from errors import NoOperationPermissions, InvalidHeaders
 
 
-def process_csv_to_fhir(incoming_message_body: dict) -> None:
+def process_csv_to_fhir(incoming_message_body: dict, encoding="utf-8", start_row=0) -> 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.
     """
     try:
-        interim_message_body = file_level_validation(incoming_message_body=incoming_message_body)
+        interim_message_body = file_level_validation(incoming_message_body=incoming_message_body,
+                                                     encoding=encoding)
     except (InvalidHeaders, NoOperationPermissions, Exception):  # pylint: disable=broad-exception-caught
         # If the file is invalid, processing should cease immediately
         return
@@ -36,27 +37,65 @@ 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)
-
-        # 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)
-
-        logger.info("Total rows processed: %s", row_count)
+    try:
+        for row in csv_reader:
+            if row_count >= start_row:
+                row_count += 1
+                row_id = f"{file_id}^{row_count}"
+                logger.info("MESSAGE ID : %s", row_id)
+                # concat dict to string for logging
+                # row_str = ", ".join(f"{v}" for k, v in row.items())
+                # print(f"Processing row {row_count}: {row_str[:20]}")
+
+                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)
+
+                logger.info("Total rows processed: %s", row_count)
+    except Exception as error:  # pylint: disable=broad-exception-caught
+        # encoder = "latin-1"
+        encoder = "cp1252"
+        print(f"Error processing: {error}.")
+        print(f"Encode error at row {row_count} with {encoding}. Switch to {encoder}")
+        # 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
+        new_reader = get_csv_content_dict_reader(file_key, encoding=encoder)
+        start_row = row_count
+        row_count = 0
+        for row in new_reader:
+            row_count += 1
+            if row_count > start_row:
+                row_id = f"{file_id}^{row_count}"
+                logger.info("MESSAGE ID : %s", row_id)
+                original_representation = ", ".join(f"{v}" for k, v in row.items())
+                if original_representation[:20] == "9473089333, DORTHY, ":
+                    print(f"Processing row {row_count}: {original_representation[:40]}")
+
+                details_from_processing = process_row(target_disease, allowed_operations, row)
+
+                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)
+
+                logger.info("Total rows processed: %s", row_count)
 
     update_audit_table_status(file_key, file_id, FileStatus.PREPROCESSED)
 
@@ -66,6 +105,7 @@ def main(event: str) -> None:
     logger.info("task started")
     start = time.time()
     try:
+        # SAW - error thrown here when invalid character using windows-1252
         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)

recordprocessor/src/batch_processor.py

@@ -61,7 +61,7 @@ def move_file(bucket_name: str, source_file_key: str, destination_file_key: str)
 
 
 @file_level_validation_logging_decorator
-def file_level_validation(incoming_message_body: dict) -> dict:
+def file_level_validation(incoming_message_body: dict, encoding="utf-8") -> dict:
     """
     Validates that the csv headers are correct and that the supplier has permission to perform at least one of
     the requested operations. Uploads the inf ack file and moves the source file to the processing folder.
@@ -70,22 +70,16 @@ def file_level_validation(incoming_message_body: dict) -> dict:
     to reflect the file has been processed and the filename lambda is invoked with the next file in the queue.
     """
     try:
-        logger.info("SAW> file_level_validation...1")
         message_id = incoming_message_body.get("message_id")
-        logger.info("SAW> file_level_validation...2")
         vaccine = incoming_message_body.get("vaccine_type").upper()
-        logger.info("SAW> file_level_validation...3")
         supplier = incoming_message_body.get("supplier").upper()
-        logger.info("SAW> file_level_validation...4")
         file_key = incoming_message_body.get("filename")
-        logger.info("SAW> file_level_validation...5")
         permission = incoming_message_body.get("permission")
-        logger.info("SAW> file_level_validation...6")
         created_at_formatted_string = incoming_message_body.get("created_at_formatted_string")
-        logger.info("SAW> file_level_validation...7")
 
         # Fetch the data
-        csv_reader = get_csv_content_dict_reader(file_key)
+        # SAW Conversion ERROR here
+        csv_reader = get_csv_content_dict_reader(file_key, encoding)
         logger.info("SAW> file_level_validation...8")
 
         validate_content_headers(csv_reader)

recordprocessor/src/file_level_validation.py

@@ -3,7 +3,7 @@
 import os
 from csv import DictReader
 from io import TextIOWrapper
-from clients import s3_client, logger
+from clients import s3_client
 
 
 def get_environment() -> str:
@@ -13,20 +13,26 @@ 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, encoding="utf-8") -> DictReader:
     """Returns the requested file contents from the source bucket in the form of a DictReader"""
-    logger.info("SAW> get_csv_content_dict_reader..1")
-    logger.info("SAW> Fetch file: %s", file_key)
-    logger.info("SAW> SOURCE_BUCKET_NAME: %s", os.getenv("SOURCE_BUCKET_NAME"))
     response = s3_client.get_object(Bucket=os.getenv("SOURCE_BUCKET_NAME"), Key=file_key)
-    logger.info("SAW> get_csv_content_dict_reader..2")
     binary_io = response["Body"]
-    logger.info("SAW> get_csv_content_dict_reader..3")
-    text_io = TextIOWrapper(binary_io, encoding="utf-8", newline="")
-    logger.info("SAW> get_csv_content_dict_reader..4")
+    text_io = TextIOWrapper(binary_io, encoding=encoding, newline="")
     return DictReader(text_io, delimiter="|")
 
 
 def create_diagnostics_dictionary(error_type, status_code, error_message) -> dict:
     """Returns a dictionary containing the error_type, statusCode, and error_message"""
     return {"error_type": error_type, "statusCode": status_code, "error_message": error_message}
+
+
+def fix_cp1252_text(text: str, encoder: str, decoder: str) -> str:
+    raw_bytes = text.encode(encoder, errors='replace')  # Or 'latin-1' if needed
+    corrected_text = raw_bytes.decode(decoder, errors='replace')
+    return corrected_text
+
+
+def dict_decode(source_dict: dict, decoder: str) -> dict:
+    """Decode all string values in a dictionary from one encoding to another."""
+
+    return {k: v.decode(decoder) if isinstance(v, bytes) else v for k, v in source_dict.items()}

recordprocessor/src/utils_for_recordprocessor.py

@@ -0,0 +1,97 @@
+import unittest
+import os
+from io import BytesIO
+from unittest.mock import patch
+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
+    from batch_processor import process_csv_to_fhir
+
+
+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.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")
+        self.mock_map_target_disease = self.map_target_disease_patcher.start()
+        self.s3_get_object_patcher = patch("utils_for_recordprocessor.s3_client.get_object")
+        self.mock_s3_get_object = self.s3_get_object_patcher.start()
+        self.make_and_move_patcher = patch("file_level_validation.make_and_upload_ack_file")
+        self.mock_make_and_move = self.make_and_move_patcher.start()
+        self.make_and_move_patcher = patch("file_level_validation.move_file")
+        self.mock_move_file = self.make_and_move_patcher.start()
+        # get_permitted_operations
+        self.get_permitted_operations_patcher = patch("file_level_validation.get_permitted_operations")
+        self.mock_get_permitted_operations = self.get_permitted_operations_patcher.start()
+
+        # self.validate_content_headers_patcher = patch("file_level_validation.validate_content_headers")
+        # self.mock_validate_content_headers = self.validate_content_headers_patcher.start()
+
+    def tearDown(self):
+        patch.stopall()
+
+    def test_process_csv_to_fhir_success(self):
+        # Setup mocks
+        print("test_process_csv_to_fhir_success")
+        try:
+            test_csv_path = os.path.join(
+                os.path.dirname(__file__), "test_data", "windows-1252-accented-e.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
+
+            # Read CSV from test_csv_path as utf-8
+            ret1 = {"Body": BytesIO(b"".join(data))}
+            ret2 = {"Body": BytesIO(b"".join(data))}
+            self.mock_s3_get_object.side_effect = [ret1, ret2]
+            self.mock_map_target_disease.return_value = "RSV"
+
+            message_body = {
+                        "message_id": "file123",
+                        "vaccine_type": "covid",
+                        "supplier": "test-supplier",
+                        "filename": "file-key-1",
+                        "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"}
+
+            self.mock_map_target_disease.return_value = "RSV"
+
+            process_csv_to_fhir(message_body)
+        except Exception as e:
+            print(f"Exception during test: {e}")
+
+    def test_fix_cp1252(self):
+        # create a cp1252 string that contains an accented E
+        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")