upload file to s3 and update db after doing so

Feature/dont drop postcode

backend/address2UPRN/README.md

@@ -32,9 +32,9 @@ Step 3) Alright, now lets make the input for postcode-splitter sqs to get the ba
 postcode-splitter-sqs => https://eu-west-2.console.aws.amazon.com/sqs/v3/home?region=eu-west-2#/queues/https%3A%2F%2Fsqs.eu-west-2.amazonaws.com%2F337213553626%2Fpostcode-splitter-queue-dev
 
 {
-    "task_id": "ea615ac3-ac28-46c4-8bff-2431c5b9c13d",
-    "sub_task_id": "85a23b67-8f18-4299-9bf0-69bfb87adbc7",
-    "s3_uri": "s3://retrofit-data-dev/ara_raw_inputs/eon/eon(Sheet1).csv"
+    "sub_task_id": "c5afbd49-f0cd-4930-82bf-bafc5243a34a",
+    "task_id": "67a4b3f0-cc7a-4e8a-b314-deb783e0eedb",
+    "s3_uri": "s3://retrofit-data-dev/ara_raw_inputs/eon/pickering ferens/Pickering  Ferens - SHDF W3 Post Bid Stage MDS - Template - Vr4(in).csv"
 }
 
 Each batch of csv should be saved in retrofit-data-dev/ara_postcode_splitter_batches/<task-id>/<sub-task-id>/<timestamp:uuid4>.csv

backend/app/db/functions/uploaded_files_functions.py

@@ -0,0 +1,25 @@
+from typing import Optional
+
+from sqlalchemy import select
+
+from backend.app.db.connection import db_read_session
+from backend.app.db.models.uploaded_file import (
+    FileSourceEnum,
+    FileTypeEnum,
+    UploadedFile,
+)
+
+
+def get_uploaded_file_by_listing_type_and_source(
+    hubspot_listing_id: int,
+    file_type: FileTypeEnum,
+    file_source: FileSourceEnum,
+) -> Optional[UploadedFile]:
+    with db_read_session() as session:
+        statement = select(UploadedFile).where(
+            UploadedFile.hubspot_listing_id == hubspot_listing_id,
+            UploadedFile.file_type == file_type,
+            UploadedFile.file_source == file_source,
+        )
+
+        return session.exec(statement).one_or_none()

backend/app/db/models/uploaded_file.py

@@ -14,6 +14,8 @@ class FileTypeEnum(enum.Enum):
     PAR_PHOTO_PACK = "par_photo_pack"
     PAS_2023_PROPERTY = "pas_2023_property"
     PAS_2023_OCCUPANCY = "pas_2023_occupancy"
+    ECMK_SITE_NOTE = "ecmk_site_note"
+    ECMK_RD_SAP_SITE_NOTE = "ecmk_rd_sap_site_note"
 
 
 class FileSourceEnum(enum.Enum):

backend/ecmk_fetcher/.~lock.hubspot-crm-exports-southern-ra-lite-programme-3103-2026-03-31-2.xlsx#

@@ -0,0 +1 @@
+,daniel,daniel-Dell-15-DC15250,07.04.2026 11:47,/home/daniel/snap/onlyoffice-desktopeditors/1067/.local/share/onlyoffice;

backend/ecmk_fetcher/address_list.py

@@ -1,16 +1,15 @@
 import re
 from dataclasses import dataclass
-from typing import Any, Dict, Optional, cast
+from typing import Any, Dict, Optional
 from openpyxl import Workbook, load_workbook
 from openpyxl.worksheet.worksheet import Worksheet
-from openpyxl.cell.cell import Cell
 
 
 @dataclass
 class PropertyRow:
     row_index: int
     address: str
