mirror of
https://github.com/Hestia-Homes/Model.git
synced 2026-07-12 13:29:04 +00:00
50 lines
1.6 KiB
Python
50 lines
1.6 KiB
Python
import os
|
|
import boto3
|
|
from typing import Any, Optional
|
|
|
|
from applications.magic_plan.magic_plan_trigger_request import MagicPlanTriggerRequest
|
|
from infrastructure.magic_plan.config import MagicPlanConfig
|
|
from infrastructure.magic_plan.magic_plan_client import MagicPlanClient
|
|
from infrastructure.s3.s3_client import S3Client
|
|
from orchestration.magic_plan_orchestrator import MagicPlanOrchestrator
|
|
from domain.magicplan.models import Plan
|
|
from domain.tasks.tasks import Source
|
|
from utilities.aws_lambda.task_handler import task_handler
|
|
from utilities.logger import setup_logger
|
|
|
|
logger = setup_logger()
|
|
|
|
|
|
@task_handler(task_source="magic_plan", source=Source.HUBSPOT_DEAL)
|
|
def handler(body: dict[str, Any], context: Any) -> Optional[str]:
|
|
config = MagicPlanConfig.from_env(os.environ)
|
|
payload = MagicPlanTriggerRequest.model_validate(body)
|
|
client = MagicPlanClient(
|
|
customer_id=config.customer_id,
|
|
api_key=config.api_key,
|
|
)
|
|
|
|
boto3_client: Any = boto3.client # type: ignore
|
|
boto_s3: Any = boto3_client("s3")
|
|
s3_client = S3Client(
|
|
boto_s3_client=boto_s3, bucket="retrofit-energy-assessments-dev"
|
|
)
|
|
# TODO: read s3_bucket from env var so staging/prod use the correct bucket
|
|
|
|
plan: Optional[Plan] = MagicPlanOrchestrator(client, s3_client).run(payload)
|
|
if plan:
|
|
logger.info("Saved MagicPlan plan uid=%s", plan.uid)
|
|
return plan.uid
|
|
|
|
return None
|
|
|
|
|
|
if __name__ == "__main__":
|
|
event = {
|
|
"Records": [
|
|
{
|
|
"body": '{"address": "2 Laburnum Way Bromley BR2 8BZ", "hubspot_deal_id": "local-test-deal"}',
|
|
}
|
|
]
|
|
}
|
|
handler(event, None)
|