added landlord description overrides

.github/workflows/_deploy_lambda.yml

@@ -92,6 +92,9 @@ on:
 
       TF_VAR_magicplan_api_key:
         required: false
+
+      TF_VAR_openai_api_key:
+        required: false
 jobs:
   deploy:
     runs-on: ubuntu-latest
@@ -163,6 +166,7 @@ jobs:
           TF_VAR_hubspot_api_key: ${{ secrets.TF_VAR_hubspot_api_key }}
           TF_VAR_magicplan_customer_id: ${{ secrets.TF_VAR_magicplan_customer_id }}
           TF_VAR_magicplan_api_key: ${{ secrets.TF_VAR_magicplan_api_key }}
+          TF_VAR_openai_api_key: ${{ secrets.TF_VAR_openai_api_key }}
         run: |
           ECR_REPO_URL_VAR=""
           if [[ -n "${{ inputs.ecr_repo }}" ]]; then
@@ -213,6 +217,7 @@ jobs:
           TF_VAR_hubspot_api_key: ${{ secrets.TF_VAR_hubspot_api_key }}
           TF_VAR_magicplan_customer_id: ${{ secrets.TF_VAR_magicplan_customer_id }}
           TF_VAR_magicplan_api_key: ${{ secrets.TF_VAR_magicplan_api_key }}
+          TF_VAR_openai_api_key: ${{ secrets.TF_VAR_openai_api_key }}
         run: |
           EXTRA_VARS=""
           if [[ -n "${{ inputs.ecr_repo }}" ]]; then

.github/workflows/deploy_terraform.yml

@@ -203,6 +203,47 @@ jobs:
       AWS_SECRET_ACCESS_KEY: ${{ secrets.DEV_AWS_SECRET_ACCESS_KEY }}
       AWS_REGION: ${{ secrets.DEV_AWS_REGION }}
 
+  # ============================================================
+  # Build Landlord Description Overrides image and Push
+  # ============================================================
+  landlordDescriptionOverrides_image:
+    needs: [determine_stage, shared_terraform]
+    uses: ./.github/workflows/_build_image.yml
+    with:
+      ecr_repo: landlord_description_overrides-${{ needs.determine_stage.outputs.stage }}
+      dockerfile_path: applications/landlord_description_overrides/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 Landlord Description Overrides Lambda
+  # ============================================================
+  landlordDescriptionOverrides_lambda:
+    needs: [landlordDescriptionOverrides_image, determine_stage]
+    uses: ./.github/workflows/_deploy_lambda.yml
+    with:
+      lambda_name: landlordDescriptionOverrides
+      lambda_path: deployment/terraform/lambda/landlordDescriptionOverrides
+      stage: ${{ needs.determine_stage.outputs.stage }}
+      ecr_repo: landlord_description_overrides-${{ needs.determine_stage.outputs.stage }}
+      image_digest: ${{ needs.landlordDescriptionOverrides_image.outputs.image_digest }}
+      terraform_apply: ${{ needs.determine_stage.outputs.terraform_apply }}
+    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 }}
+      TF_VAR_openai_api_key: ${{ secrets.DEV_OPENAI_API_KEY }}
+
   # ============================================================
   # Build Bulk Address2UPRN Combiner image and Push
   # ============================================================

.github/workflows/lambda_smoke_tests.yml

@@ -43,6 +43,16 @@ jobs:
       build_context: .
       service_name: postcode-splitter-ddd
 
+  # ============================================================
+  # Landlord Description Overrides
+  # ============================================================
+  landlord_description_overrides_smoke_test:
+    uses: ./.github/workflows/_smoke_test_lambda.yml
+    with:
+      dockerfile_path: applications/landlord_description_overrides/Dockerfile
+      build_context: .
+      service_name: landlord-description-overrides
+
   # ============================================================
   # Bulk Address2UPRN Combiner
   # ============================================================

applications/landlord_description_overrides/Dockerfile

@@ -15,7 +15,7 @@ ENV POSTGRES_DATABASE=${DEV_DB_NAME}
 
 WORKDIR /var/task
 
-COPY applications/postcode_splitter/requirements.txt .
+COPY applications/landlord_description_overrides/requirements.txt .
 RUN pip install --no-cache-dir -r requirements.txt
 
 # Copy the layered source the handler imports from. The new splitter pulls

applications/landlord_description_overrides/handler.py

@@ -1,14 +1,12 @@
 import logging
 import os
 from typing import Any
-from uuid import UUID
 
 import boto3
 
 from applications.landlord_description_overrides.landlord_description_overrides_trigger_body import (
     LandlordDescriptionOverridesTriggerBody,
 )
-from domain.addresses.unstandardised_address import AddressList
 from domain.landlord_description_overrides.built_form_type import BuiltFormType
 from domain.landlord_description_overrides.property_type import PropertyType
 from domain.landlord_description_overrides.roof_type import RoofType
@@ -33,36 +31,90 @@ from infrastructure.postgres.landlord_wall_type_override_postgres_repository imp
     LandlordWallTypeOverridePostgresRepository,
 )
 from infrastructure.s3.csv_s3_client import CsvS3Client
+from infrastructure.s3.s3_uri import parse_s3_uri
 from orchestration.classifiable_column import ClassifiableColumn
 from orchestration.landlord_description_overrides_orchestrator import (
     LandlordDescriptionOverridesOrchestrator,
 )
+from orchestration.task_orchestrator import TaskOrchestrator
 from repositories.unstandardised_address.unstandardised_address_list_csv_s3_repository import (
     UnstandardisedAddressListCsvS3Repository,
 )
+from utilities.aws_lambda.subtask_handler import subtask_handler
 
 logger = logging.getLogger(__name__)
 
 
+def _build_columns(
+    column_mapping: dict[str, str], chat_gpt: ChatGPT, session: Any
+) -> list[ClassifiableColumn[Any]]:
+    """One ClassifiableColumn per mapped category.
+
+    ``column_mapping`` is ``{category -> source CSV header}``. One header may
+    feed several categories -- e.g. ``"Property Type"`` -> property_type and
+    built_form_type -- which falls out naturally because each is a separate
+    entry. Unknown categories are skipped.
+    """
+    factories = {
+        "property_type": lambda src: ClassifiableColumn(
+            name="property_type",
+            source_column=src,
+            classifier=ChatGptColumnClassifier(
+                chat_gpt, PropertyType, PropertyType.UNKNOWN
+            ),
+            repo=LandlordPropertyTypeOverridePostgresRepository(session),
+        ),
+        "built_form_type": lambda src: ClassifiableColumn(
+            name="built_form_type",
+            source_column=src,
+            classifier=ChatGptColumnClassifier(
+                chat_gpt, BuiltFormType, BuiltFormType.UNKNOWN
+            ),
+            repo=LandlordBuiltFormTypeOverridePostgresRepository(session),
+        ),
+        "wall_type": lambda src: ClassifiableColumn(
+            name="wall_type",
+            source_column=src,
+            classifier=ChatGptColumnClassifier(
+                chat_gpt,
+                WallType,
+                WallType.UNKNOWN,
+                extra_instructions=wall_type_construction_date_prompt_hint(),
+            ),
+            repo=LandlordWallTypeOverridePostgresRepository(session),
+        ),
+        "roof_type": lambda src: ClassifiableColumn(
+            name="roof_type",
+            source_column=src,
+            classifier=ChatGptColumnClassifier(
+                chat_gpt, RoofType, RoofType.UNKNOWN
+            ),
+            repo=LandlordRoofTypeOverridePostgresRepository(session),
+        ),
+    }
+
+    columns: list[ClassifiableColumn[Any]] = []
+    for category, source_column in column_mapping.items():
+        factory = factories.get(category)
+        if factory is None:
+            logger.warning("Unknown classifier category %r; skipping.", category)
+            continue
+        columns.append(factory(source_column))
+    return columns
+
+
+@subtask_handler()
 def handler(
-    body: dict[str, Any],
-    context: Any,
-) -> dict[str, list[str]]:
-    # TODO: replace with ``LandlordDescriptionOverridesTriggerBody.model_validate(body)``
-    # once this lambda is wired into the parent task pipeline via the SQS
-    # subtask envelope. Until then the trigger fields are hard-coded so the
-    # local invoker can exercise the full path. See ADR-0003 §Out of scope.
-    trigger = LandlordDescriptionOverridesTriggerBody(
-        task_id=UUID("00000000-0000-0000-0000-000000000001"),
-        sub_task_id=UUID("00000000-0000-0000-0000-000000000002"),
-        s3_uri="s3://retrofit-data-dev/bulk_onboarding_inputs/hyde2 (1).csv",
-        portfolio_id=730,
-    )
+    body: dict[str, Any], context: Any, task_orchestrator: TaskOrchestrator
+) -> dict[str, int]:
+    trigger = LandlordDescriptionOverridesTriggerBody.model_validate(body)
 
-    bucket = "retrofit-data-dev"
+    # The classifier reads the ORIGINAL upload (raw landlord headers), so the S3
+    # bucket comes from the trigger URI rather than a fixed env var.
+    bucket, _key = parse_s3_uri(trigger.s3_uri)
 
