init db

Pashub to ara: save files to s3 and record in new database table

.github/workflows/deploy_terraform.yml

@@ -496,17 +496,10 @@ jobs:
       ecr_repo: hubspot-etl-${{ needs.determine_stage.outputs.stage }}
       dockerfile_path: etl/hubspot/scripts/scraper/handler/Dockerfile
       build_context: .
-      build_args: |
-        DEV_DB_HOST=$DEV_DB_HOST
-        DEV_DB_PORT=$DEV_DB_PORT
-        DEV_DB_NAME=$DEV_DB_NAME
     secrets:
       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 }}
-      DEV_DB_HOST: ${{ secrets.DEV_DB_HOST }}
-      DEV_DB_PORT: ${{ secrets.DEV_DB_PORT }}
-      DEV_DB_NAME: ${{ secrets.DEV_DB_NAME }}
 
   # ============================================================
   # Deploy Hubspot ETL Lambda
@@ -522,6 +515,9 @@ jobs:
       image_digest: ${{ needs.hubspot_etl_image.outputs.image_digest }}
       terraform_apply: ${{ needs.determine_stage.outputs.terraform_apply }}
     secrets:
+      TF_VAR_db_host: ${{ secrets.DEV_DB_HOST }}
+      TF_VAR_db_name: ${{ secrets.DEV_DB_NAME }}
+      TF_VAR_db_port: ${{ secrets.DEV_DB_PORT }}
       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 }}

.github/workflows/unit_tests.yml

@@ -11,6 +11,21 @@ jobs:
     name: Tests (Docker)
     runs-on: ubuntu-latest
 
+    services:
+      postgres:
+        image: postgres:15
+        env:
+          POSTGRES_USER: test
+          POSTGRES_PASSWORD: test
+          POSTGRES_DB: test
+        ports:
+          - 5432:5432
+        options: >-
+          --health-cmd pg_isready
+          --health-interval 10s
+          --health-timeout 5s
+          --health-retries 5
+
     steps:
       - name: Checkout code
         uses: actions/checkout@v4
@@ -18,9 +33,26 @@ jobs:
       - name: Build test image
         run: docker build -f Dockerfile.test -t model-test .
 
+      - name: Initialise database schema
+        run: |
+          docker run --rm \
+            --network host \
+            -e DB_HOST=localhost \
+            -e DB_NAME=test \
+            -e DB_USERNAME=test \
+            -e DB_PASSWORD=test \
+            -e DB_PORT=5432 \
+            model-test python scripts/init_db.py
+
       - name: Run tests
         run: |
           docker run --rm \
+            --network host \
             -e EPC_AUTH_TOKEN=${{ secrets.DEV_EPC_AUTH_TOKEN }} \
             -e HUBSPOT_API_KEY=${{ secrets.HUBSPOT_API_KEY }} \
-            model-test pytest -m 'not integration'
+            -e DB_HOST=localhost \
+            -e DB_NAME=test \
+            -e DB_USERNAME=test \
+            -e DB_PASSWORD=test \
+            -e DB_PORT=5432 \
+            model-test pytest -vv -m 'not integration'

backend/app/config.py

@@ -66,6 +66,18 @@ class Settings(BaseSettings):
     ORDNANCE_SURVEY_API_KEY: str = "changeme"
 
     HUBSPOT_API_KEY: Optional[str] = None
+    # Sharepoint
+    SHAREPOINT_CLIENT_ID: Optional[str] = None
+    SHAREPOINT_CLIENT_SECRET: Optional[str] = None
+    SHAREPOINT_TENANT_ID: Optional[str] = None
+    DOMNA_SHAREPOINT_ID: Optional[str] = None
+    OSMOSIS_ACD_SHAREPOINT_ID: Optional[str] = None
+    PRIVATE_PAY_SHAREPOINT_ID: Optional[str] = None
+    SOCIAL_HOUSING_WAVE_3_SHAREPOINT_ID: Optional[str] = None
+
+    # Pas Hub
+    PASHUB_EMAIL: Optional[str] = None
+    PASHUB_PASSWORD: Optional[str] = None
 
     # Optional AWS creds (only required in local)
     AWS_ACCESS_KEY_ID: Optional[str] = None
@@ -91,7 +103,6 @@ def get_prediction_buckets():
         "carbon_change_predictions": get_settings().CARBON_PREDICTIONS_BUCKET,
         "heating_kwh_predictions": get_settings().HEATING_KWH_PREDICTIONS_BUCKET,
         "hotwater_kwh_predictions": get_settings().HOTWATER_KWH_PREDICTIONS_BUCKET,
