mirror of
https://github.com/Hestia-Homes/Model.git
synced 2026-06-08 11:17:27 +00:00
46 lines
1.4 KiB
Python
46 lines
1.4 KiB
Python
from typing import Any
|
|
import boto3
|
|
from orchestration.landlord_description_overrides_orchestrator import (
|
|
SALOrchestrator,
|
|
)
|
|
from infrastructure.csv_s3_client import CsvS3Client
|
|
from repositories.raw_address.raw_address_csv_s3_repository import (
|
|
RawAddressCsvS3Repository,
|
|
)
|
|
from domain.addresses.raw_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)
|
|
raw_address_repo = RawAddressCsvS3Repository(csv_client, bucket)
|
|
|
|
orchestrator = SALOrchestrator(
|
|
raw_address_repo=raw_address_repo,
|
|
)
|
|
|
|
list_of_raw_address: AddressList = orchestrator.get_raw_addresses(
|
|
input_s3_uri=s3_uri
|
|
)
|
|
|
|
col_to_desc_map = orchestrator.get_col_to_description_mappings(
|
|
list_of_raw_address=list_of_raw_address
|
|
)
|
|
|
|
# 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"]}
|