[PRMT-601] Allow filepath to be used for demographics and without numbering logic (#771)

Author: NogaNHS

lambdas/enums/lloyd_george_pre_process_format.py

@@ -0,0 +1,6 @@
+from enum import StrEnum
+
+
+class LloydGeorgePreProcessFormat(StrEnum):
+    GENERAL = "general"
+    USB = "usb"

lambdas/handlers/bulk_upload_metadata_preprocessor_handler.py

@@ -1,5 +1,9 @@
-from services.bulk_upload_metadata_preprocessor_service import (
-    MetadataPreprocessorService,
+from enums.lloyd_george_pre_process_format import LloydGeorgePreProcessFormat
+from services.bulk_upload.metadata_general_preprocessor import (
+    MetadataGeneralPreprocessor,
+)
+from services.bulk_upload.metadata_usb_preprocessor import (
+    MetadataUsbPreprocessorService,
 )
 from utils.audit_logging_setup import LoggingService
 from utils.decorators.ensure_env_var import ensure_environment_variables
@@ -16,7 +20,11 @@
 @handle_lambda_exceptions
 def lambda_handler(event, _context):
     practice_directory = event.get("practiceDirectory")
+    raw_pre_format_type = event.get(
+        "preFormatType", LloydGeorgePreProcessFormat.GENERAL
+    )
 
+    pre_processor_service = get_pre_process_service(raw_pre_format_type)
     if not practice_directory:
         logger.info(
             "Failed to start metadata pre-processor due to missing practice directory"
@@ -26,5 +34,20 @@ def lambda_handler(event, _context):
     logger.info(
         f"Starting metadata pre-processor for practice directory: {practice_directory}"
     )
-    metadata_service = MetadataPreprocessorService(practice_directory)
+
+    metadata_service = pre_processor_service(practice_directory)
     metadata_service.process_metadata()
+
+
+def get_pre_process_service(raw_pre_format_type):
+    try:
+        pre_format_type = LloydGeorgePreProcessFormat(raw_pre_format_type)
+        if pre_format_type == LloydGeorgePreProcessFormat.GENERAL:
+            return MetadataGeneralPreprocessor
+        elif pre_format_type == LloydGeorgePreProcessFormat.USB:
+            return MetadataUsbPreprocessorService
+    except ValueError:
+        logger.warning(
+            f"Invalid preFormatType: '{raw_pre_format_type}', defaulting to {LloydGeorgePreProcessFormat.GENERAL}."
+        )
+        return MetadataGeneralPreprocessor

lambdas/services/bulk_upload/metadata_general_preprocessor.py

@@ -0,0 +1,63 @@
+from services.bulk_upload_metadata_preprocessor_service import (
+    MetadataPreprocessorService,
+)
+from utils.audit_logging_setup import LoggingService
+from utils.exceptions import InvalidFileNameException
+from utils.filename_utils import (
+    assemble_lg_valid_file_name_full_path,
+    extract_date_from_bulk_upload_file_name,
+    extract_document_number_bulk_upload_file_name,
+    extract_document_path_for_lloyd_george_record,
+    extract_file_extension_from_bulk_upload_file_name,
+    extract_lloyd_george_record_from_bulk_upload_file_name,
+    extract_nhs_number_from_bulk_upload_file_name,
+    extract_patient_name_from_bulk_upload_file_name,
+)
+
+logger = LoggingService(__name__)
+
+
+class MetadataGeneralPreprocessor(MetadataPreprocessorService):
+    def validate_record_filename(self, file_name: str, *args, **kwargs) -> str:
+        try:
+            file_path_prefix, current_file_name = (
+                extract_document_path_for_lloyd_george_record(file_name)
+            )
+            first_document_number, second_document_number, current_file_name = (
+                extract_document_number_bulk_upload_file_name(current_file_name)
+            )
+            current_file_name = extract_lloyd_george_record_from_bulk_upload_file_name(
+                current_file_name
+            )
+            patient_name, current_file_name = (
+                extract_patient_name_from_bulk_upload_file_name(current_file_name)
+            )
+
+            if sum(c.isdigit() for c in current_file_name) != 18:
+                logger.info("Failed to find NHS number or date")
+                raise InvalidFileNameException("Incorrect NHS number or date format")
+
+            nhs_number, current_file_name = (
+                extract_nhs_number_from_bulk_upload_file_name(current_file_name)
+            )
+            date, current_file_name = extract_date_from_bulk_upload_file_name(
+                current_file_name
+            )
+            file_extension = extract_file_extension_from_bulk_upload_file_name(
+                current_file_name
+            )
+            file_name = assemble_lg_valid_file_name_full_path(
+                file_path_prefix,
+                first_document_number,
+                second_document_number,
+                patient_name,
+                nhs_number,
+                date,
+                file_extension,
+            )
+            logger.info(f"Finished processing, new file name is: {file_name}")
+            return file_name
+
+        except InvalidFileNameException as error:
+            logger.error(f"Failed to process {file_name} due to error: {error}")
+            raise error

lambdas/services/bulk_upload/metadata_usb_preprocessor.py

@@ -0,0 +1,123 @@
+import os
+from collections import defaultdict
+from datetime import date
+
+from models.staging_metadata import NHS_NUMBER_FIELD_NAME
+from services.bulk_upload_metadata_preprocessor_service import (
+    MetadataPreprocessorService,
+)
+from utils.audit_logging_setup import LoggingService
+from utils.exceptions import InvalidFileNameException
+from utils.filename_utils import (
+    assemble_lg_valid_file_name_full_path,
+    extract_date_from_bulk_upload_file_name,
+    extract_document_number_bulk_upload_file_name,
+    extract_document_path,
+    extract_nhs_number_from_bulk_upload_file_name,
+    extract_patient_name_from_bulk_upload_file_name,
+)
+
+logger = LoggingService(__name__)
+
+
+class MetadataUsbPreprocessorService(MetadataPreprocessorService):
+    def __init__(self, practice_directory: str):
+        super().__init__(practice_directory)
+        self.nhs_number_counts = defaultdict(int)
+
+    def generate_renaming_map(self, metadata_rows: list[dict]):
+        valid_metadata_rows = []
+        rejected_rows = []
+        rejected_reasons = []
+
+        for row in metadata_rows:
+            file_name = row.get("FILEPATH", "N/A")
+            try:
+                nhs_number = row.get(NHS_NUMBER_FIELD_NAME, "N/A")
+                self._validate_file_extension(file_name)
+                self._count_files_for_patient(nhs_number)
+                valid_metadata_rows.append(row)
+
+            except InvalidFileNameException as error:
+                rejected_rows.append(row)
+                rejected_reasons.append({"FILEPATH": file_name, "REASON": str(error)})
+
+        renaming_map, super_rejected_rows, super_rejected_reasons = (
+            super().generate_renaming_map(valid_metadata_rows)
+        )
+
+        rejected_rows.extend(super_rejected_rows)
+        rejected_reasons.extend(super_rejected_reasons)
+
+        return renaming_map, rejected_rows, rejected_reasons
+
+    def validate_record_filename(
+        self, file_path, metadata_nhs_number=None, *args, **kwargs
+    ) -> str:
+        self._validate_signal_file_for_patient(metadata_nhs_number)
+        directory_path, file_name = extract_document_path(file_path)
+
+        self._validate_document_parts(file_path, file_name)
+
+        (
+            nhs_number,
+            patient_name,
+            date_of_birth,
+        ) = self._extract_metadata_from_path(directory_path)
+
+        if nhs_number != metadata_nhs_number:
+            logger.warning(
+                f"File as it does not match the metadata NHS number: {file_path}"
+            )
+
+        return assemble_lg_valid_file_name_full_path(
+            file_path_prefix=directory_path + "/",
+            first_document_number=1,
+            second_document_number=1,
+            patient_name=patient_name,
+            nhs_number=nhs_number,
+            date_object=date_of_birth,
+            file_extension=".pdf",
+        )
+
+    def _count_files_for_patient(self, nhs_number):
+        self.nhs_number_counts[nhs_number] += 1
+
+    def _validate_signal_file_for_patient(self, nhs_number):
+        if self.nhs_number_counts[nhs_number] > 1:
+            raise InvalidFileNameException(
+                f"More than one file is found for {nhs_number}"
+            )
+
+    def _validate_file_extension(self, file_name: str) -> str:
+        file_extension = os.path.splitext(file_name)[1]
+        if file_extension != ".pdf":
+            logger.info("Rejecting file as it is not a PDF")
+            raise InvalidFileNameException(
+                f"File extension {file_extension} is not supported"
+            )
+        return file_extension
+
+    def _validate_document_parts(self, file_path: str, file_name: str):
+        try:
+            numbers = extract_document_number_bulk_upload_file_name(file_name)
+        except InvalidFileNameException:
+            numbers = None
+
+        if numbers:
+            first_document_number, total_document_number, _ = numbers
+            if first_document_number != 1 or total_document_number != 1:
+                logger.info(
+                    f"Rejecting file as it is part of a multi-part document: {file_path}"
+                )
+                raise InvalidFileNameException("Multi-part documents are not supported")
+
+    def _extract_metadata_from_path(self, directory_path: str) -> tuple[str, str, date]:
+        nhs_number, remaining_path = extract_nhs_number_from_bulk_upload_file_name(
+            directory_path
+        )
+        patient_name, remaining_path = extract_patient_name_from_bulk_upload_file_name(
+            remaining_path
+        )
+        date_of_birth, _ = extract_date_from_bulk_upload_file_name(remaining_path)
+        return nhs_number, patient_name, date_of_birth