mirror of
https://github.com/Hestia-Homes/Model.git
synced 2026-06-30 13:10:47 +00:00
78 lines
2.6 KiB
Python
78 lines
2.6 KiB
Python
import boto3
|
|
import json
|
|
import math
|
|
import asyncio
|
|
import random
|
|
|
|
from datetime import datetime
|
|
|
|
from fastapi import APIRouter, Depends
|
|
from backend.app.dependencies import validate_token
|
|
from backend.app.plan.schemas import PlanTriggerRequest
|
|
from backend.app.config import get_settings
|
|
from sqlalchemy.orm import sessionmaker
|
|
from utils.logger import setup_logger
|
|
from backend.app.db.connection import db_engine
|
|
from backend.app.db.functions.recommendations_functions import create_scenario
|
|
import pandas as pd
|
|
from backend.app.whlg.schema import WHLGElligibilityRequest
|
|
|
|
from utils.s3 import read_csv_from_s3
|
|
from sqlalchemy.dialects.postgresql import insert
|
|
from backend.app.db.connection import get_db_session
|
|
from backend.app.db.models.whlg import Whlg
|
|
from backend.app.db.functions.whlg_functions import upsert_whlg_postcode
|
|
|
|
logger = setup_logger()
|
|
|
|
|
|
if get_settings().ENVIRONMENT == "local":
|
|
router = APIRouter(
|
|
prefix="/whlg",
|
|
tags=["whlg"],
|
|
)
|
|
|
|
else:
|
|
router = APIRouter(
|
|
prefix="/whlg",
|
|
tags=["whlg"],
|
|
dependencies=[Depends(validate_token)],
|
|
responses={404: {"description": "Not found"}}
|
|
)
|
|
|
|
@router.get("/")
|
|
async def whlg_entrypoint():
|
|
# body needs to include postcode, UPRN [task ID?]
|
|
#
|
|
# Refer to the plan trigger route for code
|
|
# 1) Create an event schema and store it in the schemas file
|
|
# 2) Build the tasks functions
|
|
# 3) Read in the funding csx. This can be found as such:
|
|
# whlg_eligible_postcodes = read_csv_from_s3(
|
|
# bucket_name=get_settings().DATA_BUCKET,
|
|
# filepath="funding/whlg eligible postcodes.csv",
|
|
# )
|
|
# whlg_eligible_postcodes = pd.DataFrame(whlg_eligible_postcodes)
|
|
# Check the postcode against this file
|
|
# We need to store this somewhere????!!!??!??!?!?!?!??!??!??!??!??!??!??!??!??!??! Create a new table!
|
|
# Update subtask to be complete
|
|
# Once this is complete, build the logs stuff, add the cloudwatch logs ID to the database
|
|
return {"hello": "from whlg"}
|
|
|
|
|
|
@router.post("/eligible")
|
|
async def eligiable(body: WHLGElligibilityRequest):
|
|
postcode = body.postcode or ""
|
|
postcode = postcode.lower().replace(" ", "")
|
|
|
|
whlg_eligible_postcodes = read_csv_from_s3(
|
|
bucket_name=get_settings().DATA_BUCKET,
|
|
filepath="funding/whlg eligible postcodes.csv",
|
|
)
|
|
whlg_eligible_postcodes = pd.DataFrame(whlg_eligible_postcodes)
|
|
whlg_eligible_postcodes['Postcode'] = whlg_eligible_postcodes['Postcode'].str.replace(' ', '', regex=False)
|
|
|
|
is_eligible = postcode in whlg_eligible_postcodes['Postcode'].values
|
|
return {"whlg_eligible": is_eligible}
|
|
|
|
|