-    processed: bool
+    listing_id: str
 
 
 def extract_addresses_from_spreadsheet(
@@ -22,7 +21,7 @@ def extract_addresses_from_spreadsheet(
     header_row: int = 1
     id_col: Optional[int] = None
     deal_name_col: Optional[int] = None
-    processed_col: Optional[int] = None
+    listing_id_col: Optional[int] = None
 
     # find columns
     for col in range(1, ws.max_column + 1):
@@ -33,75 +32,33 @@ def extract_addresses_from_spreadsheet(
             id_col = col
         elif value == "deal name":
             deal_name_col = col
-        elif value == "processed":
-            processed_col = col
+        elif value == "associated listing ids":
+            listing_id_col = col
 
-    if id_col is None or deal_name_col is None:
+    if id_col is None or deal_name_col is None or listing_id_col is None:
         raise Exception("Missing required columns")
 
-    # create processed column if missing
-    if processed_col is None:
-        processed_col = ws.max_column + 1
-        cast(Cell, ws.cell(row=header_row, column=processed_col)).value = "processed"
-
     properties: Dict[str, PropertyRow] = {}
 
     for row in range(2, ws.max_row + 1):
         id_val: Any = ws.cell(row=row, column=id_col).value
         deal_name: Any = ws.cell(row=row, column=deal_name_col).value
+        listing_id: Any = ws.cell(row=row, column=listing_id_col).value
 
-        if not id_val or not deal_name:
+        if not id_val or not deal_name or not listing_id:
             continue
 
-        processed_val: Any = ws.cell(row=row, column=processed_col).value
-        processed: bool = str(processed_val).lower() == "true"
-
         property_id: str = str(id_val).strip()
 
         properties[property_id] = PropertyRow(
             row_index=row,
             address=extract_succinct_address(str(deal_name)),
-            processed=processed,
+            listing_id=listing_id,
         )
 
     return properties
 
 
-def mark_properties_as_processed(
-    filepath: str,
-    property_map: Dict[str, PropertyRow],
-) -> None:
-    wb: Workbook = load_workbook(filepath)
-    ws: Worksheet = wb["Southern RA-Lite Programme 3103"]
-
-    header_row: int = 1
-
-    # find processed column
-    processed_col: int | None = None
-
-    for col in range(1, ws.max_column + 1):
-        value = ws.cell(row=header_row, column=col).value
-        if value and str(value).strip().lower() == "processed":
-            processed_col = col
-            break
-
-    if processed_col is None:
-        raise Exception("Processed column not found")
-
-    # update rows
-    for property_row in property_map.values():
-        if property_row.processed:
-            cast(
-                Cell,
-                ws.cell(
-                    row=property_row.row_index,
-                    column=processed_col,
-                ),
-            ).value = True
-
-    wb.save(filepath)
-
-
 def extract_succinct_address(deal_name: str) -> str:
     left_part = deal_name.split("|")[0].strip()
 

backend/ecmk_fetcher/browser.py

@@ -50,6 +50,7 @@ def get_first_row_signature(page: Page) -> str:
 
 
 def go_to_next_page(page: Page) -> bool:
+    logger.info("Going to next page")
     before = get_first_row_signature(page)
 
     page.locator("#assessmentDatatable_next a").click()

backend/ecmk_fetcher/processor.py

@@ -8,10 +8,13 @@ from playwright.sync_api import (
     BrowserContext,
 )
 
+from backend.app.db.functions.uploaded_files_functions import (
+    get_uploaded_file_by_listing_type_and_source,
+)
+from backend.app.db.models.uploaded_file import FileSourceEnum, FileTypeEnum
 from backend.ecmk_fetcher.address_list import (
     PropertyRow,
     extract_addresses_from_spreadsheet,
-    mark_properties_as_processed,
 )
 from backend.ecmk_fetcher.browser import (
     attach_debug_listeners,
@@ -21,11 +24,21 @@ from backend.ecmk_fetcher.browser import (
     go_to_next_page,
     login,
 )
-from backend.ecmk_fetcher.reports import REPORT_TYPES, build_property_id
-from backend.ecmk_fetcher.sharepoint import upload_file_to_sharepoint
+from backend.ecmk_fetcher.reports import (
+    REPORT_TYPES,
+    build_property_id,
+    map_report_type_to_db_file_type,
+)
+from backend.ecmk_fetcher.upload import (
+    upload_file_to_s3_and_update_db,
+    upload_file_to_sharepoint,
+)
+from utils.logger import setup_logger
 from utils.sharepoint.domna_sharepoint_client import DomnaSharepointClient
 from utils.sharepoint.domna_sites import DomnaSites
 
+logger = setup_logger()
+
 
 def run_job() -> None:
     username: str = ""
@@ -46,6 +59,8 @@ def run_job() -> None:
 
     sharepoint_base_path: str = "/Projects/Southern Housing/SH-SURV-26-001/Assessments"
 
+    s3_bucket: str = "retrofit-energy-assessments-dev"
+
     with sync_playwright() as p:
         browser: Browser = p.chromium.launch(headless=True)
         context: BrowserContext = browser.new_context()
@@ -86,24 +101,44 @@ def run_job() -> None:
                         if not property_row:
                             continue
 
-                        if property_row.processed:
-                            continue
+                        logger.info(f"Match found for property {address}")
 
                         sharepoint_address: str = property_row.address
 
                         go_to_assessment_details(page, row)
 
-                        all_uploaded: bool = True
-
                         for report_type in REPORT_TYPES:
+                            hubspot_listing_id: str = property_row.listing_id
+                            try:
+                                db_file_type: FileTypeEnum = (
+                                    map_report_type_to_db_file_type(report_type)
+                                )
+
+                            except ValueError:
+                                logger.error(
+                                    f"Unknown report type {report_type}, skipping file"
+                                )
+                                continue
+
+                            if get_uploaded_file_by_listing_type_and_source(
+                                hubspot_listing_id=int(hubspot_listing_id),
+                                file_type=db_file_type,
+                                file_source=FileSourceEnum.ECMK,
+                            ):
+                                logger.debug("File already uploaded to s3, skipping")
+                                continue
+
                             file_path: str | None = download_with_retry(
                                 page, report_type
                             )
 
                             if not file_path:
-                                all_uploaded = False
                                 continue
 
+                            logger.info(
+                                f"Successfully downloaded file {os.path.basename(file_path)} from ECMK"
+                            )
+
                             try:
                                 upload_file_to_sharepoint(
                                     client=sharepoint_client,
@@ -111,16 +146,24 @@ def run_job() -> None:
                                     base_path=sharepoint_base_path,
                                     subpath=sharepoint_address,
                                 )
+                                logger.info(
+                                    f"Successfully loaded {os.path.basename(file_path)} to sharepoint for {address}"
+                                )
+
+                                # Upload to s3 and update db
+                                upload_file_to_s3_and_update_db(
+                                    bucket=s3_bucket,
+                                    file_path=file_path,
+                                    hubspot_listing_id=hubspot_listing_id,
+                                    file_type=db_file_type,
+                                )
+
                             except Exception:
-                                all_uploaded = False
                                 raise
                             finally:
                                 if os.path.exists(file_path):
                                     os.remove(file_path)
 
-                        if all_uploaded:
-                            property_row.processed = True
-
                         page.go_back()
                         page.wait_for_selector(
                             "#assessmentDatatable tbody tr", timeout=15000
@@ -135,5 +178,3 @@ def run_job() -> None:
         finally:
             context.close()
             browser.close()
-
-        mark_properties_as_processed(filepath, property_map)

backend/ecmk_fetcher/reports.py

@@ -1,5 +1,7 @@
 from enum import Enum
 
+from backend.app.db.models.uploaded_file import FileTypeEnum
+
 
 class FileDownloadButtonType(Enum):
     ASSESSOR_HUB_SITENOTE_REPORT = 11
@@ -15,6 +17,16 @@ REPORT_TYPES = [
 ]
 
 
+def map_report_type_to_db_file_type(report_type: int) -> FileTypeEnum:
+    match report_type:
+        case FileDownloadButtonType.ASSESSOR_HUB_SITENOTE_REPORT.value:
+            return FileTypeEnum.ECMK_SITE_NOTE
+        case FileDownloadButtonType.SITENOTE_REPORT.value:
+            return FileTypeEnum.ECMK_RD_SAP_SITE_NOTE
+        case _:
+            raise ValueError("Unknown report type")
+
+
 def build_report_selector(report_type: int) -> str:
     return f"a.download-report-btn[data-report-type='{report_type}']"
 

backend/ecmk_fetcher/sharepoint.py

@@ -1,20 +0,0 @@
-import os
-
-from utils.sharepoint.domna_sharepoint_client import DomnaSharepointClient
-
-
-def upload_file_to_sharepoint(
-    client: DomnaSharepointClient,
-    file_path: str,
-    base_path: str,
-    subpath: str,
-) -> None:
-    filename = os.path.basename(file_path)
-
-    full_path = f"{base_path}/{subpath}/1. Retrofit Assessment/A. Assessment"
-
-    client.upload_file(
-        file_path=file_path,
-        sharepoint_path=full_path,
-        file_name=filename,
-    )

backend/ecmk_fetcher/upload.py

@@ -0,0 +1,49 @@
+from datetime import datetime, timezone
+import os
+
+from backend.app.db.connection import db_session
+from backend.app.db.models.uploaded_file import (
+    FileSourceEnum,
+    FileTypeEnum,
+    UploadedFile,
+)
+from utils.s3 import upload_file_to_s3
+from utils.sharepoint.domna_sharepoint_client import DomnaSharepointClient
+
+
+def upload_file_to_sharepoint(
+    client: DomnaSharepointClient,
+    file_path: str,
+    base_path: str,
+    subpath: str,
+) -> None:
+    filename = os.path.basename(file_path)
+
+    full_path = f"{base_path}/{subpath}/1. Retrofit Assessment/A. Assessment"
+
+    client.upload_file(
+        file_path=file_path,
+        sharepoint_path=full_path,
+        file_name=filename,
+    )
+
+
+def upload_file_to_s3_and_update_db(
+    bucket: str, file_path: str, hubspot_listing_id: str, file_type: FileTypeEnum
+) -> None:
+    key: str = f"documents/hubspot_listing_id/{hubspot_listing_id}"
+    upload_file_to_s3(file_path, bucket, key)
+
+    uploaded_file = UploadedFile(
+        s3_file_bucket=bucket,
+        s3_file_key=key,
+        s3_upload_timestamp=datetime.now(timezone.utc),
+        hubspot_listing_id=hubspot_listing_id,
+        file_source=FileSourceEnum.ECMK.value,
+        file_type=file_type,
+    )
+
+    with db_session() as session:
+        # TODO: we should do multiple files at once to reduce db trips
+        session.add(uploaded_file)
+        session.commit()

infrastructure/terraform/lambda/_template/variables.tf

@@ -19,7 +19,7 @@ variable "image_digest" {
 
 variable "maximum_concurrency" {
   type        = number
-  default     = null
+  default     = 2
   description = "Maximum number of concurrent Lambda invocations from SQS (2-1000). null = no limit."
 }
 

infrastructure/terraform/lambda/hubspot_deal_etl/variables.tf

@@ -19,13 +19,13 @@ variable "image_digest" {
 
 variable "maximum_concurrency" {
   type        = number
-  default     = null
+  default     = 2
   description = "Maximum number of concurrent Lambda invocations from SQS (2-1000). null = no limit."
 }
 
 variable "batch_size" {
   type    = number
-  default = 1
+  default = 5
 }
 
 locals {