-
         # Score model - SAP re-baselining model
         "retrofit_sap_baseline_predictions": get_settings().SAP_BASELINE_PREDICTIONS_BUCKET,
         "retrofit_carbon_baseline_predictions": get_settings().CARBON_BASELINE_PREDICTIONS_BUCKET,

backend/app/db/models/organisation.py

@@ -43,7 +43,7 @@ class HubspotDealData(SQLModel, table=True):
     created_at: datetime = Field(
         sa_column=Column(
             DateTime(timezone=True),
-            server_default=text("NOW() AT TIME ZONE 'utc'"),
+            server_default=text("(NOW() AT TIME ZONE 'utc')"),
             nullable=False,
         )
     )
@@ -51,7 +51,7 @@ class HubspotDealData(SQLModel, table=True):
     updated_at: datetime = Field(
         sa_column=Column(
             DateTime(timezone=True),
-            server_default=text("NOW() AT TIME ZONE 'utc'"),
+            server_default=text("(NOW() AT TIME ZONE 'utc')"),
             onupdate=func.now(),
             nullable=False,
         )

backend/app/db/models/uploaded_file.py

@@ -0,0 +1,44 @@
+import enum
+from sqlalchemy import TIMESTAMP, BigInteger, Column, Text, Enum as SqlEnum
+
+from backend.app.db.base import Base
+
+
+class FileTypeEnum(enum.Enum):
+    PHOTO_PACK = "photo_pack"
+    SITE_NOTE = "site_note"
+    RD_SAP_SITE_NOTE = "rd_sap_site_note"
+    PAS_2023_VENTILATION = "pas_2023_ventilation"
+    PAS_2023_CONDITION = "pas_2023_condition"
+    PAS_SIGNIFICANCE = "pas_significance"
+    PAR_PHOTO_PACK = "par_photo_pack"
+    PAS_2023_PROPERTY = "pas_2023_property"
+    PAS_2023_OCCUPANCY = "pas_2023_occupancy"
+
+
+class FileSourceEnum(enum.Enum):
+    PAS_HUB = "pas hub"
+    SHAREPOINT = "sharepoint"
+    HUBSPOT = "hubspot"
+
+
+class UploadedFile(Base):
+    __tablename__ = "uploaded_files"
+
+    id = Column(BigInteger, primary_key=True, autoincrement=True)
+
+    s3_file_bucket = Column(Text, nullable=False)
+    s3_file_key = Column(Text, nullable=False)
+    s3_upload_timestamp = Column(TIMESTAMP(timezone=True), nullable=False)
+
+    landlord_property_id = Column(Text, nullable=True)
+    uprn = Column(BigInteger, nullable=True)
+    hubspot_listing_id = Column(BigInteger, nullable=True)
+
+    file_type = Column(
+        SqlEnum(FileTypeEnum, name="file_type", create_type=False), nullable=True
+    )
+
+    file_source = Column(
+        SqlEnum(FileSourceEnum, name="file_source", create_type=False), nullable=True
+    )

backend/app/plan/router.py

@@ -174,6 +174,8 @@ async def trigger_plan_entrypoint(body: PlanTriggerRequest):
                 service="plan_engine",
                 inputs=data,
                 task_only=True,
+                source=SourceEnum.PORTFOLIO,
+                source_id=str(body.portfolio_id),
             )
 
             subtask_interface = SubTaskInterface()
@@ -222,6 +224,8 @@ async def trigger_plan_entrypoint(body: PlanTriggerRequest):
                 service="plan_engine",
                 inputs=data,
                 task_only=False,
+                source=SourceEnum.PORTFOLIO,
+                source_id=str(body.portfolio_id),
             )
             data["task_id"] = str(task_id)
             data["subtask_id"] = str(subtask_id)

backend/export/property_scenarios/main.py

