Merge pull request #948 from Hestia-Homes/feature/hubspot_etl

added task handelr

.github/workflows/_deploy_lambda.yml

@@ -80,6 +80,8 @@ on:
         required: false
       TF_VAR_pashub_password:
         required: false
+      TF_VAR_hubspot_api_key:
+        required: false
 jobs:
   deploy:
     runs-on: ubuntu-latest
@@ -146,6 +148,7 @@ jobs:
           TF_VAR_social_housing_wave_3_sharepoint_id: ${{ secrets.TF_VAR_social_housing_wave_3_sharepoint_id }}
           TF_VAR_pashub_email: ${{ secrets.TF_VAR_pashub_email }}
           TF_VAR_pashub_password: ${{ secrets.TF_VAR_pashub_password }}
+          TF_VAR_hubspot_api_key: ${{ secrets.TF_VAR_hubspot_api_key }}
         run: |
           ECR_REPO_URL_VAR=""
           if [[ -n "${{ inputs.ecr_repo }}" ]]; then
@@ -191,6 +194,7 @@ jobs:
           TF_VAR_social_housing_wave_3_sharepoint_id: ${{ secrets.TF_VAR_social_housing_wave_3_sharepoint_id }}
           TF_VAR_pashub_email: ${{ secrets.TF_VAR_pashub_email }}
           TF_VAR_pashub_password: ${{ secrets.TF_VAR_pashub_password }}
+          TF_VAR_hubspot_api_key: ${{ secrets.TF_VAR_hubspot_api_key }}
         run: |
           EXTRA_VARS=""
           if [[ -n "${{ inputs.ecr_repo }}" ]]; then

.github/workflows/deploy_terraform.yml

@@ -518,6 +518,7 @@ jobs:
       TF_VAR_db_host: ${{ secrets.DEV_DB_HOST }}
       TF_VAR_db_name: ${{ secrets.DEV_DB_NAME }}
       TF_VAR_db_port: ${{ secrets.DEV_DB_PORT }}
+      TF_VAR_hubspot_api_key: ${{ secrets.HUBSPOT_API_KEY }}
       AWS_ACCESS_KEY_ID: ${{ secrets.DEV_AWS_ACCESS_KEY_ID }}
       AWS_SECRET_ACCESS_KEY: ${{ secrets.DEV_AWS_SECRET_ACCESS_KEY }}
       AWS_REGION: ${{ secrets.DEV_AWS_REGION }}

backend/app/db/models/organisation.py

@@ -40,6 +40,32 @@ class HubspotDealData(SQLModel, table=True):
     coordination_status: Optional[str] = Field(default=None)
     design_status: Optional[str] = Field(default=None)
 
