Merge branch 'master' into dependabot/pip/ack_backend/pip-minor-patch-2249d2c524

Author: dlzhry2nhs

package-lock.json

@@ -12,6 +12,6 @@
   "homepage": "https://github.com/NHSDigital/immunisation-fhir-api",
   "devDependencies": {
     "license-checker": "^25.0.1",
-    "@redocly/cli": "^2.0.8"
+    "@redocly/cli": "^2.1.0"
   }
 }

package.json

@@ -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
+        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, err = 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/batch_processor.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/file_level_validation.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="|")
 
 

recordprocessor/src/utils_for_recordprocessor.py

@@ -0,0 +1,8 @@
+NHS_NUMBER|PERSON_FORENAME|PERSON_SURNAME|PERSON_DOB|PERSON_GENDER_CODE|PERSON_POSTCODE|DATE_AND_TIME|SITE_CODE|SITE_CODE_TYPE_URI|UNIQUE_ID|UNIQUE_ID_URI|ACTION_FLAG|PERFORMING_PROFESSIONAL_FORENAME|PERFORMING_PROFESSIONAL_SURNAME|RECORDED_DATE|PRIMARY_SOURCE|VACCINATION_PROCEDURE_CODE|VACCINATION_PROCEDURE_TERM|DOSE_SEQUENCE|VACCINE_PRODUCT_CODE|VACCINE_PRODUCT_TERM|VACCINE_MANUFACTURER|BATCH_NUMBER|EXPIRY_DATE|SITE_OF_VACCINATION_CODE|SITE_OF_VACCINATION_TERM|ROUTE_OF_VACCINATION_CODE|ROUTE_OF_VACCINATION_TERM|DOSE_AMOUNT|DOSE_UNIT_CODE|DOSE_UNIT_TERM|INDICATION_CODE|LOCATION_CODE|LOCATION_CODE_TYPE_URI
+"9473081898"|"DELYTH"|"PRESCOT"|"20030731"|"2"|"NE23 1NW"|"20250730T11525000"|"RJ1"|"https://fhir.nhs.uk/Id/ods-organization-code"|"MMR_CDFDPS-1330325--27-0801"|"HTTPS://ravsTest-123"|"new"|""|""|"20250730"|"FALSE"|"308081000000105"|"Measles mumps and rubella vaccination - first dose (procedure)"|"1"|"34925111000001104"|"Priorix vaccine powder and solvent for solution for injection 0.5ml pre-filled syringes (GlaxoSmithKline UK Ltd)"|"GlaxoSmithKline UK Ltd"|"NIQUTMBYAHKSGIERGSCB"|"20230519"|"368209003"|"Right arm"|"78421000"|"Intramuscular"|"0.5"|"3318611000001103"|"Pre-filled disposable injection"|"443684005"|"X99999"|"https://fhir.nhs.uk/Id/ods-organization-code"
+"9473084064"|"DORTHY"|"DICEMBRE"|"20020403"|"2"|"NE29 6BA"|"20250730T11304900"|"RJ1"|"https://fhir.nhs.uk/Id/ods-organization-code"|"MMR_CDFDPS-1330325--27-0802"|"HTTPS://ravsTest-123"|"new"|""|""|"20250730"|"FALSE"|"38598009"|"Administration of vaccine product containing only Measles morbillivirus and Mumps orthorubulavirus and Rubella virus antigens (procedure)"|"3"|"13968211000001108"|"M-M-RVAXPRO vaccine powder and solvent for suspension for injection 0.5ml pre-filled syringes (Merck Sharp & Dohme (UK) Ltd)"|"Merck Sharp & Dohme (UK) Ltd"|"HJIXUMURNAOVKRKIECZH"|"20230519"|"368209003"|"Right arm"|"78421000"|"Intramuscular"|"0.5"|"3318611000001103"|"Pre-filled disposable injection"|"443684006"|"X99999"|"https://fhir.nhs.uk/Id/ods-organization-code"
+"9473087667"|"SHAYE"|"PUHAR"|"20010903"|"0"|"NE39 2HB"|"20250730T18191400"|"RJ1"|"https://fhir.nhs.uk/Id/ods-organization-code"|"MMR_CDFDPS-1330325--27-0803"|"HTTPS://ravsTest-123"|"new"|""|""|"20250730"|"FALSE"|"170431005"|"Administration of booster dose of vaccine product containing only Measles morbillivirus and Mumps orthorubulavirus and Rubella virus antigens (procedure)"|"3"|"13968211000001108"|"M-M-RVAXPRO vaccine powder and solvent for suspension for injection 0.5ml pre-filled syringes (Merck Sharp & Dohme (UK) Ltd)"|"Merck Sharp & Dohme (UK) Ltd"|"MSBDJRRNHLWPFZQYDFGM"|"20230519"|"368209003"|"Right arm"|"78421000"|"Intramuscular"|"0.5"|"3318611000001103"|"Pre-filled disposable injection"|"443684007"|"X99999"|"https://fhir.nhs.uk/Id/ods-organization-code"
+"9473088736"|"TANESHA"|"SIEGELER"|"20010130"|"2"|"NE46 1EA"|"20250730T14452700"|"B23002"|"https://fhir.nhs.uk/Id/ods-organization-code"|"MMR_CDFDPS-1330325--27-0804"|"HTTPS://ravsTest-123"|"new"|""|""|"20250730"|"FALSE"|"170432003"|"Administration of pre-school booster dose of vaccine product containing only Measles morbillivirus and Mumps orthorubulavirus and Rubella virus antigens"|"2"|"34925111000001104"|"Priorix vaccine powder and solvent for solution for injection 0.5ml pre-filled syringes (GlaxoSmithKline UK Ltd)"|"GlaxoSmithKline UK Ltd"|"YRJNNROADZHBJHBNCHNG"|"20230519"|"368209003"|"Right arm"|"78421000"|"Intramuscular"|"0.5"|"3318611000001103"|"Pre-filled disposable injection"|"443684008"|"X99999"|"https://fhir.nhs.uk/Id/ods-organization-code"
+"9473089333"|"HELAH"|"JUDD"|"20010908"|"2"|"NE47 9HQ"|"20250730T13504900"|"B23002"|"https://fhir.nhs.uk/Id/ods-organization-code"|"MMR_CDFDPS-1330325--27-0805"|"HTTPS://ravsTest-123"|"new"|""|""|"20250730"|"FALSE"|"170433008"|"Administration of second dose of vaccine product containing only Measles morbillivirus and Mumps orthorubulavirus and Rubella virus antigens (procedure)"|"1"|"13968211000001108"|"M-M-RVAXPRO vaccine powder and solvent for suspension for injection 0.5ml pre-filled syringes (Merck Sharp & Dohme (UK) Ltd)"|"Merck Sharp & Dohme (UK) Ltd"|"SNOJRTGYMZLPMGDNRBVD"|"20230519"|"368209003"|"Right arm"|"78421000"|"Intramuscular"|"0.5"|"3318611000001103"|"Pre-filled disposable injection"|"443684009"|"X99999"|"https://fhir.nhs.uk/Id/ods-organization-code"
+""|"DELYTH"|"PRESCOT"|"20030731"|"2"|"NE23 1NW"|"20250730T11525000"|"RJ1"|"https://fhir.nhs.uk/Id/ods-organization-code"|"MMR_CDFDPS-1330325--27-0806"|"HTTPS://ravsTest-123"|"new"|""|""|"20250730"|"FALSE"|"308081000000105"|"Measles mumps and rubella vaccination - first dose (procedure)"|"1"|"34925111000001104"|"Priorix vaccine powder and solvent for solution for injection 0.5ml pre-filled syringes (GlaxoSmithKline UK Ltd)"|"GlaxoSmithKline UK Ltd"|"NIQUTMBYAHKSGIERGSCB"|"20230519"|"368209003"|"Right arm"|"78421000"|"Intramuscular"|"0.5"|"3318611000001103"|"Pre-filled disposable injection"|"443684005"|"X99999"|"https://fhir.nhs.uk/Id/ods-organization-code"
+"0"|"DORTHY"|"DIC�MBRE"|"20020403"|"2"|"NE29 6BA"|"20250730T11304900"|"RJ1"|"https://fhir.nhs.uk/Id/ods-organization-code"|"MMR_CDFDPS-1330325--27-0807"|"HTTPS://ravsTest-123"|"new"|""|""|"20250730"|"FALSE"|"38598009"|"Administration of vaccine product containing only Measles morbillivirus and Mumps orthorubulavirus and Rubella virus antigens (procedure)"|"3"|"13968211000001108"|"M-M-RVAXPRO vaccine powder and solvent for suspension for injection 0.5ml pre-filled syringes (Merck Sharp & Dohme (UK) Ltd)"|"Merck Sharp & Dohme (UK) Ltd"|"HJIXUMURNAOVKRKIECZH"|"20230519"|"368209003"|"Right arm"|"78421000"|"Intramuscular"|"0.5"|"3318611000001103"|"Pre-filled disposable injection"|"443684006"|"X99999"|"https://fhir.nhs.uk/Id/ods-organization-code"