@@ -134,10 +134,18 @@ def handler(event: Mapping[str, Any], context: Optional[Any]) -> Mapping[str, Un
     body_dict = {
         "task_id": "test",
         "subtask_id": "test",
-        "portfolio_id": 569,
+        "portfolio_id": 655,
         "scenario_ids": [],
         "default_plans_only": True,
     }
+
+    body_dict = {
+        "task_id": "test",
+        "subtask_id": "test",
+        "portfolio_id": 655,
+        "scenario_ids": [1174],
+        "default_plans_only": False,
+    }
     :param event: Lambda event containing export request details
     :param context: Lambda context (not used in this handler but included for completeness)
     :return: HTTP response indicating success or failure of the export operation

backend/export/tests/conftest.py

@@ -2,6 +2,8 @@ import pytest
 from sqlalchemy import create_engine
 from sqlalchemy.orm import sessionmaker
 from backend.app.db.base import Base
+from sqlmodel import SQLModel
+import backend.app.db.models.organisation  # noqa: F401 — registers Organisation with SQLModel.metadata
 
 
 @pytest.fixture(scope="function")
@@ -25,12 +27,14 @@ def engine(postgresql):
 
     # Create tables once per test session
     Base.metadata.create_all(engine)
+    SQLModel.metadata.create_all(engine)
 
     # Yeild will split this function into two phase. 1) setup and 2) teardown, the latter of which will run after all
     # tests have completed
     yield engine
 
     # Clean-up after entire test session
+    SQLModel.metadata.drop_all(engine)
     Base.metadata.drop_all(engine)
     engine.dispose()
 

backend/pashub_fetcher/core_files.py

@@ -1,4 +1,7 @@
 from enum import Enum
+from typing import Optional
+
+from backend.app.db.models.uploaded_file import FileTypeEnum
 
 
 class CoreFiles(Enum):
@@ -11,3 +14,23 @@ class CoreFiles(Enum):
     PAR_PHOTOPACK = "PAR Photo Pack"
     PAS2023_PROPERTY = "PAS 2023 Property Assessment Report"
     PAS2023_OCCUPANCY = "PAS 2023 Occupancy Assessment Report"
+
+
+CORE_TO_FILETYPE_MAP = {
+    CoreFiles.PHOTOPACK: FileTypeEnum.PHOTO_PACK.value,
+    CoreFiles.SITENOTE: FileTypeEnum.SITE_NOTE.value,
+    CoreFiles.RDSAP_SITENOTE: FileTypeEnum.RD_SAP_SITE_NOTE.value,
+    CoreFiles.PAS2023_VENTILATION: FileTypeEnum.PAS_2023_VENTILATION.value,
+    CoreFiles.PAS2023_CONDITION: FileTypeEnum.PAS_2023_CONDITION.value,
+    CoreFiles.PAS_SIGNIFICANCE: FileTypeEnum.PAS_SIGNIFICANCE.value,
+    CoreFiles.PAR_PHOTOPACK: FileTypeEnum.PAR_PHOTO_PACK.value,
+    CoreFiles.PAS2023_PROPERTY: FileTypeEnum.PAS_2023_PROPERTY.value,
+    CoreFiles.PAS2023_OCCUPANCY: FileTypeEnum.PAS_2023_OCCUPANCY.value,
+}
+
+
+def infer_file_type(filename: str) -> Optional[str]:
+    for core_file, file_type in CORE_TO_FILETYPE_MAP.items():
+        if filename.startswith(core_file.value):
+            return file_type
+    return None

backend/pashub_fetcher/handler/handler.py

@@ -1,13 +1,22 @@
+from datetime import datetime, timezone
 import os
 import re
-from typing import Any, Dict, List, Mapping
+from typing import Any, Dict, List, Mapping, Optional
 from openpyxl import load_workbook
 
+from backend.app.config import get_settings
+from backend.app.db.connection import db_session
+from backend.app.db.models.uploaded_file import (
+    FileSourceEnum,
+    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.sharepoint_subfolders import SharepointSubfolders
 from backend.pashub_fetcher.token_getter import get_token_from_local_storage
 from utils.logger import setup_logger
+from utils.s3 import upload_file_to_s3
 from utils.sharepoint.domna_sharepoint_client import DomnaSharepointClient
 from utils.sharepoint.domna_sites import DomnaSites
 
@@ -17,7 +26,8 @@ logger = setup_logger()
 
 def extract_jobs(filepath: str) -> List[Job]:
     wb = load_workbook(filepath, data_only=True)
-    ws = wb["watford warm homes (wave 3) mai"]
+    # ws = wb["watford warm homes (wave 3) mai"]
+    ws = wb["filtered"]
 
     HEADER_ROW = 3
 
@@ -28,7 +38,8 @@ def extract_jobs(filepath: str) -> List[Job]:
             headers[value.strip()] = col
 
     name_col = headers["Name"]
-    link_col = headers["Pashub Link"]
+    # link_col = headers["Pashub Link"]
+    link_col = headers["PasHub Link"]
 
     jobs: List[Job] = []
 
@@ -39,88 +50,170 @@ def extract_jobs(filepath: str) -> List[Job]:
         if not name or not link:
             continue
 
-        link = str(link)
-
-        match = re.search(r"/jobs/([0-9a-fA-F\-]+)/", link)
+        match = re.search(r"/jobs/([0-9a-fA-F\-]+)/", str(link))
         if not match:
             continue
 
-        job_id = match.group(1)
-
-        jobs.append({"id": job_id, "address": str(name)})
+        jobs.append(
+            {
+                "id": match.group(1),
+                "address": str(name),
+            }
+        )
 
     return jobs
 
 
+def get_pashub_client(email: str, password: str) -> PashubClient:
+    token = get_token_from_local_storage(email, password)
+    logger.info("Token extracted successfully")
+    return PashubClient(token=token)
+
+
+def upload_job_to_sharepoint(
+    sharepoint_client: DomnaSharepointClient,
+    base_path: str,
+    job: Job,
+    job_files: List[str],
+) -> None:
+    job_path = f"{base_path}/{job['address']}"
+
+    # Create main job folder
+    sharepoint_client.makedir(job["address"], base_path)
+
+    # Create subfolders
+    for folder in SharepointSubfolders:
+        sharepoint_client.makedir(folder.value, job_path)
+
+    # Upload into assessment folder
+    assessment_path = f"{job_path}/{SharepointSubfolders.ASSESSMENT.value}"
+
+    for file_path in job_files:
+        filename = file_path.split("/")[-1]
+
+        sharepoint_client.upload_file(
+            file_path,
+            assessment_path,
+            filename,
+        )
+
+
+def upload_job_to_s3_and_update_db(job_files: List[str], uprn: str) -> None:
+    bucket = "retrofit-energy-assessments-dev"
+
+    base_path = f"documents/uprn/{uprn}"
+
+    uploaded_files: List[UploadedFile] = []
+
+    for file_path in job_files:
+        filename = os.path.basename(file_path)
+        file_key = f"{base_path}/{filename}"
+
+        upload_file_to_s3(file_path, bucket, file_key)
+
+        # load row to db
+        uploaded_files.append(
+            UploadedFile(
+                s3_file_bucket=bucket,
+                s3_file_key=file_key,
+                s3_upload_timestamp=datetime.now(timezone.utc),
+                uprn=int(uprn),
+                file_source=FileSourceEnum.PAS_HUB.value,
+                file_type=infer_file_type(filename),
+            )
+        )
+
+    with db_session() as session:
+        session.add_all(uploaded_files)
+        session.commit()
+
+    pass
+
+
+def process_job(
+    job: Job,
+    pashub_client: PashubClient,
+    sharepoint_client: DomnaSharepointClient,
+    base_path: str,
+) -> List[str]:
+    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:
+        logger.info(f"No UPRN found for job {job_id}")
+
+    job_files: List[str] = pashub_client.get_core_evidence_files_by_job_id(job_id)
+
+    if uprn:
+        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)
+
+    return job_files
+
+
 def handler(event: Mapping[str, Any], context: Any) -> None:
+    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, "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 = "random@test.com"
-    pas_hub_password = "my_fake_password"
+    pas_hub_email = settings.PASHUB_EMAIL
+    pas_hub_password = settings.PASHUB_PASSWORD
 
-    try:
-        token: str = get_token_from_local_storage(pas_hub_email, pas_hub_password)
-        logger.info(f"Token extracted successfully")
-    except:
-        logger.error("Error getting auth token from Pas Hub")
-        raise
+    if (not pas_hub_email) or (not pas_hub_password):
+        raise ValueError("Pas Hub credentials not provided")
+
+    pashub_client = get_pashub_client(pas_hub_email, pas_hub_password)
 
-    pashub_client = PashubClient(token=token)
     sharepoint_client = DomnaSharepointClient(
         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] = []
-    BASE_PATH = "/Osmosis-ACD Projects/Watford Warm Homes/Watford Property Folders (Shared with Client)"  # TODO: get from request body
 
     for job in jobs:
         try:
-            job_files: List[str] = pashub_client.get_core_evidence_files_by_job_id(
-                job["id"]
+            files = process_job(
+                job,
+                pashub_client,
+                sharepoint_client,
+                BASE_PATH,
             )
-
-            # Upload files to sharepoint
-            job_path = f"{BASE_PATH}/{job['address']}"
-
-            sharepoint_client.makedir(job["address"], BASE_PATH)
-
-            for folder in SharepointSubfolders:
-                sharepoint_client.makedir(folder.value, job_path)
-
-            assessment_path = f"{job_path}/{SharepointSubfolders.ASSESSMENT.value}"
-
-            for file_path in job_files:
-                filename = file_path.split("/")[-1]
-
-                sharepoint_client.upload_file(
-                    file_path,
-                    assessment_path,
-                    filename,
-                )
-
-            saved_file_paths.extend(job_files)
+            saved_file_paths.extend(files)
 
         except UnauthorizedError:
             logger.warning("Token expired - refreshing")
 
-            token = get_token_from_local_storage(pas_hub_email, pas_hub_password)
-
-            pashub_client = PashubClient(token=token)
-
-            # retry once
-            saved_file_paths.extend(
-                pashub_client.get_core_evidence_files_by_job_id(job["id"])
+            pashub_client = get_pashub_client(
+                pas_hub_email,
+                pas_hub_password,
             )
 
-    print(f"saved {len(saved_file_paths)} files")
+            # retry once
+            files = process_job(
+                job,
+                pashub_client,
+                sharepoint_client,
+                BASE_PATH,
+            )
+            saved_file_paths.extend(files)
+
+    logger.info(f"Saved {len(saved_file_paths)} files")
 
 
 if __name__ == "__main__":
     event = {"Records": [{"body": "{}"}]}
-
     handler(event, None)

backend/pashub_fetcher/handler/test_handler.py

@@ -1,7 +1,23 @@
 from typing import Any, Mapping
 import json
 
+from utils.logger import setup_logger
+
+logger = setup_logger()
+
 
 def handler(event: Mapping[str, Any], context: Any) -> None:
-    print("Received event:")
-    print(json.dumps(event, indent=2))
+    logger.info("Received event:")
+
+    for record in event.get("Records", []):
+        body_str = record.get("body", "")
+
+        try:
+            body_obj = json.loads(body_str)
+        except json.JSONDecodeError:
+            logger.error("Failed to parse body as JSON")
+            logger.info(body_str)
+            continue
+
+        logger.info("Parsed message body:")
+        logger.info(json.dumps(body_obj, indent=2))

backend/pashub_fetcher/pashub_client.py

@@ -71,6 +71,21 @@ class PashubClient:
 
         return saved_files
 
+    def get_uprn_by_job_id(self, job_id: str) -> Optional[str]:
+        logger.info(f"Getting UPRN for job ID {job_id}")
+        url = f"{self.base}/jobs/{job_id}"
+
+        r = self.session.get(url)
+        if r.status_code == 401:
+            raise UnauthorizedError("Token expired or invalid")
+
+        r.raise_for_status()
+
+        try:
+            return r.json()["uprn"]
+        except Exception:
+            return None
+
     def _get_core_file_type(self, file: EvidenceFileData) -> Optional[CoreFiles]:
         for core_file in CoreFiles:
             if file.file_name.startswith(core_file.value):

backend/pashub_fetcher/pashub_to_ara_trigger_request.py

@@ -0,0 +1,18 @@
+from typing import Optional
+from pydantic import BaseModel
+
+
+class PashubToAraTriggerRequest(BaseModel):
+    pashub_link: (
+        str  # e.g. https://pashub.net/jobs/12345-abcd-1234-abcd-12345abcde/details
+    )
+
+    address: Optional[str] = None
+    sharepoint_link: Optional[str] = None
+    uprn: Optional[str] = None
+    landlord_property_id: Optional[str] = None
+    deal_stage: Optional[str] = None
+
+    @property
+    def pashub_job_id(self) -> str:
+        return self.pashub_link.split("/")[-2]

backend/tests/test_rebaselining_pipeline.py

@@ -2,6 +2,10 @@ import os
 import pickle
 import pandas as pd
 import pytest
+from datetime import datetime
+from backend.ml_models.api import ModelApi
+from backend.app.utils import sap_to_epc
+from backend.app.config import get_prediction_buckets
 
 
 def load_sample_certificates():
@@ -60,12 +64,6 @@ def load_cleaning_data():
 
 @pytest.mark.integration
 def test_rebaselining_pipeline_with_real_data():
-    import pandas as pd
-    from datetime import datetime
-    from backend.ml_models.api import ModelApi
-    from backend.app.utils import sap_to_epc
-    from backend.app.config import get_prediction_buckets
-
     df = load_sample_certificates()
     cleaning_data = load_cleaning_data()
     input_properties = [make_property_from_row(row, cleaning_data=cleaning_data) for _, row in df.iterrows()]

etl/hubspot/hubspotClient.py

@@ -25,7 +25,7 @@ from hubspot.crm.associations.v4.models import (  # type: ignore[reportMissingTy
     ForwardPaging as AssociationsPaging,
     NextPage as AssociationsPagingNext,
 )
-from etl.hubspot.hubspotDataTodB import CompanyData
+from etl.hubspot.hubspotDataTodB import CompanyData, HubspotDataToDb
 
 
 from backend.app.config import get_settings
@@ -217,8 +217,15 @@ class HubspotClient:
     def get_deal_info_for_db(
         self, deal_id: str
     ) -> tuple[dict[str, str], Optional[str], Optional[dict[str, str]]]:
+
         deal: dict[str, str] = self.from_deal_id_get_info(deal_id)
         company: Optional[str] = self.from_deal_id_get_associated_company_id(deal_id)
+
+        if company:
+            company_data: CompanyData = self.get_company_information(company)
+            dbloader: HubspotDataToDb = HubspotDataToDb()
+            dbloader.upsert_company(company_data)
+
         listing: Optional[dict[str, str]] = self.from_deal_id_get_associated_listing(
             deal_id
         )

etl/hubspot/hubspotDataTodB.py

@@ -63,7 +63,7 @@ class HubspotDataToDb:
 
     def new_record_to_hubspot_data(self, deal_data, company, listing, hubspot_client):
         print("⚠️ Deprecated — use the new interface instead.")
-        return self.upsert_hubspot_deal(deal_data, company, listing, hubspot_client)
+        return self.upsert_deal(deal_data, company, listing, hubspot_client)
 
     def find_all_deals_with_company_id(self, company_id):
         """Returns a list of deals for a given company_id."""
@@ -90,7 +90,7 @@ class HubspotDataToDb:
                 sha256.update(chunk)
         return sha256.hexdigest()
 
-    def update_deal(self, deal_in_db, hubspot_client):
+    def update_deal_with_checks(self, deal_in_db, hubspot_client) -> bool:
         """
         Checks if a deal needs updating and syncs it with HubSpot.
         Also handles major_condition_issue_photos file upload to S3 with integrity check.
@@ -164,7 +164,7 @@ class HubspotDataToDb:
             print(
                 f"❗ Discrepancies found for deal_id {deal_in_db.deal_id} — syncing with HubSpot."
             )
-            self.upsert_hubspot_deal(hs_deal, hs_company_id, hs_listing, hubspot_client)
+            self.upsert_deal(hs_deal, hs_company_id, hs_listing, hubspot_client)
             return False
 
         # Handle photo upload if it exists but S3 URL is missing
@@ -219,7 +219,7 @@ class HubspotDataToDb:
 
         return True
 
-    def upsert_hubspot_deal(self, deal_data, company, listing, hubspot_client):
+    def upsert_deal(self, deal_data, company, listing, hubspot_client):
         """
         Inserts or updates a deal record.
         Also uploads photos if present and adds S3 URL.

etl/hubspot/s3_uploader.py

@@ -13,8 +13,8 @@ class S3Uploader:
 
     def __init__(
         self,
-        aws_access_key: str = "AKIAU5A36PPNK7RXX52V",
-        aws_secret_key: str = "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",
+        aws_access_key: str,
+        aws_secret_key: str,
         region: str = "eu-west-2",
     ):
         self.aws_access_key = aws_access_key

etl/hubspot/scripts/onboarding/new_organisation.py

@@ -1,3 +1,13 @@
+"""
+README.md
+
+This is a simple script to showcase how a new organisation can be
+added to AraDb.
+
+This has been made reduntant due to doing this process when ever
+hubspot has a webhook
+"""
+
 from etl.hubspot.hubspotClient import HubspotClient, Companies
 
 from etl.hubspot.hubspotDataTodB import HubspotDataToDb, CompanyData

etl/hubspot/scripts/scraper/handler/Dockerfile

@@ -1,16 +1,6 @@
 FROM public.ecr.aws/lambda/python:3.10
 # FROM python:3.11.10-bullseye
 
-
-ARG DEV_DB_HOST
-ARG DEV_DB_PORT
-ARG DEV_DB_NAME
-
-ENV DB_HOST=${DEV_DB_HOST}
-ENV DB_PORT=${DEV_DB_PORT}
-ENV DB_NAME=${DEV_DB_NAME}
-
-
 # Set working directory (Lambda task root)
 WORKDIR /var/task
 

etl/hubspot/scripts/scraper/main.py

@@ -7,18 +7,15 @@
 6) [TODO]The new sexy deal properties, move it over
 """
 
-from backend.utils.subtasks import subtask_handler
 from etl.hubspot.hubspotClient import HubspotClient
 from etl.hubspot.hubspotDataTodB import HubspotDataToDb
 from typing import Any
 
 
-@subtask_handler()
+# @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 = {
-            "task_id": "e31f2f21-175b-4a91-a3ec-a6baa325e917",
-            "sub_task_id": "8673913b-1a88-42d7-8578-0449123d94b0",
             "hubspot_deal_id": "254427203793",
         }
 
@@ -29,15 +26,15 @@ def handler(body: dict[str, Any], context: Any, local: bool = False) -> None:
             "Missing Hubspot Deal ID in SQS body request, 'hubspot_deal_id'"
         )
 
-    hubspot = HubspotClient()
-    dbloader = HubspotDataToDb()
+    hubspot: HubspotClient = HubspotClient()
+    dbloader: HubspotDataToDb = HubspotDataToDb()
 
     deal = dbloader.find_deal_with_deal_id(hubspot_deal_id)
 
     if deal:
-        dbloader.update_deal(deal, hubspot)
+        dbloader.update_deal_with_checks(deal, hubspot)
     else:
         deal, company, listing = hubspot.get_deal_info_for_db(hubspot_deal_id)
-        dbloader.upsert_hubspot_deal(deal, company, listing, hubspot)
+        dbloader.upsert_deal(deal, company, listing, hubspot)
 
     print("Finsihed running")

infrastructure/terraform/lambda/_template/variables.tf

@@ -35,3 +35,4 @@ locals {
 output "resolved_image_uri" {
   value = local.image_uri
 }
+

infrastructure/terraform/lambda/hubspot_deal_etl/main.tf

@@ -0,0 +1,44 @@
+data "terraform_remote_state" "shared" {
+  backend = "s3"
+  config = {
+    bucket = "assessment-model-terraform-state"
+    key = "env:/${var.stage}/terraform.tfstate"
+    region = "eu-west-2"
+  }
+}
+
+
+data "aws_secretsmanager_secret_version" "db_credentials" {
+  secret_id = "${var.stage}/assessment_model/db_credentials"
+}
+
+locals {
+  db_credentials = jsondecode(data.aws_secretsmanager_secret_version.db_credentials.secret_string)
+}
+
+
+module "hubspot_deal_etl" {
+  source = "../../modules/lambda_with_sqs"
+
+  name  = "hubspot_deal_etl"
+  stage = var.stage
+
+  image_uri = local.image_uri
+
+  # Optional: Set maximum_concurrency to limit concurrent SQS-triggered invocations (2-1000)
+  maximum_concurrency = var.maximum_concurrency
+
+  batch_size = var.batch_size
+
+  environment = {
+    STAGE = var.stage
+    LOG_LEVEL = "info"
+    DB_USERNAME = local.db_credentials.db_assessment_model_username
+    DB_PASSWORD = local.db_credentials.db_assessment_model_password
+  }
+}
+
+resource "aws_iam_role_policy_attachment" "lambda_s3_policy" {
+  role       = module.lambda.role_name
+  policy_arn = data.terraform_remote_state.shared.outputs.hubspot_etl_s3_read_and_write_arn
+}

infrastructure/terraform/lambda/hubspot_deal_etl/provider.tf

@@ -0,0 +1,16 @@
+terraform {
+  required_providers {
+    aws = {
+      source  = "hashicorp/aws"
+      version = ">= 5.0"
+    }
+  }
+
+  backend "s3" {
+    bucket = "hubspot-etl-bucket-terraform-state"
+    key    = "terraform.tfstate"
+    region = "eu-west-2"
+  }
+
+  required_version = ">= 1.2.0"
+}

infrastructure/terraform/lambda/hubspot_deal_etl/variables.tf

@@ -0,0 +1,50 @@
+variable "lambda_name" {
+  type        = string
+  description = "Logical name of the lambda (e.g. address2uprn)"
+}
+
+variable "stage" {
+  description = "Deployment stage (e.g. dev, prod)"
+  type        = string
+}
+variable "ecr_repo_url" {
+  type        = string
+  description = "ECR repository URL (no tag, no digest)"
+}
+
+variable "image_digest" {
+  type        = string
+  description = "Image digest (sha256:...)"
+}
+
+variable "maximum_concurrency" {
+  type        = number
+  default     = null
+  description = "Maximum number of concurrent Lambda invocations from SQS (2-1000). null = no limit."
+}
+
+variable "batch_size" {
+  type    = number
+  default = 1
+}
+
+locals {
+  image_uri = "${var.ecr_repo_url}@${var.image_digest}"
+}
+
+output "resolved_image_uri" {
+  value = local.image_uri
+}
+
+
+variable "db_host" {
+  type = string
+}
+
+variable "db_name" {
+  type = string
+}
+
+variable "db_port" {
+  type = string
+}

infrastructure/terraform/shared/main.tf

@@ -664,12 +664,12 @@ module "hubspot_etl_s3_read_and_write" {
   source = "../modules/s3_iam_policy"
 
   policy_name        = "HubspotETLReadandWriteS3"
-  policy_description = "Allow ordnance Lambda to read and write from retrofit-data bucket"
+  policy_description = "Allow hubspot_etl_lambda Lambda to read and write from retrofit-data bucket"
   bucket_arns        = ["arn:aws:s3:::retrofit-data-${var.stage}"]
   actions            = ["s3:GetObject", "s3:ListBucket", "s3:PutObject"]
   resource_paths     = ["/*"]
 }
 
-output "ordnance_s3_read_and_write_arn" {
+output "hubspot_etl_s3_read_and_write_arn" {
   value = module.hubspot_etl_s3_read_and_write.policy_arn
 }

pyrightconfig.json

@@ -2,7 +2,7 @@
   "typeCheckingMode": "strict",
   "venvPath": "/Users/khalimconn-kowlessar/opt/anaconda3/envs/",
   "venv": "Fastapi-backend",
-"include": [
+  "include": [
     "."
   ]
 }

recommendations/Costs.py

@@ -209,7 +209,7 @@ class Costs:
                 # Try and get the region after converting the keys to lower
                 self.region = {
                     k.lower(): v for k, v in county_to_region_map.items()
-                }.get(self.property.property.epc_record.local_authority_label.lower(), None)
+                }.get(self.property.epc_record.local_authority_label.lower(), None)
 
             if self.region is None:
                 logger.warning("No region found for county %s, defaulting to South East England",

scripts/init_db.py

@@ -0,0 +1,5 @@
+from sqlmodel import SQLModel
+import backend.app.db.models.organisation  # noqa: F401
+from backend.app.db.connection import db_engine
+
+SQLModel.metadata.create_all(db_engine)

sfr/principal_pitch/2_export_data.py

@@ -26,15 +26,13 @@ from backend.app.db.functions.materials_functions import get_materials
 from collections import defaultdict
 from sqlalchemy import func
 
-# PORTFOLIO_ID = 206
-# SCENARIOS = [389]
-PORTFOLIO_ID = 633
-SCENARIOS = [1146]
+PORTFOLIO_ID = 639
+SCENARIOS = [1157]
 scenario_names = {
-    1146: "Most Economic",
+    1157: "EPC C - no EWI solid floor",
 }
 
-project_name = "WCHG EPC D rated properties"
+project_name = "Instagroup Sample"
 
 
 def get_data(portfolio_id, scenario_ids):

utils/s3.py

@@ -472,3 +472,14 @@ def list_xmls_in_s3_folder(bucket_name, folder_name):
             f"Failed to list XML files in folder {folder_name} in bucket {bucket_name}: {str(e)}"
         )
         return []
+
+
+def upload_file_to_s3(file_path: str, bucket_name: str, file_key: str) -> None:
+    try:
+        s3 = boto3.resource("s3")
+        bucket = s3.Bucket(bucket_name)
+        bucket.upload_file(file_path, file_key)
+        logger.info(f"Uploaded {file_path} to s3://{bucket_name}/{file_key}")
+    except Exception as e:
+        logger.error(f"Failed to upload {file_path} to S3: {e}")
+        raise