+    listing_id: Optional[str] = Field(default=None)
+    pashub_link: Optional[str] = Field(default=None)
+    sharepoint_link: Optional[str] = Field(default=None)
+    dampmould_growth: Optional[str] = Field(default=None)
+    pre_sap: Optional[str] = Field(default=None)
+    coordinator: Optional[str] = Field(default=None)
+    mtp_completion_date: Optional[datetime] = Field(default=None)
+    mtp_re_model_completion_date: Optional[datetime] = Field(default=None)
+    ioe_v3_completion_date: Optional[datetime] = Field(default=None)
+    proposed_measures: Optional[str] = Field(default=None)
+    approved_package: Optional[str] = Field(default=None)
+    designer: Optional[str] = Field(default=None)
+    design_completion_date: Optional[datetime] = Field(default=None)
+    actual_measures_installed: Optional[str] = Field(default=None)
+    installer: Optional[str] = Field(default=None)
+    installer_handover: Optional[str] = Field(default=None)
+    lodgement_status: Optional[str] = Field(default=None)
+    measures_lodgement_date: Optional[datetime] = Field(default=None)
+    lodgement_date: Optional[datetime] = Field(default=None)
+    expected_commencement_date: Optional[datetime] = Field(default=None)
+    surveyor: Optional[str] = Field(default=None)
+    confirmed_survey_date: Optional[datetime] = Field(default=None)
+    confirmed_survey_time: Optional[str] = Field(default=None)
+    surveyed_date: Optional[datetime] = Field(default=None)
+    design_type: Optional[str] = Field(default=None)
+
     created_at: datetime = Field(
         sa_column=Column(
             DateTime(timezone=True),

backend/pashub_fetcher/handler/handler.py

@@ -1,4 +1,5 @@
 from datetime import datetime, timezone
+import json
 import os
 import re
 from typing import Any, Dict, List, Mapping, Optional
@@ -11,8 +12,12 @@ from backend.app.db.models.uploaded_file import (
     UploadedFile,
 )
 from backend.pashub_fetcher.core_files import infer_file_type
+
 from backend.pashub_fetcher.job import Job
 from backend.pashub_fetcher.pashub_client import PashubClient, UnauthorizedError
+from backend.pashub_fetcher.pashub_to_ara_trigger_request import (
+    PashubToAraTriggerRequest,
+)
 from backend.pashub_fetcher.sharepoint_subfolders import SharepointSubfolders
 from backend.pashub_fetcher.token_getter import get_token_from_local_storage
 from utils.logger import setup_logger
@@ -72,21 +77,21 @@ def get_pashub_client(email: str, password: str) -> PashubClient:
 
 def upload_job_to_sharepoint(
     sharepoint_client: DomnaSharepointClient,
-    base_path: str,
-    job: Job,
+    # base_path: str,
+    sharepoint_link: str,
     job_files: List[str],
 ) -> None:
-    job_path = f"{base_path}/{job['address']}"
+    # job_path = f"{base_path}/{job['address']}"
 
     # Create main job folder
-    sharepoint_client.makedir(job["address"], base_path)
+    # sharepoint_client.makedir(job["address"], base_path)
 
     # Create subfolders
-    for folder in SharepointSubfolders:
-        sharepoint_client.makedir(folder.value, job_path)
+    # for folder in SharepointSubfolders:
+    #     sharepoint_client.makedir(folder.value, job_path)
 
     # Upload into assessment folder
-    assessment_path = f"{job_path}/{SharepointSubfolders.ASSESSMENT.value}"
+    assessment_path = f"{sharepoint_link}/{SharepointSubfolders.ASSESSMENT.value}"
 
     for file_path in job_files:
         filename = file_path.split("/")[-1]
@@ -131,14 +136,14 @@ def upload_job_to_s3_and_update_db(job_files: List[str], uprn: str) -> None:
 
 
 def process_job(
-    job: Job,
+    job: PashubToAraTriggerRequest,
     pashub_client: PashubClient,
     sharepoint_client: DomnaSharepointClient,
-    base_path: str,
 ) -> List[str]:
-    job_id = job["id"]
+    job_id = job.pashub_job_id
+
+    uprn: Optional[str] = job.uprn or pashub_client.get_uprn_by_job_id(job_id)
 
-    uprn: Optional[str] = pashub_client.get_uprn_by_job_id(job_id)
     if uprn:
         logger.info(f"Got UPRN {uprn} for job {job_id}")
     else:
@@ -150,24 +155,22 @@ def process_job(
         logger.info("Uploading files to s3")
         upload_job_to_s3_and_update_db(job_files, uprn)
 
-    upload_job_to_sharepoint(sharepoint_client, base_path, job, job_files)
+    # # Comment out sharepoint loading for now:
+    # Seems like the sharepoint link in pas hub is inconsistent in terms
+    # of whether it points to a property or a project
+
+    # if job.sharepoint_link:
+    #     upload_job_to_sharepoint(sharepoint_client, job.sharepoint_link, job_files)
 
     return job_files
 
 
 def handler(event: Mapping[str, Any], context: Any) -> None:
+    logger.info("Received message")
+    logger.info(f"Number of events: {len(event.get('Records', []))}")
+
     settings = get_settings()
 
-    BASE_DIR = os.path.dirname(os.path.dirname(__file__))
-    # filepath = os.path.join(BASE_DIR, "Watford_Warm_Homes_Wave_3_RA Downloads .xlsx")
-    filepath = os.path.join(
-        BASE_DIR,
-        "The_Guinness_Partnership_AtkinsR_alis_Coordination_Design_Board_1774881298.xlsx",
-    )
-
-    jobs: List[Job] = extract_jobs(filepath)
-    logger.info("Successfully loaded jobs from spreadsheet")
-
     pas_hub_email = settings.PASHUB_EMAIL
     pas_hub_password = settings.PASHUB_PASSWORD
 
@@ -180,36 +183,46 @@ def handler(event: Mapping[str, Any], context: Any) -> None:
         sharepoint_location=DomnaSites.SOCIAL_HOUSING_WAVE_3
     )
 
-    BASE_PATH = "/Osmosis-ACD Projects/Watford Warm Homes/Watford Property Folders (Shared with Client)"
-
     saved_file_paths: List[str] = []
 
-    for job in jobs:
+    for record in event.get("Records", []):
         try:
-            files = process_job(
-                job,
-                pashub_client,
-                sharepoint_client,
-                BASE_PATH,
-            )
-            saved_file_paths.extend(files)
+            body_dict = json.loads(record["body"])
+            logger.debug("Validating request body")
 
-        except UnauthorizedError:
-            logger.warning("Token expired - refreshing")
+            payload = PashubToAraTriggerRequest.model_validate(body_dict)
 
-            pashub_client = get_pashub_client(
-                pas_hub_email,
-                pas_hub_password,
-            )
+            logger.debug("Successfully validated request body")
 
-            # retry once
-            files = process_job(
-                job,
-                pashub_client,
-                sharepoint_client,
-                BASE_PATH,
-            )
-            saved_file_paths.extend(files)
+            try:
+                files: List[str] = process_job(
+                    payload,
+                    pashub_client,
+                    sharepoint_client,
+                )
+                saved_file_paths.extend(files)
+
+            except UnauthorizedError:
+                logger.warning("Token expired - refreshing")
+
+                pashub_client = get_pashub_client(
+                    pas_hub_email,
+                    pas_hub_password,
+                )
+
+                # retry once
+                files: List[str] = process_job(
+                    payload,
+                    pashub_client,
+                    sharepoint_client,
+                )
+                saved_file_paths.extend(files)
+
+        except Exception as e:
+            logger.info("Handler exception")
+            logger.error(f"Failed to process record: {e}")
+
+    logger.info("Successfully loaded jobs from spreadsheet")
 
     logger.info(f"Saved {len(saved_file_paths)} files")
 

backend/utils/subtasks.py

@@ -5,7 +5,8 @@ from typing import Callable, Any
 from uuid import UUID
 import json
 
-from backend.app.db.functions.tasks.Tasks import SubTaskInterface
+from backend.app.db.functions.tasks.Tasks import SubTaskInterface, TasksInterface
+from utils.logger import setup_logger
 
 
 def subtask_handler():
@@ -93,3 +94,76 @@ def subtask_handler():
         return wrapper
 
     return decorator
+
+
+def task_handler():
+    """
+    Decorator that wraps a Lambda handler and automatically:
+
+    - Parses body from the first SQS record (or uses the event dict directly)
+    - Creates a fresh Task + SubTask in the database
+    - Marks the subtask as in progress
+    - Executes the handler, passing the parsed body
+    - Marks complete on success, failed on exception (and re-raises)
+    """
+
+    def decorator(func: Callable[..., Any]):
+
+        task_source = f"{func.__module__}.{func.__qualname__}"
+
+        @wraps(func)
+        def wrapper(event: dict[str, Any], context: Any, *args, **kwargs):
+
+            logger = setup_logger()
+
+            # Parse body: Records-style SQS or plain dict event
+            if "Records" in event:
+                raw_body = event["Records"][0].get("body", {})
+                if isinstance(raw_body, str):
+                    try:
+                        body = json.loads(raw_body)
+                    except Exception:
+                        body = {}
+                else:
+                    body = raw_body or {}
+            else:
+                body = event
+
+            # Create fresh task + subtask
+            logger.info("Creating task for source: %s", task_source)
+            task_id, subtask_id = TasksInterface.create_task(
+                task_source=task_source,
+                inputs=body,
+            )
+            logger.info("Created task_id=%s subtask_id=%s", task_id, subtask_id)
+
+            interface = SubTaskInterface()
+
+            interface.update_subtask_status(
+                subtask_id=subtask_id,
+                status="in progress",
+            )
+
+            try:
+                result = func(body, context, *args, **kwargs)
+
+                interface.update_subtask_status(
+                    subtask_id=subtask_id,
+                    status="complete",
+                    outputs={"result": result} if result else None,
+                )
+                logger.info("Task %s completed successfully", task_id)
+                return result
+
+            except Exception as e:
+                logger.exception("Task %s failed: %s", task_id, e)
+                interface.update_subtask_status(
+                    subtask_id=subtask_id,
+                    status="failed",
+                    outputs={"error": str(e)},
+                )
+                raise
+
+        return wrapper
+
+    return decorator

etl/hubspot/hubspotClient.py

@@ -189,6 +189,7 @@ class HubspotClient:
         )
 
         listing_info: dict[str, str] = cast(dict[str, str], listing.properties)  # type: ignore[reportUnknownMemberType]
+        listing_info["listing_id"] = listing_id
         self.logger.info(f"Listing info for deal {deal_id}: {listing_info}")
         return listing_info
 
@@ -201,13 +202,37 @@ class HubspotClient:
                 "dealname",
                 "dealstage",
                 "pipeline",
-                "outcome",  # outcome,
-                "outcome_notes",  # outcome notes
+                "outcome",
+                "outcome_notes",
                 "project_code",
                 "major_condition_issue_description",
                 "major_condition_issue_photos",
-                "coordination_status__stage_1_",  # Coordiantion Status (Stage 1),
-                "retrofit_design_status",  # Retrofit Design Status
+                "coordination_status__stage_1_",
+                "retrofit_design_status",
+                "pashub_link",
+                "sharepoint_link",
+                "dampmould_growth",
+                "pre_sap",
+                "coordinator",
+                "mtp_completion_date",
+                "mtp_re_model_completion_date",
+                "ioe_v3_completion_date",
+                "proposed_measures",
+                "approved_package",
+                "designer",
+                "design_completion_date",
+                "actual_measures_installed",
+                "installer",
+                "installer_handover",
+                "lodgement_status",
+                "measures_lodgement_date",
+                "lodgement_date",
+                "expected_commencement_date",
+                "surveyor",
+                "confirmed_survey_date",
+                "confirmed_survey_time",
+                "surveyed_date",
+                "design_type",
             ],
         )
 

