[PRMP-754] Allow create doc ref service to process other file types (#929)

Co-authored-by: NogaNHS <[email protected]>

Author: kamenbachvarov-nhs

lambdas/enums/lambda_error.py

@@ -108,6 +108,11 @@ def create_error_body(self, params: Optional[dict] = None, **kwargs) -> str:
         "message": "ODS code does not match any of the allowed.",
         "fhir_coding": FhirIssueCoding.INVALID,
     }
+    DocRefUnauthorizedOdsCode = {
+        "err_code": "DR_4010",
+        "message": "The user is not authorised to upload documents for this patient",
+        "fhir_coding": UKCoreSpineError.ACCESS_DENIED,
+    }
     DocRefPresign = {
         "err_code": "DR_5001",
         "message": "An error occurred when creating pre-signed url for document reference",

lambdas/enums/snomed_codes.py

@@ -17,6 +17,10 @@ class SnomedCodes(Enum):
     GENERAL_MEDICAL_PRACTICE = SnomedCode(
         code="1060971000000108", display_name="General practice service"
     )
+    EHR = SnomedCode(code="717301000000104", display_name="Electronic Health Record")
+    EHR_ATTACHMENTS = SnomedCode(
+        code="24511000000107", display_name="Electronic Health Record Attachments"
+    )
     # Temporary snomed code used.
     PATIENT_DATA = SnomedCode(
         code="717391000000106", display_name="Confidential patient data"

lambdas/enums/supported_document_types.py

@@ -10,6 +10,8 @@
 class SupportedDocumentTypes(StrEnum):
     ARF = "ARF"
     LG = SnomedCodes.LLOYD_GEORGE.value.code
+    EHR = SnomedCodes.EHR.value.code
+    EHR_ATTACHMENTS = SnomedCodes.EHR_ATTACHMENTS.value.code
 
     @staticmethod
     def list():
@@ -33,12 +35,20 @@ def get_dynamodb_table_name(self) -> str:
         document_type_to_table_name = {
             SupportedDocumentTypes.ARF: os.getenv("DOCUMENT_STORE_DYNAMODB_NAME"),
             SupportedDocumentTypes.LG: os.getenv("LLOYD_GEORGE_DYNAMODB_NAME"),
+            SupportedDocumentTypes.EHR: os.getenv("LLOYD_GEORGE_DYNAMODB_NAME"),
+            SupportedDocumentTypes.EHR_ATTACHMENTS: os.getenv(
+                "LLOYD_GEORGE_DYNAMODB_NAME"
+            ),
         }
         return document_type_to_table_name[self]
 
     def get_s3_bucket_name(self) -> str:
         lookup_dict = {
             SupportedDocumentTypes.ARF: os.getenv("DOCUMENT_STORE_BUCKET_NAME"),
             SupportedDocumentTypes.LG: os.getenv("LLOYD_GEORGE_BUCKET_NAME"),
+            SupportedDocumentTypes.EHR: os.getenv("LLOYD_GEORGE_BUCKET_NAME"),
+            SupportedDocumentTypes.EHR_ATTACHMENTS: os.getenv(
+                "LLOYD_GEORGE_BUCKET_NAME"
+            ),
         }
         return lookup_dict[self]

lambdas/enums/upload_forbidden_file_extensions.py

@@ -0,0 +1,82 @@
+from enum import StrEnum
+from pathlib import Path
+from typing import List, Optional
+
+
+class ForbiddenFileType(StrEnum):
+    ACTION = "ACTION"
+    APK = "APK"
+    APP = "APP"
+    BAT = "BAT"
+    BIN = "BIN"
+    CAB = "CAB"
+    CMD = "CMD"
+    COM = "COM"
+    COMMAND = "COMMAND"
+    CPL = "CPL"
+    CSH = "CSH"
+    EX_ = "EX_"
+    EXE = "EXE"
+    GADGET = "GADGET"
+    INF1 = "INF1"
+    INS = "INS"
+    INX = "INX"
+    IPA = "IPA"
+    ISU = "ISU"
+    JOB = "JOB"
+    JSE = "JSE"
+    KSH = "KSH"
+    LNK = "LNK"
+    MSC = "MSC"
+    MSI = "MSI"
+    MSP = "MSP"
+    MST = "MST"
+    OSX = "OSX"
+    OUT = "OUT"
+    PAF = "PAF"
+    PIF = "PIF"
+    PRG = "PRG"
+    PS1 = "PS1"
+    RAR = "RAR"
+    REG = "REG"
+    RGS = "RGS"
+    RUN = "RUN"
+    SCR = "SCR"
+    SCT = "SCT"
+    SHB = "SHB"
+    SHS = "SHS"
+    U3P = "U3P"
+    VB = "VB"
+    VBE = "VBE"
+    VBS = "VBS"
+    VBSCRIPT = "VBSCRIPT"
+    WORKFLOW = "WORKFLOW"
+    WS = "WS"
+    WSF = "WSF"
+    WSH = "WSH"
+    _7Z = "7Z"
+
+
+def is_file_type_allowed(
+    filename: str, accepted_file_types: Optional[List[str]] = None
+) -> bool:
+    """
+    if accepted_file_types is empty or None:
+        → allow any file except forbidden ones
+    if accepted_file_types is provided:
+        → only allow extensions listed there
+    """
+    if "." not in filename or filename.startswith("."):
+        return False
+
+    path = Path(filename)
+    extension = path.suffix.upper().lstrip(".")
+
+    if extension in {file_type.value for file_type in ForbiddenFileType}:
+        return False
+
+    if accepted_file_types:
+        return extension in accepted_file_types
+
+    # no accepted_file_types provided, accept all non-forbidden types
+    return True

lambdas/models/upload_file_config.py

@@ -0,0 +1,24 @@
+from typing import Any, List
+
+from pydantic import BaseModel, ConfigDict
+from pydantic.alias_generators import to_camel
+
+
+class DocumentConfig(BaseModel):
+    model_config = ConfigDict(
+        validate_by_alias=True,
+        validate_by_name=True,
+        populate_by_name=True,
+        alias_generator=to_camel,
+    )
+    snomed_code: str
+    display_name: str
+    can_be_updated: bool
+    associated_snomed: str
+    multifile_upload: bool
+    multifile_zipped: bool
+    multifile_review: bool
+    can_be_discarded: bool
+    stitched: bool
+    accepted_file_types: List[str]
+    content: List[Any]

lambdas/services/base/s3_service.py

@@ -120,16 +120,28 @@ def copy_across_bucket(
         retry_on_conflict: bool = True,
     ):
         copy_source_params = {"Bucket": source_bucket, "Key": source_file_key}
-        copy_object_params = {"Bucket": dest_bucket, "Key": dest_file_key, "CopySource": copy_source_params, "StorageClass": "INTELLIGENT_TIERING"}
+        copy_object_params = {
+            "Bucket": dest_bucket,
+            "Key": dest_file_key,
+            "CopySource": copy_source_params,
+            "StorageClass": "INTELLIGENT_TIERING",
+        }
         if if_none_match:
-            copy_object_params["IfNoneMatch"] = '*'
+            copy_object_params["IfNoneMatch"] = "*"
         try:
             return self.client.copy_object(**copy_object_params)
         except ClientError as e:
             if e.response["ResponseMetadata"]["HTTPStatusCode"] == 409:
                 logger.info(f"Copy failed due to conflict, retrying: {e}")
                 if retry_on_conflict:
-                    return self.copy_across_bucket(source_bucket, source_file_key, dest_bucket, dest_file_key, if_none_match, False)
+                    return self.copy_across_bucket(
+                        source_bucket,
+                        source_file_key,
+                        dest_bucket,
+                        dest_file_key,
+                        if_none_match,
+                        False,
+                    )
                 else:
                     raise e
             else: