mirror of
https://github.com/Hestia-Homes/Model.git
synced 2026-06-08 11:17:27 +00:00
44 lines
1.5 KiB
Python
44 lines
1.5 KiB
Python
from typing import Any
|
|
import boto3
|
|
from orchestration.sal_orchestrator import (
|
|
SALOrchestrator,
|
|
)
|
|
from infrastructure.csv_s3_client import CsvS3Client
|
|
from repositories.unsanitised_address.unsanitised_address_list_csv_s3_repository import (
|
|
UnsanitisedAddressListCsvS3Repository,
|
|
)
|
|
from domain.addresses.unsanitised_address import AddressList
|
|
|
|
|
|
def handler(
|
|
body: dict[str, Any],
|
|
context: Any,
|
|
) -> dict[str, list[str]]:
|
|
|
|
s3_uri = "s3://retrofit-data-dev/bulk_onboarding_inputs/hyde2 (1).csv"
|
|
bucket = "retrofit-data-dev"
|
|
|
|
# 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 # noqa
|
|
boto_s3: Any = boto3_client("s3")
|
|
|
|
csv_client = CsvS3Client(boto_s3, bucket)
|
|
unsanitised_address_repo = UnsanitisedAddressListCsvS3Repository(csv_client, bucket)
|
|
|
|
sal = SALOrchestrator(
|
|
unsanitised_address_repo=unsanitised_address_repo,
|
|
)
|
|
|
|
addressList: AddressList = sal.get_unsanitised_addresses(input_s3_uri=s3_uri)
|
|
|
|
col_to_desc_map = sal.get_col_to_description_mappings(
|
|
list_of_unsanitised_address=addressList
|
|
)
|
|
|
|
# Read csv of user input
|
|
# get the column and unique variations of each description
|
|
# { walls: "wall variation 1", "wall varition 2"}
|
|
# Call chatgpt(input from landlord, our way of understanding the mapping) Retrun -> lanlordMapped
|
|
|
|
return {"hello world": ["hello world"]}
|