etl/hubspot/hubspotDataTodB.py

@@ -2,7 +2,7 @@ from backend.app.db.connection import db_read_session
 from backend.app.db.models.organisation import Organisation, HubspotDealData
 from sqlmodel import select
 from datetime import datetime, timezone
-from typing import TypedDict
+from typing import TypedDict, Optional
 from etl.hubspot.s3_uploader import S3Uploader
 import hashlib
 import os
@@ -82,6 +82,14 @@ class HubspotDataToDb:
                 .one_or_none()
             )
 
+    def _parse_hs_date(self, value: Optional[str]) -> Optional[datetime]:
+        if not value:
+            return None
+        try:
+            return datetime.fromisoformat(value.replace("Z", "+00:00"))
+        except ValueError:
+            return None
+
     def _sha256(self, file_path: str) -> str:
         """Compute SHA-256 checksum of a file."""
         sha256 = hashlib.sha256()
@@ -114,6 +122,10 @@ class HubspotDataToDb:
                 deal_in_db.deal_id == hs_deal.get("hs_object_id"), "deal_id mismatch"
             ),
             soft_assert(deal_in_db.company_id == hs_company_id, "company_id mismatch"),
+            soft_assert(
+                deal_in_db.listing_id == hs_listing.get("listing_id"),
+                "listing_id mismatch",
+            ),
             soft_assert(
                 deal_in_db.landlord_property_id == hs_listing.get("owner_property_id"),
                 "landlord_property_id mismatch",
@@ -157,6 +169,102 @@ class HubspotDataToDb:
                 deal_in_db.design_status == hs_deal.get("retrofit_design_status"),
                 "retrofit design mismatch",
             ),