-    # boto3.client is overloaded per-service in the installed stubs; cast
-    # to Any so the strict-mode checker treats it as opaque.
+    # boto3.client is overloaded per-service in the installed stubs; cast to Any
+    # so the strict-mode checker treats it as opaque.
     boto3_client: Any = (
         boto3.client
     )  # pyright: ignore[reportUnknownMemberType, reportUnknownVariableType]
@@ -73,74 +125,23 @@ def handler(
         csv_client, bucket
     )
 
-    # One transactional session per handler invocation: the context manager
-    # commits on clean exit and rolls back on exception, so the handler never
-    # invokes ``.commit()`` itself -- transaction semantics live in the
-    # infrastructure layer.
+    # Raw rows, not load_batch: the original upload carries the description
+    # columns but not the canonical address/postcode columns load_batch requires.
+    rows = csv_client.read_rows(trigger.s3_uri)
+
     engine = make_engine(PostgresConfig.from_env(os.environ))
     with transactional_session(engine) as session:
         chat_gpt = ChatGPT()
-        # The "Property Type" CSV column is read by two classifiers: the
-        # landlord's free-text (e.g. "semi-detached house") encodes both the
-        # dwelling kind (PropertyType) and how it joins to neighbours
-        # (BuiltFormType). Each classification lands in its own table.
-        columns: list[ClassifiableColumn[Any]] = [
-            ClassifiableColumn(
-                name="property_type",
-                source_column="Property Type",
-                classifier=ChatGptColumnClassifier(
-                    chat_gpt, PropertyType, PropertyType.UNKNOWN
-                ),
-                repo=LandlordPropertyTypeOverridePostgresRepository(session),
-            ),
-            ClassifiableColumn(
-                name="built_form_type",
-                source_column="Property Type",
-                classifier=ChatGptColumnClassifier(
-                    chat_gpt, BuiltFormType, BuiltFormType.UNKNOWN
-                ),
-                repo=LandlordBuiltFormTypeOverridePostgresRepository(session),
-            ),
-            ClassifiableColumn(
-                name="wall_type",
-                source_column="Walls",
-                classifier=ChatGptColumnClassifier(
-                    chat_gpt,
-                    WallType,
-                    WallType.UNKNOWN,
-                    extra_instructions=wall_type_construction_date_prompt_hint(),
-                ),
-                repo=LandlordWallTypeOverridePostgresRepository(session),
-            ),
-            ClassifiableColumn(
-                name="roof_type",
-                source_column="Roofs",
-                classifier=ChatGptColumnClassifier(
-                    chat_gpt, RoofType, RoofType.UNKNOWN
-                ),
-                repo=LandlordRoofTypeOverridePostgresRepository(session),
-            ),
-        ]
-
+        columns = _build_columns(trigger.column_mapping, chat_gpt, session)
         orchestrator = LandlordDescriptionOverridesOrchestrator(
             unstandardised_address_repo=unstandardised_address_repo,
             columns=columns,
         )
-
-        addressList: AddressList = orchestrator.get_unstandardised_addresses(
-            input_s3_uri=trigger.s3_uri
+        classified = orchestrator.classify_and_persist_from_rows(
+            rows, portfolio_id=trigger.portfolio_id
         )
 
-        # Cap the batch to the first 20 while the ChatGPT path is under test.
-        # Remove before wiring into the production subtask pipeline.
-        addressList = AddressList(addressList[:20])
-
-        classified = orchestrator.classify_and_persist(
-            addressList, portfolio_id=trigger.portfolio_id
-        )
-        for column, mapping in classified.items():
-            logger.info(
-                "Classified %d descriptions for column %r.", len(mapping), column
-            )
-
-    return {"hello": ["200"]}
+    counts = {name: len(mapping) for name, mapping in classified.items()}
+    for name, n in counts.items():
+        logger.info("Classified %d descriptions for column %r.", n, name)
+    return counts

applications/landlord_description_overrides/landlord_description_overrides_trigger_body.py

@@ -13,3 +13,7 @@ class LandlordDescriptionOverridesTriggerBody(BaseModel):
     # Python ``int`` is unbounded so the Pydantic side stays simple; the
     # SQLModel row class pins the storage to ``BigInteger``.
     portfolio_id: int
+    # category -> source CSV header (the classifier subset of the upload
+    # mapping). Defaulted so a malformed/empty message classifies nothing
+    # rather than failing validation.
+    column_mapping: dict[str, str] = {}

applications/landlord_description_overrides/requirements.txt

@@ -2,3 +2,4 @@ boto3
 pydantic
 sqlmodel
 psycopg2-binary
+openai

backend/app/bulk_uploads/router.py

@@ -13,6 +13,7 @@ from backend.app.bulk_uploads.schema import (
     CombinedResultsResponse,
     CombinerTriggerRequest,
     FlagsSummary,
+    LandlordOverridesTriggerRequest,
     PostcodeSplitterTriggerRequest,
 )
 from backend.app.bulk_uploads.scoring import score_bucket
@@ -92,6 +93,26 @@ async def trigger_combiner(req: CombinerTriggerRequest):
     }
 
 
+@router.post("/trigger-landlord-overrides", status_code=202)
+async def trigger_landlord_overrides(req: LandlordOverridesTriggerRequest):
+    settings = get_settings()
+
+    try:
+        sqs = boto3.client("sqs", settings.AWS_DEFAULT_REGION)
+        response = sqs.send_message(
+            QueueUrl=settings.LANDLORD_OVERRIDES_SQS_URL,
+            MessageBody=req.model_dump_json(),
+        )
+    except Exception as e:
+        raise HTTPException(status_code=500, detail=f"SQS error: {e}")
+
+    return {
+        "task_id": req.task_id,
+        "sub_task_id": req.sub_task_id,
+        "sqs_message_id": response.get("MessageId"),
+    }
+
+
 @router.get("/{task_id}/combined-results", response_model=CombinedResultsResponse)
 async def get_combined_results(
     task_id: UUID,

backend/app/bulk_uploads/schema.py

@@ -14,6 +14,15 @@ class CombinerTriggerRequest(BaseModel):
     sub_task_id: str
 
 
+class LandlordOverridesTriggerRequest(BaseModel):
+    task_id: str
+    sub_task_id: str
+    s3_uri: str
+    portfolio_id: int
+    # category -> source CSV header (the classifier subset of the upload mapping)
+    column_mapping: dict[str, str]
+
+
 class FlagsSummary(BaseModel):
     duplicates: int
     missing: int

backend/app/config.py

@@ -42,6 +42,7 @@ class Settings(BaseSettings):
     MAGICPLAN_SQS_URL: str = "changeme"
     POSTCODE_SPLITTER_SQS_URL: str = "changeme"
     COMBINER_SQS_URL: str = "changeme"
+    LANDLORD_OVERRIDES_SQS_URL: str = "changeme"
 
     # Third parties
     EPC_AUTH_TOKEN: str = "changeme"

deployment/terraform/lambda/landlordDescriptionOverrides/main.tf

@@ -0,0 +1,50 @@
+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 "lambda" {
+  source = "../../modules/lambda_with_sqs"
+
+  name  = "landlord-description-overrides"
+  stage = var.stage
+
+  image_uri = local.image_uri
+
+  # The classifier calls OpenAI once per distinct description per column, so it
+  # is latency-bound. 300s leaves headroom under the queue's 1000s visibility
+  # timeout. batch_size = 1 keeps one upload per invocation, so a single bad
+  # record cannot redrive its siblings. maximum_concurrency caps fan-out to
+  # respect OpenAI rate limits.
+  timeout             = 300
+  batch_size          = 1
+  maximum_concurrency = 5
+
+  environment = merge(
+    {
+      STAGE             = var.stage
+      LOG_LEVEL         = "info"
+      POSTGRES_USERNAME = local.db_credentials.db_assessment_model_username
+      POSTGRES_PASSWORD = local.db_credentials.db_assessment_model_password
+      OPENAI_API_KEY    = var.openai_api_key
+    },
+  )
+}
+
+# Attach S3 read policy so the handler can read the original upload CSV.
+resource "aws_iam_role_policy_attachment" "landlord_overrides_s3_read" {
+  role       = module.lambda.role_name
+  policy_arn = data.terraform_remote_state.shared.outputs.landlord_overrides_s3_read_arn
+}

deployment/terraform/lambda/landlordDescriptionOverrides/outputs.tf

@@ -0,0 +1,9 @@
+output "landlord_description_overrides_queue_url" {
+  value       = module.lambda.queue_url
+  description = "URL of the Landlord Description Overrides SQS queue (wire into the FastAPI LANDLORD_OVERRIDES_SQS_URL)"
+}
+
+output "landlord_description_overrides_queue_arn" {
+  value       = module.lambda.queue_arn
+  description = "ARN of the Landlord Description Overrides SQS queue"
+}

deployment/terraform/lambda/landlordDescriptionOverrides/provider.tf

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

deployment/terraform/lambda/landlordDescriptionOverrides/variables.tf

@@ -0,0 +1,33 @@
+variable "lambda_name" {
+  type        = string
+  description = "Logical name of the lambda (e.g. landlordDescriptionOverrides)"
+}
+
+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 "openai_api_key" {
+  type        = string
+  description = "OpenAI API key used by the ChatGPT column classifier"
+  sensitive   = true
+}
+
+locals {
+  image_uri = "${var.ecr_repo_url}@${var.image_digest}"
+}
+
+output "resolved_image_uri" {
+  value = local.image_uri
+}

deployment/terraform/shared/main.tf

@@ -268,11 +268,11 @@ output "retrofit_heat_baseline_predictions_bucket_name" {
 
 // We make this bucket presignable, because we want to generate download links for the frontend
 module "retrofit_energy_assessments" {
-  source             = "../modules/s3_presignable_bucket"
-  bucketname         = "retrofit-energy-assessments-${var.stage}"
-  allowed_origins    = var.allowed_origins
-  environment        = var.stage
-  enable_versioning  = true
+  source            = "../modules/s3_presignable_bucket"
+  bucketname        = "retrofit-energy-assessments-${var.stage}"
+  allowed_origins   = var.allowed_origins
+  environment       = var.stage
+  enable_versioning = true
 }
 
 output "retrofit_energy_assessments_bucket_name" {
@@ -494,6 +494,35 @@ output "postcode_splitter_s3_read_arn" {
   value = module.postcode_splitter_s3_read.policy_arn
 }
 
+################################################
+# Landlord Description Overrides – Lambda
+################################################
+module "landlord_description_overrides_state_bucket" {
+  source      = "../modules/tf_state_bucket"
+  bucket_name = "landlord-description-overrides-terraform-state"
+}
+
+module "landlord_description_overrides_registry" {
+  source = "../modules/container_registry"
+  name   = "landlord_description_overrides"
+  stage  = var.stage
+}
+
+# S3 policy for the landlord classifier to read the original upload CSV.
+module "landlord_overrides_s3_read" {
+  source = "../modules/s3_iam_policy"
+
+  policy_name        = "LandlordOverridesReadS3"
+  policy_description = "Allow landlord description overrides Lambda to read from retrofit-data bucket"
+  bucket_arns        = ["arn:aws:s3:::retrofit-data-${var.stage}"]
+  actions            = ["s3:GetObject", "s3:ListBucket"]
+  resource_paths     = ["/*"]
+}
+
+output "landlord_overrides_s3_read_arn" {
+  value = module.landlord_overrides_s3_read.policy_arn
+}
+
 ################################################
 # Bulk Address2UPRN Combiner – Lambda ECR
 ################################################
@@ -729,7 +758,7 @@ module "hubspot_etl_bucket" {
 module "hubspot_etl_registry" {
   source = "../modules/container_registry"
   name   = "hubspot-etl"
-  stage = var.stage
+  stage  = var.stage
 
 }
 

orchestration/landlord_description_overrides_orchestrator.py

@@ -81,3 +81,39 @@ class LandlordDescriptionOverridesOrchestrator:
                 continue
             column.repo.upsert_all(portfolio_id, mapping)
         return classified
+
+    def classify_and_persist_from_rows(
+        self, rows: list[dict[str, str]], portfolio_id: int
+    ) -> dict[str, dict[str, Enum]]:
+        """Classify + persist straight from raw CSV rows.
+
+        Unlike ``classify_and_persist``, this does not build an ``AddressList``,
+        so it has no canonical address/postcode requirement -- the classifier
+        only needs the raw description cells. Used when reading the original
+        landlord upload (raw headers) rather than the address-matching CSV.
+        """
+        col_to_desc = self._descriptions_from_rows(rows)
+        classified = {
+            column.name: column.classifier.classify(
+                col_to_desc.get(column.source_column, set())
+            )
+            for column in self._columns
+        }
+        for column in self._columns:
+            mapping = classified[column.name]
+            if not mapping:
+                continue
+            column.repo.upsert_all(portfolio_id, mapping)
+        return classified
+
+    @staticmethod
+    def _descriptions_from_rows(rows: list[dict[str, str]]) -> dict[str, set[str]]:
+        mappings: dict[str, set[str]] = {}
+        for row in rows:
+            for key, value in row.items():
+                bucket = mappings.setdefault(key, set())
+                for variant in (value or "").split(","):
+                    variant = variant.strip().lower()
+                    if variant:
+                        bucket.add(variant)
+        return mappings