+            soft_assert(
+                deal_in_db.pashub_link == hs_deal.get("pashub_link"),
+                "pashub_link mismatch",
+            ),
+            soft_assert(
+                deal_in_db.sharepoint_link == hs_deal.get("sharepoint_link"),
+                "sharepoint_link mismatch",
+            ),
+            soft_assert(
+                deal_in_db.dampmould_growth == hs_deal.get("dampmould_growth"),
+                "dampmould_growth mismatch",
+            ),
+            soft_assert(
+                deal_in_db.pre_sap == hs_deal.get("pre_sap"),
+                "pre_sap mismatch",
+            ),
+            soft_assert(
+                deal_in_db.coordinator == hs_deal.get("coordinator"),
+                "coordinator mismatch",
+            ),
+            soft_assert(
+                deal_in_db.mtp_completion_date == self._parse_hs_date(hs_deal.get("mtp_completion_date")),
+                "mtp_completion_date mismatch",
+            ),
+            soft_assert(
+                deal_in_db.mtp_re_model_completion_date == self._parse_hs_date(hs_deal.get("mtp_re_model_completion_date")),
+                "mtp_re_model_completion_date mismatch",
+            ),
+            soft_assert(
+                deal_in_db.ioe_v3_completion_date == self._parse_hs_date(hs_deal.get("ioe_v3_completion_date")),
+                "ioe_v3_completion_date mismatch",
+            ),
+            soft_assert(
+                deal_in_db.proposed_measures == hs_deal.get("proposed_measures"),
+                "proposed_measures mismatch",
+            ),
+            soft_assert(
+                deal_in_db.approved_package == hs_deal.get("approved_package"),
+                "approved_package mismatch",
+            ),
+            soft_assert(
+                deal_in_db.designer == hs_deal.get("designer"),
+                "designer mismatch",
+            ),
+            soft_assert(
+                deal_in_db.design_completion_date == self._parse_hs_date(hs_deal.get("design_completion_date")),
+                "design_completion_date mismatch",
+            ),
+            soft_assert(
+                deal_in_db.actual_measures_installed == hs_deal.get("actual_measures_installed"),
+                "actual_measures_installed mismatch",
+            ),
+            soft_assert(
+                deal_in_db.installer == hs_deal.get("installer"),
+                "installer mismatch",
+            ),
+            soft_assert(
+                deal_in_db.installer_handover == hs_deal.get("installer_handover"),
+                "installer_handover mismatch",
+            ),
+            soft_assert(
+                deal_in_db.lodgement_status == hs_deal.get("lodgement_status"),
+                "lodgement_status mismatch",
+            ),
+            soft_assert(
+                deal_in_db.measures_lodgement_date == self._parse_hs_date(hs_deal.get("measures_lodgement_date")),
+                "measures_lodgement_date mismatch",
+            ),
+            soft_assert(
+                deal_in_db.lodgement_date == self._parse_hs_date(hs_deal.get("lodgement_date")),
+                "lodgement_date mismatch",
+            ),
+            soft_assert(
+                deal_in_db.expected_commencement_date == self._parse_hs_date(hs_deal.get("expected_commencement_date")),
+                "expected_commencement_date mismatch",
+            ),
+            soft_assert(
+                deal_in_db.surveyor == hs_deal.get("surveyor"),
+                "surveyor mismatch",
+            ),
+            soft_assert(
+                deal_in_db.confirmed_survey_date == self._parse_hs_date(hs_deal.get("confirmed_survey_date")),
+                "confirmed_survey_date mismatch",
+            ),
+            soft_assert(
+                deal_in_db.confirmed_survey_time == hs_deal.get("confirmed_survey_time"),
+                "confirmed_survey_time mismatch",
+            ),
+            soft_assert(
+                deal_in_db.surveyed_date == self._parse_hs_date(hs_deal.get("surveyed_date")),
+                "surveyed_date mismatch",
+            ),
+            soft_assert(
+                deal_in_db.design_type == hs_deal.get("design_type"),
+                "design_type mismatch",
+            ),
         ]
 
         # If discrepancies found, update from HubSpot
@@ -211,6 +319,9 @@ class HubspotDataToDb:
                         f"⚠️ Failed to download/upload photo for deal_id {deal_in_db.deal_id}: {e}"
                     )
                     # Continue without the file — don't crash the entire update
+                finally:
+                    if "local_file" in locals() and os.path.exists(local_file):
+                        os.remove(local_file)
             else:
                 print(f"⚠️ Photo URL missing for deal_id {deal_in_db.deal_id}")
 
@@ -238,6 +349,7 @@ class HubspotDataToDb:
                 for attr, value in {
                     "dealname": deal_data.get("dealname"),
                     "dealstage": deal_data.get("dealstage"),
+                    "listing_id": listing.get("listing_id"),
                     "landlord_property_id": listing.get("owner_property_id"),
                     "uprn": listing.get("national_uprn"),
                     "outcome": deal_data.get("outcome"),
@@ -250,16 +362,34 @@ class HubspotDataToDb:
                     "major_condition_issue_photos": deal_data.get(
                         "major_condition_issue_photos"
                     ),
-                    "major_condition_issue_description": deal_data.get(
-                        "major_condition_issue_description"
-                    ),
-                    "major_condition_issue_photos": deal_data.get(
-                        "major_condition_issue_photos"
-                    ),
                     "coordination_status": deal_data.get(
                         "coordination_status__stage_1_"
                     ),
                     "design_status": deal_data.get("retrofit_design_status"),
+                    "pashub_link": deal_data.get("pashub_link"),
+                    "sharepoint_link": deal_data.get("sharepoint_link"),
+                    "dampmould_growth": deal_data.get("dampmould_growth"),
+                    "pre_sap": deal_data.get("pre_sap"),
+                    "coordinator": deal_data.get("coordinator"),
+                    "mtp_completion_date": self._parse_hs_date(deal_data.get("mtp_completion_date")),
+                    "mtp_re_model_completion_date": self._parse_hs_date(deal_data.get("mtp_re_model_completion_date")),
+                    "ioe_v3_completion_date": self._parse_hs_date(deal_data.get("ioe_v3_completion_date")),
+                    "proposed_measures": deal_data.get("proposed_measures"),
+                    "approved_package": deal_data.get("approved_package"),
+                    "designer": deal_data.get("designer"),
+                    "design_completion_date": self._parse_hs_date(deal_data.get("design_completion_date")),
+                    "actual_measures_installed": deal_data.get("actual_measures_installed"),
+                    "installer": deal_data.get("installer"),
+                    "installer_handover": deal_data.get("installer_handover"),
+                    "lodgement_status": deal_data.get("lodgement_status"),
+                    "measures_lodgement_date": self._parse_hs_date(deal_data.get("measures_lodgement_date")),
+                    "lodgement_date": self._parse_hs_date(deal_data.get("lodgement_date")),
+                    "expected_commencement_date": self._parse_hs_date(deal_data.get("expected_commencement_date")),
+                    "surveyor": deal_data.get("surveyor"),
+                    "confirmed_survey_date": self._parse_hs_date(deal_data.get("confirmed_survey_date")),
+                    "confirmed_survey_time": deal_data.get("confirmed_survey_time"),
+                    "surveyed_date": self._parse_hs_date(deal_data.get("surveyed_date")),
+                    "design_type": deal_data.get("design_type"),
                 }.items():
                     setattr(existing, attr, value or getattr(existing, attr))
 
@@ -288,6 +418,9 @@ class HubspotDataToDb:
                                 f"⚠️ Failed to download photo for deal_id {existing.deal_id}: {e}"
                             )
                             # Continue without the file — don't crash the update
+                        finally:
+                            if "local_file" in locals() and os.path.exists(local_file):
+                                os.remove(local_file)
                     else:
                         print(f"⚠️ Photo URL missing for deal_id {existing.deal_id}")
 
@@ -302,6 +435,7 @@ class HubspotDataToDb:
                     deal_id=deal_id,
                     dealname=deal_data.get("dealname"),
                     dealstage=deal_data.get("dealstage"),
+                    listing_id=listing.get("listing_id"),
                     landlord_property_id=listing.get("owner_property_id"),
                     uprn=listing.get("national_uprn"),
                     outcome=deal_data.get("outcome"),
@@ -316,6 +450,30 @@ class HubspotDataToDb:
                     ),
                     coordination_status=deal_data.get("coordination_status__stage_1_"),
                     design_status=deal_data.get("retrofit_design_status"),
+                    pashub_link=deal_data.get("pashub_link"),
+                    sharepoint_link=deal_data.get("sharepoint_link"),
+                    dampmould_growth=deal_data.get("dampmould_growth"),
+                    pre_sap=deal_data.get("pre_sap"),
+                    coordinator=deal_data.get("coordinator"),
+                    mtp_completion_date=self._parse_hs_date(deal_data.get("mtp_completion_date")),
+                    mtp_re_model_completion_date=self._parse_hs_date(deal_data.get("mtp_re_model_completion_date")),
+                    ioe_v3_completion_date=self._parse_hs_date(deal_data.get("ioe_v3_completion_date")),
+                    proposed_measures=deal_data.get("proposed_measures"),
+                    approved_package=deal_data.get("approved_package"),
+                    designer=deal_data.get("designer"),
+                    design_completion_date=self._parse_hs_date(deal_data.get("design_completion_date")),
+                    actual_measures_installed=deal_data.get("actual_measures_installed"),
+                    installer=deal_data.get("installer"),
+                    installer_handover=deal_data.get("installer_handover"),
+                    lodgement_status=deal_data.get("lodgement_status"),
+                    measures_lodgement_date=self._parse_hs_date(deal_data.get("measures_lodgement_date")),
+                    lodgement_date=self._parse_hs_date(deal_data.get("lodgement_date")),
+                    expected_commencement_date=self._parse_hs_date(deal_data.get("expected_commencement_date")),
+                    surveyor=deal_data.get("surveyor"),
+                    confirmed_survey_date=self._parse_hs_date(deal_data.get("confirmed_survey_date")),
+                    confirmed_survey_time=deal_data.get("confirmed_survey_time"),
+                    surveyed_date=self._parse_hs_date(deal_data.get("surveyed_date")),
+                    design_type=deal_data.get("design_type"),
                 )
 
                 # Handle upload at insert time
@@ -335,6 +493,9 @@ class HubspotDataToDb:
                             f"⚠️ Failed to download photo for deal_id {new_record.deal_id}: {e}"
                         )
                         # Continue without the file — don't crash the insert
+                    finally:
+                        if "local_file" in locals() and os.path.exists(local_file):
+                            os.remove(local_file)
 
                 session.add(new_record)
                 session.commit()

etl/hubspot/scripts/scraper/bulk_load.py

@@ -0,0 +1,41 @@
+from etl.hubspot.hubspotClient import HubspotClient, Companies, Pipeline
+from etl.hubspot.scripts.scraper.main import handler
+from tqdm import tqdm
+import json
+
+
+PIPELINE_ID = Pipeline.OPERATIONS_SOCIAL_HOUSING.value
+
+companies = list([Companies.THE_GUINESS_PARTNERSHIP, Companies.SOUTHERN_HOUSING_GROUP])
+
+
+def bulk_load(companies: list[Companies] | None = None) -> None:
+    """
+    Load all deals from the given companies (defaults to all Companies enum values)
+    into the database, filtered to the Operations/Social Housing pipeline.
+    """
+    hubspot = HubspotClient()
+    targets = companies or list(Companies)
+
+    for company in tqdm(targets, desc="Companies", unit="co"):
+        company_id = company.value
+        deal_ids = hubspot.get_deal_ids_from_company(company_id)
+
+        processed = 0
+        with tqdm(deal_ids, desc=company.name, unit="deal", leave=False) as deal_bar:
+            for deal_id in deal_bar:
+                deal_data = hubspot.from_deal_id_get_info(deal_id)
+                if deal_data.get("pipeline") != PIPELINE_ID:
+                    deal_bar.set_postfix({"status": "skip", "deal": deal_id})
+                    continue
+
+                deal_bar.set_postfix({"status": "uploading", "deal": deal_id})
+                handler({"Records": [{"body": json.dumps({"hubspot_deal_id": deal_id})}]}, context=None)
+                processed += 1
+                deal_bar.set_postfix({"status": "done", "deal": deal_id})
+
+        tqdm.write(f"[{company.name}] {processed}/{len(deal_ids)} deals in pipeline")
+
+
+if __name__ == "__main__":
+    bulk_load(companies)

etl/hubspot/scripts/scraper/main.py

@@ -4,21 +4,18 @@
 3) [completed] Load the db and check if upsert it into the table
 4) [completed]Getting working on a AWS lambda
 5) [completed] subtask and tasks history
-6) [TODO]The new sexy deal properties, move it over
+6) [completed]The new sexy deal properties, move it over
 """
 
 from etl.hubspot.hubspotClient import HubspotClient
 from etl.hubspot.hubspotDataTodB import HubspotDataToDb
+from backend.utils.subtasks import task_handler
 from typing import Any
+import json
 
 
-# @subtask_handler() TODO: Do this without subtask_handler but task_handler() that creates task_id and subtask_id
-def handler(body: dict[str, Any], context: Any, local: bool = False) -> None:
-    if local is True:
-        body = {
-            "hubspot_deal_id": "254427203793",
-        }
-
+@task_handler()
+def handler(body: dict[str, Any], context: Any) -> None:
     hubspot_deal_id = body.get("hubspot_deal_id", "")
 
     if hubspot_deal_id == "":
@@ -36,5 +33,3 @@ def handler(body: dict[str, Any], context: Any, local: bool = False) -> None:
     else:
         deal, company, listing = hubspot.get_deal_info_for_db(hubspot_deal_id)
         dbloader.upsert_deal(deal, company, listing, hubspot)
-
-    print("Finsihed running")

infrastructure/terraform/lambda/hubspot_deal_etl/main.tf

@@ -35,10 +35,14 @@ module "hubspot_deal_etl" {
     LOG_LEVEL = "info"
     DB_USERNAME = local.db_credentials.db_assessment_model_username
     DB_PASSWORD = local.db_credentials.db_assessment_model_password
+    DB_HOST     = var.db_host
+    DB_NAME     = var.db_name
+    DB_PORT     = var.db_port
+    HUBSPOT_API_KEY = var.hubspot_api_key
   }
 }
 
 resource "aws_iam_role_policy_attachment" "lambda_s3_policy" {
-  role       = module.lambda.role_name
+  role       = module.hubspot_deal_etl.role_name
   policy_arn = data.terraform_remote_state.shared.outputs.hubspot_etl_s3_read_and_write_arn
 }

infrastructure/terraform/lambda/hubspot_deal_etl/variables.tf

@@ -41,6 +41,11 @@ variable "db_host" {
   type = string
 }
 
+variable "hubspot_api_key" {
+  type = string
+}
+
+
 variable "db_name" {
   type = string
 }

sfr/principal_pitch/2_export_data.py

@@ -26,13 +26,13 @@ from backend.app.db.functions.materials_functions import get_materials
 from collections import defaultdict
 from sqlalchemy import func
 
-PORTFOLIO_ID = 639
-SCENARIOS = [1157]
+PORTFOLIO_ID = 656
+SCENARIOS = [1177]
 scenario_names = {
-    1157: "EPC C - no EWI solid floor",
+    1177: "EPC C; Proposed Measures",
 }
 
-project_name = "Instagroup Sample"
+project_name = "Walsall Council | WH:LG"
 
 
 def get_data(portfolio_id, scenario_ids):