Merge pull request #81 from Hestia-Homes/feautre/adhoc_script_for_nick

Feautre/adhoc script for nick
This commit is contained in:
Jun-te Kim 2025-09-05 10:44:20 +01:00 committed by GitHub
commit 1eabc7214c
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
15 changed files with 737 additions and 621 deletions

View file

@ -1,17 +1,20 @@
name: Months End
on:
schedule:
- cron: '0 7 29 * *' # Runs at 7:00 AM UTC on the 29th of every month
- cron: '0 7 29 * *' # Runs at 07:00 UTC on the 29th of every month
workflow_dispatch:
push:
branches: [main, feature/month_end_automation_of_all]
jobs:
surveyed-needs-sign-off:
runs-on: ubuntu-22.04
steps:
- uses: actions/checkout@v3
- uses: actions/checkout@v4
- name: Set up Python
uses: actions/setup-python@v4
uses: actions/setup-python@v5
with:
python-version: '3.12'
@ -20,26 +23,26 @@ jobs:
pip install poetry
poetry install --no-root
- name: run script
- name: Run scripts
env:
PYTHONPATH: ${{ github.workspace }}
run: |
pwd
ls -la
poetry run python etl/month_end_automation_wave_2_layout.py
poetry run python etl/month_end_automation_wave_2_no_3.py
poetry run python etl/month_end_automation_wave_2_no_4.py
poetry run python etl/month_end_automation_wave_2_no_5.py
# poetry run python etl/month_end_automation_wave_2_no_5.py
poetry run python etl/month_end_automation_wave_2_no_6.py
poetry run python etl/month_end_automation_wave_2_no_7.py
poetry run python etl/month_end_automation_wave_2_no_8.py
poetry run python etl/month_end_automation_wave_2_no_9.py
poetry run python etl/month_end_automation_wave_2_no_10.py
# poetry run python etl/month_end_automation_wave_2_no_10.py
poetry run python etl/month_end_automation_wave_2_no_11.py
poetry run python etl/month_end_automation_wave_2_no_12.py
poetry run python etl/month_end_automation_wave_2_no_13.py
# poetry run python etl/month_end_automation_wave_2_no_13.py
poetry run python etl/month_end_automation_wave_2_no_14.py
poetry run python etl/month_end_automation_wave_2_no_15.py
poetry run python etl/month_end_automation_wave_2_no_16.py
# poetry run python etl/month_end_automation_wave_2_no_16.py
poetry run python etl/month_end_automation_wave_accent_housing.py
poetry run python etl/month_end_automation_wave_3_layout.py
env:
PYTHONPATH: ${{ github.workspace }}

4
.gitignore vendored
View file

@ -4,4 +4,6 @@ __pycache__/
data/
*ipynb
etl/survery_data.csv
foo.env.py
foo.env.py
*.xlsx
*.csv

View file

@ -13,12 +13,37 @@ def upload_to_month_end_folder(file_name_on_sp, local_file_path, add_to_path):
today = datetime.today()
# Format as "Month YYYY"
formatted_date = today.strftime("%B %Y")
sharepoint.create_dir(formatted_date, parent_folder)
sharepoint_path = parent_folder + "/" + formatted_date
# Make day month year folder
formatted_date = today.strftime("%d-%m-%y")
sharepoint.create_dir(formatted_date, sharepoint_path)
sharepoint_path += "/" + formatted_date
# Make company folder
sharepoint.create_dir(add_to_path, sharepoint_path)
sharepoint_path += "/" + add_to_path
print("Uploading to sharepoint...")
sharepoint.upload_file(local_file_path, sharepoint_path, file_name_on_sp)
print(f"Finished upload of {local_file_path} to sharepoint. It's found under {sharepoint_path}/{file_name_on_sp}")
def upload_to_nick_folder(file_name_on_sp, local_file_path, add_to_path=None):
sharepoint = SharePointScraper(SharePointInstaller.OSMOSIS_WAVE_2)
parent_folder = "General/Junte Kim/For Nick"
today = datetime.today()
formatted_date = today.strftime("%Y-%m-%d%H-%M-S")
sharepoint.create_dir(formatted_date, parent_folder)
sharepoint_path = parent_folder + "/" + formatted_date
if add_to_path:
sharepoint.create_dir(add_to_path, sharepoint_path)
sharepoint_path += "/" + add_to_path
print("Uploading to sharepoint...")
sharepoint.upload_file(local_file_path, sharepoint_path, file_name_on_sp)
print(f"Finished upload of {local_file_path} to sharepoint. It's found under {sharepoint_path}/{file_name_on_sp}")

View file

@ -1,386 +1,386 @@
# # SQLModel mapping for ConditionReportModel using BaseModel
# from sqlmodel import SQLModel, Field, Relationship, Column, JSON
# from typing import Optional, List
# import uuid
# from datetime import datetime
# from etl.models.topLevel import BaseModel, Documents
# SQLModel mapping for ConditionReportModel using BaseModel
from sqlmodel import SQLModel, Field, Relationship, Column, JSON
from typing import Optional, List
import uuid
from datetime import datetime
from etl.models.topLevel import BaseModel, Documents
# class AssessorDetails(BaseModel, table=True):
# assessor_name_and_id: str
# elmhurst_id: str
class AssessorDetails(BaseModel, table=True):
assessor_name_and_id: str
elmhurst_id: str
# class InspectionAndProject(BaseModel, table=True):
# inspection_date: str
class InspectionAndProject(BaseModel, table=True):
inspection_date: str
# class TheProperty(BaseModel, table=True):
# house_type: str
# on_which_floor_is_the_flat_located: str
# is_there_a_corridor: bool
# is_it_heated: bool
# it_there_a_balcony: bool
# classification_type: str
# orientation_front_elevation: str
# orientation_in_degrees_front_elevation: str
# exposure_zone: str
# main_wall_construction: str
class TheProperty(BaseModel, table=True):
house_type: str
on_which_floor_is_the_flat_located: str
is_there_a_corridor: bool
is_it_heated: bool
it_there_a_balcony: bool
classification_type: str
orientation_front_elevation: str
orientation_in_degrees_front_elevation: str
exposure_zone: str
main_wall_construction: str
# class ElevationInfo(BaseModel, table=True):
# elevation_type: str
# cavity_wall_depth: str
# is_insulation_present: bool
# insulation_type: str
class ElevationInfo(BaseModel, table=True):
elevation_type: str
cavity_wall_depth: str
is_insulation_present: bool
insulation_type: str
# main_elevation: Optional["MainElevation"] = Relationship(back_populates="elevation_info")
main_elevation: Optional["MainElevation"] = Relationship(back_populates="elevation_info")
# elevation_id: Optional[uuid.UUID] = Field(foreign_key="elevation.id")
# elevation_table: Optional["Elevation"] = Relationship(back_populates="info")
elevation_id: Optional[uuid.UUID] = Field(foreign_key="elevation.id")
elevation_table: Optional["Elevation"] = Relationship(back_populates="info")
# class MainElevation(BaseModel, table=True):
# elevation_info_id: uuid.UUID = Field(foreign_key="elevationinfo.id")
class MainElevation(BaseModel, table=True):
elevation_info_id: uuid.UUID = Field(foreign_key="elevationinfo.id")
# #SQLAlcemy things
# elevation_info: ElevationInfo = Relationship(back_populates="main_elevation")
#SQLAlcemy things
elevation_info: ElevationInfo = Relationship(back_populates="main_elevation")
# class Elevation(BaseModel, table=True):
# protected_conservatory_or_aonb: bool
# material_type: str
# visible_signs_of_existing_wall_insulation: str
# ground_level_bridge_the_dpc: bool
class Elevation(BaseModel, table=True):
protected_conservatory_or_aonb: bool
material_type: str
visible_signs_of_existing_wall_insulation: str
ground_level_bridge_the_dpc: bool
# info: List["ElevationInfo"] = Relationship(back_populates="elevation_table")
info: List["ElevationInfo"] = Relationship(back_populates="elevation_table")
# class GeneralInformation(BaseModel, table=True):
# assessor_detail_id: uuid.UUID = Field(foreign_key="assessordetails.id")
# inspection_and_project_id: uuid.UUID = Field(foreign_key="inspectionandproject.id")
# the_property_id: uuid.UUID = Field(foreign_key="theproperty.id")
# main_elevation_id: uuid.UUID = Field(foreign_key="mainelevation.id")
# elevations_id: uuid.UUID = Field(foreign_key="elevation.id")
class GeneralInformation(BaseModel, table=True):
assessor_detail_id: uuid.UUID = Field(foreign_key="assessordetails.id")
inspection_and_project_id: uuid.UUID = Field(foreign_key="inspectionandproject.id")
the_property_id: uuid.UUID = Field(foreign_key="theproperty.id")
main_elevation_id: uuid.UUID = Field(foreign_key="mainelevation.id")
elevations_id: uuid.UUID = Field(foreign_key="elevation.id")
# assessor_details: AssessorDetails = Relationship()
# inspection_and_project: InspectionAndProject = Relationship()
# the_property: TheProperty = Relationship()
# main_elevation: MainElevation = Relationship()
# elevations: Elevation = Relationship()
assessor_details: AssessorDetails = Relationship()
inspection_and_project: InspectionAndProject = Relationship()
the_property: TheProperty = Relationship()
main_elevation: MainElevation = Relationship()
elevations: Elevation = Relationship()
# class PropertyAccess(BaseModel, table=True):
# are_there_any_road_restriction_in_the_locality: bool
# is_on_street_parking_available: bool
# are_there_any_overhead_wires_or_cables: bool
# is_the_access_gated: bool
# is_there_restricted_space_for_contractors_to_access_the_wall_area: bool
# is_there_restricted_space_for_contractors_to_access_the_roof_area: bool
# more_than_1_5_meters_in_width_to_fence_or__along_the_full_gable_elevation: bool
# is_access_to_the_rear_provided_by_use_of_a_ginnel: bool
# is_access_to_the_rear_provided_by_use_of_a_secured_alleyway: bool
class PropertyAccess(BaseModel, table=True):
are_there_any_road_restriction_in_the_locality: bool
is_on_street_parking_available: bool
are_there_any_overhead_wires_or_cables: bool
is_the_access_gated: bool
is_there_restricted_space_for_contractors_to_access_the_wall_area: bool
is_there_restricted_space_for_contractors_to_access_the_roof_area: bool
more_than_1_5_meters_in_width_to_fence_or__along_the_full_gable_elevation: bool
is_access_to_the_rear_provided_by_use_of_a_ginnel: bool
is_access_to_the_rear_provided_by_use_of_a_secured_alleyway: bool
# class ExternalElevation(BaseModel, table=True):
# structural_defects_of_elevation: str
# does_any_structural_defect_need_resolving_before_retrofit: bool
# any_signs_of_water_penetration_caused_by_failed_rainwater_goods_or_pipework: bool
# are_there_any_visible_signs_of_movement: bool
# are_there_any_visible_signs_of_cracking_to_the_existing_external_finish: bool
class ExternalElevation(BaseModel, table=True):
structural_defects_of_elevation: str
does_any_structural_defect_need_resolving_before_retrofit: bool
any_signs_of_water_penetration_caused_by_failed_rainwater_goods_or_pipework: bool
are_there_any_visible_signs_of_movement: bool
are_there_any_visible_signs_of_cracking_to_the_existing_external_finish: bool
# class ExternalElevationFront(BaseModel, table=True):
# external_elevation_id: uuid.UUID = Field(foreign_key="externalelevation.id")
# external_elevation: ExternalElevation = Relationship()
class ExternalElevationFront(BaseModel, table=True):
external_elevation_id: uuid.UUID = Field(foreign_key="externalelevation.id")
external_elevation: ExternalElevation = Relationship()
# class ExternalElevationRear(BaseModel, table=True):
# do_all_answers_for_the_front_elevation_apply_to_this_wall: bool
# external_elevation_id: Optional[uuid.UUID] = Field(foreign_key="externalelevation.id")
# external_elevation: Optional[ExternalElevation] = Relationship()
class ExternalElevationRear(BaseModel, table=True):
do_all_answers_for_the_front_elevation_apply_to_this_wall: bool
external_elevation_id: Optional[uuid.UUID] = Field(foreign_key="externalelevation.id")
external_elevation: Optional[ExternalElevation] = Relationship()
# class ExternalElevationGableOne(BaseModel, table=True):
# do_all_answers_for_the_front_elevation_apply_to_this_wall: bool
# external_elevation_id: Optional[uuid.UUID] = Field(foreign_key="externalelevation.id")
# external_elevation: Optional[ExternalElevation] = Relationship()
class ExternalElevationGableOne(BaseModel, table=True):
do_all_answers_for_the_front_elevation_apply_to_this_wall: bool
external_elevation_id: Optional[uuid.UUID] = Field(foreign_key="externalelevation.id")
external_elevation: Optional[ExternalElevation] = Relationship()
# class ExternalElevationGableTwo(BaseModel, table=True):
# is_there_a_fourth_external_elevation: bool
# external_elevation_id: Optional[uuid.UUID] = Field(foreign_key="externalelevation.id")
class ExternalElevationGableTwo(BaseModel, table=True):
is_there_a_fourth_external_elevation: bool
external_elevation_id: Optional[uuid.UUID] = Field(foreign_key="externalelevation.id")
# class ConservatoryOrOutbuilding(BaseModel, table=True):
# is_there_a_conservatory: bool
# is_there_a_cellar_present: bool
# is_there_an_outbuilding: bool
class ConservatoryOrOutbuilding(BaseModel, table=True):
is_there_a_conservatory: bool
is_there_a_cellar_present: bool
is_there_an_outbuilding: bool
# class AccessAndElevations(BaseModel, table=True):
# property_access_id: uuid.UUID = Field(foreign_key="propertyaccess.id")
# external_elevation_front_id: uuid.UUID = Field(foreign_key="externalelevationfront.id")
# external_elevation_back_id: uuid.UUID = Field(foreign_key="externalelevationrear.id")
# external_elevation_gable_one_id: uuid.UUID = Field(foreign_key="externalelevationgableone.id")
# external_elevation_gable_two_id: uuid.UUID = Field(foreign_key="externalelevationgabletwo.id")
# conservatory_or_out_building_id: uuid.UUID = Field(foreign_key="conservatoryoroutbuilding.id")
class AccessAndElevations(BaseModel, table=True):
property_access_id: uuid.UUID = Field(foreign_key="propertyaccess.id")
external_elevation_front_id: uuid.UUID = Field(foreign_key="externalelevationfront.id")
external_elevation_back_id: uuid.UUID = Field(foreign_key="externalelevationrear.id")
external_elevation_gable_one_id: uuid.UUID = Field(foreign_key="externalelevationgableone.id")
external_elevation_gable_two_id: uuid.UUID = Field(foreign_key="externalelevationgabletwo.id")
conservatory_or_out_building_id: uuid.UUID = Field(foreign_key="conservatoryoroutbuilding.id")
# property_access: PropertyAccess = Relationship()
# external_elevation_front: ExternalElevationFront = Relationship()
# external_elevation_back: ExternalElevationRear = Relationship()
# external_elevation_gable_one: ExternalElevationGableOne = Relationship()
# external_elevation_gable_two: ExternalElevationGableTwo = Relationship()
# conservatory_or_out_building: ConservatoryOrOutbuilding = Relationship()
property_access: PropertyAccess = Relationship()
external_elevation_front: ExternalElevationFront = Relationship()
external_elevation_back: ExternalElevationRear = Relationship()
external_elevation_gable_one: ExternalElevationGableOne = Relationship()
external_elevation_gable_two: ExternalElevationGableTwo = Relationship()
conservatory_or_out_building: ConservatoryOrOutbuilding = Relationship()
# class VentilationInfo(BaseModel, table=True):
# is_there_a_ventilation_system_present_in_the_room: bool
# any_damp_mould_or_excessive_condensation_within_the_room: bool
# are_there_sufficient_undercuts_on_the_closed_door: str
# is_there_any_open_flue_heating_appliances_within_the_room: bool
class VentilationInfo(BaseModel, table=True):
is_there_a_ventilation_system_present_in_the_room: bool
any_damp_mould_or_excessive_condensation_within_the_room: bool
are_there_sufficient_undercuts_on_the_closed_door: str
is_there_any_open_flue_heating_appliances_within_the_room: bool
# class WindowsInfo(BaseModel, table=True):
# does_the_room_have_any_windows: bool
# condition_of_the_windows: Optional[str] = None
# do_the_windows_have_trickle_vents: Optional[bool] = None
# are_the_windows_openable: Optional[bool] = None
# input_trickle_vent_product_code_or_measurement: Optional[str] = None
class WindowsInfo(BaseModel, table=True):
does_the_room_have_any_windows: bool
condition_of_the_windows: Optional[str] = None
do_the_windows_have_trickle_vents: Optional[bool] = None
are_the_windows_openable: Optional[bool] = None
input_trickle_vent_product_code_or_measurement: Optional[str] = None
# class RoomInfo(BaseModel, table=True):
# overall_condition_of_the_room: str
# does_the_room_have_any_defects: str
# are_there_any_sloped_ceiling_areas: Optional[bool] = None
class RoomInfo(BaseModel, table=True):
overall_condition_of_the_room: str
does_the_room_have_any_defects: str
are_there_any_sloped_ceiling_areas: Optional[bool] = None
# windows_info_id: uuid.UUID = Field(foreign_key="windowsinfo.id")
# ventilation_info_id: uuid.UUID = Field(foreign_key="ventilationinfo.id")
windows_info_id: uuid.UUID = Field(foreign_key="windowsinfo.id")
ventilation_info_id: uuid.UUID = Field(foreign_key="ventilationinfo.id")
# windows_info: WindowsInfo = Relationship()
# ventilation_info: VentilationInfo = Relationship()
windows_info: WindowsInfo = Relationship()
ventilation_info: VentilationInfo = Relationship()
# class Hallway(BaseModel, table=True):
# is_there_a_hallway: bool
# room_info_id: Optional[uuid.UUID] = Field(foreign_key="roominfo.id")
# room_info: Optional[RoomInfo] = Relationship()
class Hallway(BaseModel, table=True):
is_there_a_hallway: bool
room_info_id: Optional[uuid.UUID] = Field(foreign_key="roominfo.id")
room_info: Optional[RoomInfo] = Relationship()
# class LivingRoom(BaseModel, table=True):
# room_info_id: Optional[uuid.UUID] = Field(foreign_key="roominfo.id")
# room_info: Optional[RoomInfo] = Relationship()
class LivingRoom(BaseModel, table=True):
room_info_id: Optional[uuid.UUID] = Field(foreign_key="roominfo.id")
room_info: Optional[RoomInfo] = Relationship()
# class DiningRoom(BaseModel, table=True):
# is_there_a_dining_room: bool
# room_info_id: Optional[uuid.UUID] = Field(foreign_key="roominfo.id")
# room_info: Optional[RoomInfo] = Relationship()
class DiningRoom(BaseModel, table=True):
is_there_a_dining_room: bool
room_info_id: Optional[uuid.UUID] = Field(foreign_key="roominfo.id")
room_info: Optional[RoomInfo] = Relationship()
# class Kitchen(BaseModel, table=True):
# room_info_id: Optional[uuid.UUID] = Field(foreign_key="roominfo.id")
# room_info: Optional[RoomInfo] = Relationship()
# is_there_a_cooker_hood_present_in_the_room: bool
class Kitchen(BaseModel, table=True):
room_info_id: Optional[uuid.UUID] = Field(foreign_key="roominfo.id")
room_info: Optional[RoomInfo] = Relationship()
is_there_a_cooker_hood_present_in_the_room: bool
# class Utility(BaseModel, table=True):
# is_there_a_utility_room: bool
# room_info_id: Optional[uuid.UUID] = Field(foreign_key="roominfo.id")
# room_info: Optional[RoomInfo] = Relationship()
class Utility(BaseModel, table=True):
is_there_a_utility_room: bool
room_info_id: Optional[uuid.UUID] = Field(foreign_key="roominfo.id")
room_info: Optional[RoomInfo] = Relationship()
# class WC(BaseModel, table=True):
# is_there_a_seperated_wc: bool
# room_info_id: Optional[uuid.UUID] = Field(foreign_key="roominfo.id")
# room_info: Optional[RoomInfo] = Relationship()
class WC(BaseModel, table=True):
is_there_a_seperated_wc: bool
room_info_id: Optional[uuid.UUID] = Field(foreign_key="roominfo.id")
room_info: Optional[RoomInfo] = Relationship()
# class Landing(BaseModel, table=True):
# is_there_a_landing: bool
# room_info_id: Optional[uuid.UUID] = Field(foreign_key="roominfo.id")
# room_info: Optional[RoomInfo] = Relationship()
class Landing(BaseModel, table=True):
is_there_a_landing: bool
room_info_id: Optional[uuid.UUID] = Field(foreign_key="roominfo.id")
room_info: Optional[RoomInfo] = Relationship()
# class LoftSpace(BaseModel, table=True):
# is_the_main_loft_space_accessible: str
# is_there_more_than_one_loft_space: bool
class LoftSpace(BaseModel, table=True):
is_the_main_loft_space_accessible: str
is_there_more_than_one_loft_space: bool
# class RoomInRoof(BaseModel, table=True):
# is_there_a_room_in_roof: bool
# room_info_id: Optional[uuid.UUID] = Field(foreign_key="roominfo.id")
# room_info: Optional[RoomInfo] = Relationship()
class RoomInRoof(BaseModel, table=True):
is_there_a_room_in_roof: bool
room_info_id: Optional[uuid.UUID] = Field(foreign_key="roominfo.id")
room_info: Optional[RoomInfo] = Relationship()
# class Bedroom(BaseModel, table=True):
# double_or_single_bedroom: str
# room_info_id: Optional[uuid.UUID] = Field(foreign_key="roominfo.id")
# room_info: Optional[RoomInfo] = Relationship()
# rooms_id: uuid.UUID = Field(foreign_key="rooms.id")
# rooms: Optional["Rooms"] = Relationship(back_populates="bedrooms")
class Bedroom(BaseModel, table=True):
double_or_single_bedroom: str
room_info_id: Optional[uuid.UUID] = Field(foreign_key="roominfo.id")
room_info: Optional[RoomInfo] = Relationship()
rooms_id: uuid.UUID = Field(foreign_key="rooms.id")
rooms: Optional["Rooms"] = Relationship(back_populates="bedrooms")
# class Bathroom(BaseModel, table=True):
# is_this_an_ensuite_bathroom: bool
# room_info_id: Optional[uuid.UUID] = Field(foreign_key="roominfo.id")
# room_info: Optional[RoomInfo] = Relationship()
class Bathroom(BaseModel, table=True):
is_this_an_ensuite_bathroom: bool
room_info_id: Optional[uuid.UUID] = Field(foreign_key="roominfo.id")
room_info: Optional[RoomInfo] = Relationship()
# rooms_id: uuid.UUID = Field(foreign_key="rooms.id")
# rooms: Optional["Rooms"] = Relationship(back_populates="bathrooms")
rooms_id: uuid.UUID = Field(foreign_key="rooms.id")
rooms: Optional["Rooms"] = Relationship(back_populates="bathrooms")
# class Rooms(BaseModel, table=True):
# hallway_id: uuid.UUID = Field(foreign_key="hallway.id")
# hallway: Hallway = Relationship()
class Rooms(BaseModel, table=True):
hallway_id: uuid.UUID = Field(foreign_key="hallway.id")
hallway: Hallway = Relationship()
# living_room_id: uuid.UUID = Field(foreign_key="livingroom.id")
# living_room: LivingRoom = Relationship()
living_room_id: uuid.UUID = Field(foreign_key="livingroom.id")
living_room: LivingRoom = Relationship()
# dining_room_id: uuid.UUID = Field(foreign_key="diningroom.id")
# dining_room: DiningRoom = Relationship()
dining_room_id: uuid.UUID = Field(foreign_key="diningroom.id")
dining_room: DiningRoom = Relationship()
# kitchen_id: uuid.UUID = Field(foreign_key="kitchen.id")
# kitchen: Kitchen = Relationship()
kitchen_id: uuid.UUID = Field(foreign_key="kitchen.id")
kitchen: Kitchen = Relationship()
# utility_id: uuid.UUID = Field(foreign_key="utility.id")
# utility: Utility = Relationship()
utility_id: uuid.UUID = Field(foreign_key="utility.id")
utility: Utility = Relationship()
# wash_chamber_id: uuid.UUID = Field(foreign_key="wc.id")
# wash_chamber: WC = Relationship()
wash_chamber_id: uuid.UUID = Field(foreign_key="wc.id")
wash_chamber: WC = Relationship()
# landing_id: uuid.UUID = Field(foreign_key="landing.id")
# landing: Landing = Relationship()
landing_id: uuid.UUID = Field(foreign_key="landing.id")
landing: Landing = Relationship()
# loft_space_id: uuid.UUID = Field(foreign_key="loftspace.id")
# loft_space: LoftSpace = Relationship()
loft_space_id: uuid.UUID = Field(foreign_key="loftspace.id")
loft_space: LoftSpace = Relationship()
# room_in_roof_id: uuid.UUID = Field(foreign_key="roominroof.id")
# room_in_roof: RoomInRoof = Relationship()
room_in_roof_id: uuid.UUID = Field(foreign_key="roominroof.id")
room_in_roof: RoomInRoof = Relationship()
# bedrooms: List[Bedroom] = Relationship(back_populates="rooms")
# bathrooms: List[Bathroom] = Relationship(back_populates="rooms")
bedrooms: List[Bedroom] = Relationship(back_populates="rooms")
bathrooms: List[Bathroom] = Relationship(back_populates="rooms")
# class GeneralConditionHeatingSystem(BaseModel, table=True):
# is_the_heating_system_in_working_order: bool
# does_the_occupant_have_a_smart_meter: bool
# are_there_any_smart_monitoring_devices: bool
# are_the_gas_and_electricity_meters_accessible: bool
# dual_or_single_electric_meter: str
class GeneralConditionHeatingSystem(BaseModel, table=True):
is_the_heating_system_in_working_order: bool
does_the_occupant_have_a_smart_meter: bool
are_there_any_smart_monitoring_devices: bool
are_the_gas_and_electricity_meters_accessible: bool
dual_or_single_electric_meter: str
# class MainHeatingOne(BaseModel, table=True):
# as_defined_by: str
# fuel: str
# type: str
class MainHeatingOne(BaseModel, table=True):
as_defined_by: str
fuel: str
type: str
# class MainHeatingTwo(BaseModel, table=True):
# is_there_a_main_heating_two: bool
class MainHeatingTwo(BaseModel, table=True):
is_there_a_main_heating_two: bool
# class SecondaryHeating(BaseModel, table=True):
# is_there_a_secondary_heating: bool
# fuel: str
# electric_heating_type: str
# gas_heating_type: str
class SecondaryHeating(BaseModel, table=True):
is_there_a_secondary_heating: bool
fuel: str
electric_heating_type: str
gas_heating_type: str
# class HeatingByRoom(BaseModel, table=True):
# rooms_heated_by_main_system_one: List[str] = Field(sa_column=Column(JSON))
# rooms_heated_by_main_system_two: List[str] = Field(sa_column=Column(JSON))
# rooms_heated_by_secondary_heating: List[str] = Field(sa_column=Column(JSON))
# are_there_any_partially_heated_rooms: bool
# partially_heated_rooms: Optional[List[str]] = Field(sa_column=Column(JSON))
# are_there_any_unheated_rooms: bool
# unheated_rooms: List[str] = Field(sa_column=Column(JSON))
class HeatingByRoom(BaseModel, table=True):
rooms_heated_by_main_system_one: List[str] = Field(sa_column=Column(JSON))
rooms_heated_by_main_system_two: List[str] = Field(sa_column=Column(JSON))
rooms_heated_by_secondary_heating: List[str] = Field(sa_column=Column(JSON))
are_there_any_partially_heated_rooms: bool
partially_heated_rooms: Optional[List[str]] = Field(sa_column=Column(JSON))
are_there_any_unheated_rooms: bool
unheated_rooms: List[str] = Field(sa_column=Column(JSON))
# class Renewables(BaseModel, table=True):
# is_there_any_renewable_energy_system_in_place: bool
# suitable_roof_orientation_for_solar_pv_water: str
# is_there_a_water_tank: bool
# type: str
# size: str
# tank_location: str
# is_the_tank_insulated: bool
# type_of_insulation: str
# thickness_of_insulation_in_mm: int
class Renewables(BaseModel, table=True):
is_there_any_renewable_energy_system_in_place: bool
suitable_roof_orientation_for_solar_pv_water: str
is_there_a_water_tank: bool
type: str
size: str
tank_location: str
is_the_tank_insulated: bool
type_of_insulation: str
thickness_of_insulation_in_mm: int
# class HeatingSystem(BaseModel, table=True):
# general_condition_id: uuid.UUID = Field(foreign_key="generalconditionheatingsystem.id")
# general_condition: GeneralConditionHeatingSystem = Relationship()
class HeatingSystem(BaseModel, table=True):
general_condition_id: uuid.UUID = Field(foreign_key="generalconditionheatingsystem.id")
general_condition: GeneralConditionHeatingSystem = Relationship()
# main_heating_one_id: uuid.UUID = Field(foreign_key="mainheatingone.id")
# main_heating_one: MainHeatingOne = Relationship()
main_heating_one_id: uuid.UUID = Field(foreign_key="mainheatingone.id")
main_heating_one: MainHeatingOne = Relationship()
# main_heating_two_id: uuid.UUID = Field(foreign_key="mainheatingtwo.id")
# main_heating_two: MainHeatingTwo = Relationship()
main_heating_two_id: uuid.UUID = Field(foreign_key="mainheatingtwo.id")
main_heating_two: MainHeatingTwo = Relationship()
# secondary_heating_id: uuid.UUID = Field(foreign_key="secondaryheating.id")
# secondary_heating: SecondaryHeating = Relationship()
secondary_heating_id: uuid.UUID = Field(foreign_key="secondaryheating.id")
secondary_heating: SecondaryHeating = Relationship()
# heating_by_room_id: uuid.UUID = Field(foreign_key="heatingbyroom.id")
# heating_by_room: HeatingByRoom = Relationship()
heating_by_room_id: uuid.UUID = Field(foreign_key="heatingbyroom.id")
heating_by_room: HeatingByRoom = Relationship()
# renewables_id: uuid.UUID = Field(foreign_key="renewables.id")
# renewables: Renewables = Relationship()
renewables_id: uuid.UUID = Field(foreign_key="renewables.id")
renewables: Renewables = Relationship()
# class Occupant(BaseModel, table=True):
# name: str
# have_evidence_of_12_months_of_fuel_bill_data: bool
# total_number_of_occupants: int
# no_of_adult_occupants: int
# no_of_child_occupants: int
# no_of_occupant_of_a_pensionable_age: int
# are_there_any_vulnerable_people: bool
# is_there_anyone_with_a_disability: bool
# status_of_occupant: str
# landlord_wrote_that_the_tenent_agrees_assessment_been_supplied: bool
class Occupant(BaseModel, table=True):
name: str
have_evidence_of_12_months_of_fuel_bill_data: bool
total_number_of_occupants: int
no_of_adult_occupants: int
no_of_child_occupants: int
no_of_occupant_of_a_pensionable_age: int
are_there_any_vulnerable_people: bool
is_there_anyone_with_a_disability: bool
status_of_occupant: str
landlord_wrote_that_the_tenent_agrees_assessment_been_supplied: bool
# class EnergyUse(BaseModel, table=True):
# property_tenure: str
# who_is_the_electricity_payer: str
class EnergyUse(BaseModel, table=True):
property_tenure: str
who_is_the_electricity_payer: str
# class HeatingFromConditionReport(BaseModel, table=True):
# room_stat_in_temperature_in_celsius: Optional[str] = None
# room_stat_location: Optional[str] = None
# is_the_heating_pattern_known: Optional[str] = None
class HeatingFromConditionReport(BaseModel, table=True):
room_stat_in_temperature_in_celsius: Optional[str] = None
room_stat_location: Optional[str] = None
is_the_heating_pattern_known: Optional[str] = None
# class ShowerAndBath(BaseModel, table=True):
# shower_type: str
# do_you_know_the_no_of_showers_per_day_per_week: bool
# please_input_no_of_showers_and_specify_a_day_or_a_week: str
# do_you_know_the_number_of_baths_per_day_or_per_week: str
class ShowerAndBath(BaseModel, table=True):
shower_type: str
do_you_know_the_no_of_showers_per_day_per_week: bool
please_input_no_of_showers_and_specify_a_day_or_a_week: str
do_you_know_the_number_of_baths_per_day_or_per_week: str
# class FridgeAndFreezers(BaseModel, table=True):
# no_of_stand_alone_seperate_fridges: int
# no_of_stand_alone_seperate_freezers: int
# no_of_stand_alone_or_integrated_fridge_freezers: int
class FridgeAndFreezers(BaseModel, table=True):
no_of_stand_alone_seperate_fridges: int
no_of_stand_alone_seperate_freezers: int
no_of_stand_alone_or_integrated_fridge_freezers: int
# class Cooker(BaseModel,table=True):
# range_fuel: str
# normal_large_range: str
# cooker_type: str
class Cooker(BaseModel,table=True):
range_fuel: str
normal_large_range: str
cooker_type: str
# class TumbleDryer(BaseModel, table=True):
# percentage_of_annual_use: int
# space_for_outdoor_drying: bool
class TumbleDryer(BaseModel, table=True):
percentage_of_annual_use: int
space_for_outdoor_drying: bool
# class OccupantAssessment(BaseModel, table=True):
# occupant_id: uuid.UUID = Field(foreign_key="occupant.id")
# occupant: Occupant = Relationship()
class OccupantAssessment(BaseModel, table=True):
occupant_id: uuid.UUID = Field(foreign_key="occupant.id")
occupant: Occupant = Relationship()
# energy_use_id: uuid.UUID = Field(foreign_key="energyuse.id")
# energy_use: EnergyUse = Relationship()
energy_use_id: uuid.UUID = Field(foreign_key="energyuse.id")
energy_use: EnergyUse = Relationship()
# heating_id: uuid.UUID = Field(foreign_key="heatingfromconditionreport.id")
# heating: HeatingFromConditionReport = Relationship()
heating_id: uuid.UUID = Field(foreign_key="heatingfromconditionreport.id")
heating: HeatingFromConditionReport = Relationship()
# shower_and_bath_id: uuid.UUID = Field(foreign_key="showerandbath.id")
# shower_and_bath: ShowerAndBath = Relationship()
shower_and_bath_id: uuid.UUID = Field(foreign_key="showerandbath.id")
shower_and_bath: ShowerAndBath = Relationship()
# # appliances: Optional[Appliances]
# # appliances_id
# appliances: Optional[Appliances]
# appliances_id
# fridge_and_freezers_id: uuid.UUID = Field(foreign_key="fridgeandfreezers.id")
# fridge_and_freezers: FridgeAndFreezers = Relationship()
fridge_and_freezers_id: uuid.UUID = Field(foreign_key="fridgeandfreezers.id")
fridge_and_freezers: FridgeAndFreezers = Relationship()
# cooker_id: uuid.UUID = Field(foreign_key="cooker.id")
# cooker: Cooker = Relationship()
cooker_id: uuid.UUID = Field(foreign_key="cooker.id")
cooker: Cooker = Relationship()
# tumble_dryer_id: uuid.UUID = Field(foreign_key="tumbledryer.id")
# tumble_dryer: TumbleDryer = Relationship()
tumble_dryer_id: uuid.UUID = Field(foreign_key="tumbledryer.id")
tumble_dryer: TumbleDryer = Relationship()
# class ConditionReportModel(BaseModel, table=True):
# project_site_name: str
# property_reference_code: str
# property_address: str
# postcode: str
class ConditionReportModel(BaseModel, table=True):
project_site_name: str
property_reference_code: str
property_address: str
postcode: str
# general_information_id: uuid.UUID = Field(foreign_key="generalinformation.id")
# general_information: GeneralInformation = Relationship()
general_information_id: uuid.UUID = Field(foreign_key="generalinformation.id")
general_information: GeneralInformation = Relationship()
# access_and_elevations_id: uuid.UUID = Field(foreign_key="accessandelevations.id")
# access_and_elevations: AccessAndElevations = Relationship()
access_and_elevations_id: uuid.UUID = Field(foreign_key="accessandelevations.id")
access_and_elevations: AccessAndElevations = Relationship()
# rooms_id: uuid.UUID = Field(foreign_key="rooms.id")
# rooms: Rooms = Relationship()
rooms_id: uuid.UUID = Field(foreign_key="rooms.id")
rooms: Rooms = Relationship()
# heating_system_id: uuid.UUID = Field(foreign_key="heatingsystem.id")
# heating_system: HeatingSystem = Relationship()
heating_system_id: uuid.UUID = Field(foreign_key="heatingsystem.id")
heating_system: HeatingSystem = Relationship()
# occupancy_assessment_id: uuid.UUID = Field(foreign_key="occupantassessment.id")
# occupancy_assessment: OccupantAssessment = Relationship()
occupancy_assessment_id: uuid.UUID = Field(foreign_key="occupantassessment.id")
occupancy_assessment: OccupantAssessment = Relationship()

View file

@ -1,335 +1,335 @@
# from sqlmodel import Field, SQLModel, Relationship
# import uuid
# from typing import Optional, List
# from datetime import datetime
# from pydantic import EmailStr
# from sqlalchemy import Column
# from sqlalchemy.dialects.postgresql import UUID
# from etl.models.topLevel import BaseModel, Documents
from sqlmodel import Field, SQLModel, Relationship
import uuid
from typing import Optional, List
from datetime import datetime
from pydantic import EmailStr
from sqlalchemy import Column
from sqlalchemy.dialects.postgresql import UUID
from etl.models.topLevel import BaseModel, Documents
# class PreSiteNote(BaseModel, table=True):
# summary_info_id: uuid.UUID = Field(
# foreign_key="presitenotessummaryinfo.id",
# nullable=False
# )
class PreSiteNote(BaseModel, table=True):
summary_info_id: uuid.UUID = Field(
foreign_key="presitenotessummaryinfo.id",
nullable=False
)
# summary_info: Optional["PreSiteNotesSummaryInfo"] = Relationship(back_populates="pre_site_notes")
summary_info: Optional["PreSiteNotesSummaryInfo"] = Relationship(back_populates="pre_site_notes")
# # Assessor Info
# assessor_id: uuid.UUID = Field(
# foreign_key="assessorinfo.id",
# nullable=False
# )
# Assessor Info
assessor_id: uuid.UUID = Field(
foreign_key="assessorinfo.id",
nullable=False
)
# assessor: Optional["AssessorInfo"] = Relationship(back_populates="pre_site_notes")
assessor: Optional["AssessorInfo"] = Relationship(back_populates="pre_site_notes")
# pre_site_note_description_id: uuid.UUID = Field(
# foreign_key="propertydescription.id",
# nullable=True
# )
pre_site_note_description_id: uuid.UUID = Field(
foreign_key="propertydescription.id",
nullable=True
)
# pre_site_note_description: Optional["PropertyDescription"] = Relationship(back_populates="pre_site_notes")
pre_site_note_description: Optional["PropertyDescription"] = Relationship(back_populates="pre_site_notes")
# class Dimension(BaseModel, table=True):
# floor_area_m2: float
# room_height_m: float
# loss_perimeter_m: float
# party_wall_length_m: float
# property_detail_id: Optional[uuid.UUID] = Field(default=None, foreign_key="propertydetail.id")
# property_detail: Optional["PropertyDetail"] = Relationship(back_populates="dimensions")
class Dimension(BaseModel, table=True):
floor_area_m2: float
room_height_m: float
loss_perimeter_m: float
party_wall_length_m: float
property_detail_id: Optional[uuid.UUID] = Field(default=None, foreign_key="propertydetail.id")
property_detail: Optional["PropertyDetail"] = Relationship(back_populates="dimensions")
# class Walls(BaseModel, table=True):
# construction: str
# insulation: str
# insulation_thickness_mm: str
# wall_thickness_measured: bool
# wall_thickness_mm: Optional[int]
# u_value_known: bool
# u_value_w_m2_k: Optional[float]
# dry_lining: bool
# alternative_wall_present: bool
class Walls(BaseModel, table=True):
construction: str
insulation: str
insulation_thickness_mm: str
wall_thickness_measured: bool
wall_thickness_mm: Optional[int]
u_value_known: bool
u_value_w_m2_k: Optional[float]
dry_lining: bool
alternative_wall_present: bool
# class Roofs(BaseModel, table=True):
# construction: str
# insulation_type: str
# insulation_thickness: str
# u_value_known: bool
class Roofs(BaseModel, table=True):
construction: str
insulation_type: str
insulation_thickness: str
u_value_known: bool
# class Floors(BaseModel, table=True):
# floor_type: str
# ground_floor_construction: str
# ground_floor_insulation_type: Optional[str] = ""
# floor_insulation_thickness_mm: Optional[float] = -1
# u_value_known: bool
class Floors(BaseModel, table=True):
floor_type: str
ground_floor_construction: str
ground_floor_insulation_type: Optional[str] = ""
floor_insulation_thickness_mm: Optional[float] = -1
u_value_known: bool
# class Windows(BaseModel, table=True):
# glazing_type: str
# area_m2: float
# roof_window: bool
# orientation: str
# u_value_w_m2_k: int
# g_value: int
# property_detail_id: Optional[uuid.UUID] = Field(default=None, foreign_key="propertydetail.id")
# property_detail: Optional["PropertyDetail"] = Relationship(back_populates="windows")
class Windows(BaseModel, table=True):
glazing_type: str
area_m2: float
roof_window: bool
orientation: str
u_value_w_m2_k: int
g_value: int
property_detail_id: Optional[uuid.UUID] = Field(default=None, foreign_key="propertydetail.id")
property_detail: Optional["PropertyDetail"] = Relationship(back_populates="windows")
# class PropertyDetail(BaseModel, table=True):
# age_band: str
# wall_id: Optional[uuid.UUID] = Field(default=None, foreign_key="walls.id")
# roof_id: Optional[uuid.UUID] = Field(default=None, foreign_key="roofs.id")
# floor_id: Optional[uuid.UUID] = Field(default=None, foreign_key="floors.id")
class PropertyDetail(BaseModel, table=True):
age_band: str
wall_id: Optional[uuid.UUID] = Field(default=None, foreign_key="walls.id")
roof_id: Optional[uuid.UUID] = Field(default=None, foreign_key="roofs.id")
floor_id: Optional[uuid.UUID] = Field(default=None, foreign_key="floors.id")
# # Relationships
# dimensions: List[Dimension] = Relationship(back_populates="property_detail")
# windows: List[Windows] = Relationship(back_populates="property_detail")
# Relationships
dimensions: List[Dimension] = Relationship(back_populates="property_detail")
windows: List[Windows] = Relationship(back_populates="property_detail")
# class Door(BaseModel, table=True):
# no_of_doors: int
# no_of_insulated_doors: int
# u_value_w_m2_k: Optional[str]
class Door(BaseModel, table=True):
no_of_doors: int
no_of_insulated_doors: int
u_value_w_m2_k: Optional[str]
# property_description: Optional["PropertyDescription"] = Relationship(back_populates="door")
property_description: Optional["PropertyDescription"] = Relationship(back_populates="door")
# class VentilationAndCooling(BaseModel, table=True):
# no_of_open_fireplaces: int
# ventilation_type: str
# space_cooling_system_present: bool
class VentilationAndCooling(BaseModel, table=True):
no_of_open_fireplaces: int
ventilation_type: str
space_cooling_system_present: bool
# property_description: Optional["PropertyDescription"] = Relationship(back_populates="ventilation_and_cooling")
property_description: Optional["PropertyDescription"] = Relationship(back_populates="ventilation_and_cooling")
# class Lighting(BaseModel, table=True):
# total_no_of_light_fittings: int
# total_no_of_lel_fittings: int
class Lighting(BaseModel, table=True):
total_no_of_light_fittings: int
total_no_of_lel_fittings: int
# property_description: Optional["PropertyDescription"] = Relationship(back_populates="lighting")
property_description: Optional["PropertyDescription"] = Relationship(back_populates="lighting")
# class HeatingSystemControls(BaseModel, table=True):
# control_type: str
# flue_type: str
# fan_assisted_flue: bool
# heat_emitter_type: str
# electricity_meter_type: Optional[str] = ""
# mains_gas_available: Optional[bool] = False
class HeatingSystemControls(BaseModel, table=True):
control_type: str
flue_type: str
fan_assisted_flue: bool
heat_emitter_type: str
electricity_meter_type: Optional[str] = ""
mains_gas_available: Optional[bool] = False
# class HeatingFromPreSiteNotes(BaseModel, table=True):
# type: str
# heating_source: str
# efficiency_source: str
# heating_fuel: str
# brand_name: str
# model_name: str
# model_qualifer: str
# sap_2009_table: Optional[str] = ""
# percentage_of_heated_floor_area_served: Optional[str] = ""
# controls_id: Optional[uuid.UUID] = Field(default=None, foreign_key="heatingsystemcontrols.id")
class HeatingFromPreSiteNotes(BaseModel, table=True):
type: str
heating_source: str
efficiency_source: str
heating_fuel: str
brand_name: str
model_name: str
model_qualifer: str
sap_2009_table: Optional[str] = ""
percentage_of_heated_floor_area_served: Optional[str] = ""
controls_id: Optional[uuid.UUID] = Field(default=None, foreign_key="heatingsystemcontrols.id")
# property_description: Optional["PropertyDescription"] = Relationship(
# back_populates="main_heating", sa_relationship_kwargs={"foreign_keys": "[PropertyDescription.main_heating_id]"}
# )
# property_description2: Optional["PropertyDescription"] = Relationship(
# back_populates="main_heating2", sa_relationship_kwargs={"foreign_keys": "[PropertyDescription.main_heating2_id]"}
# )
property_description: Optional["PropertyDescription"] = Relationship(
back_populates="main_heating", sa_relationship_kwargs={"foreign_keys": "[PropertyDescription.main_heating_id]"}
)
property_description2: Optional["PropertyDescription"] = Relationship(
back_populates="main_heating2", sa_relationship_kwargs={"foreign_keys": "[PropertyDescription.main_heating2_id]"}
)
# class HeatingType(BaseModel, table=True):
# heating_type: str
# fuel_type: str
class HeatingType(BaseModel, table=True):
heating_type: str
fuel_type: str
# property_description: Optional["PropertyDescription"] = Relationship(back_populates="secondary_heating_type")
property_description: Optional["PropertyDescription"] = Relationship(back_populates="secondary_heating_type")
# class WaterHeating(BaseModel, table=True):
# heating_type: str
# fuel_type: str
class WaterHeating(BaseModel, table=True):
heating_type: str
fuel_type: str
# property_description: Optional["PropertyDescription"] = Relationship(back_populates="water_heating")
property_description: Optional["PropertyDescription"] = Relationship(back_populates="water_heating")
# class HotWaterCylinder(BaseModel, table=True):
# volume: str
# insulation_type: str
# insulation_thickness: str
# thermostat: bool
class HotWaterCylinder(BaseModel, table=True):
volume: str
insulation_type: str
insulation_thickness: str
thermostat: bool
# property_description: Optional["PropertyDescription"] = Relationship(back_populates="hot_water_cylinder")
property_description: Optional["PropertyDescription"] = Relationship(back_populates="hot_water_cylinder")
# class SolarWaterHeating(BaseModel, table=True):
# solar_water_heating_details_known: bool
class SolarWaterHeating(BaseModel, table=True):
solar_water_heating_details_known: bool
# property_description: Optional["PropertyDescription"] = Relationship(back_populates="solar_water_heating")
property_description: Optional["PropertyDescription"] = Relationship(back_populates="solar_water_heating")
# class ShowerAndBaths(BaseModel, table=True):
# no_of_rooms_with_baths_and_or_shower: int
# no_of_rooms_with_mixer_shower_and_no_baths: int
# no_of_rooms_with_mixer_shower_and_baths: int
class ShowerAndBaths(BaseModel, table=True):
no_of_rooms_with_baths_and_or_shower: int
no_of_rooms_with_mixer_shower_and_no_baths: int
no_of_rooms_with_mixer_shower_and_baths: int
# property_description: Optional["PropertyDescription"] = Relationship(back_populates="shower_and_baths")
property_description: Optional["PropertyDescription"] = Relationship(back_populates="shower_and_baths")
# class FlueGasHeatRecoverySystem(BaseModel, table=True):
# fghrs_present: bool
class FlueGasHeatRecoverySystem(BaseModel, table=True):
fghrs_present: bool
# property_description: Optional["PropertyDescription"] = Relationship(back_populates="flue_gas_heat_recovery_system")
property_description: Optional["PropertyDescription"] = Relationship(back_populates="flue_gas_heat_recovery_system")
# class PhotovoltaicPanel(BaseModel, table=True):
# pvs_are_connected_to_dwelling_electricity_meter: bool
# percentage_of_external_roof_area_with_pvs: str
class PhotovoltaicPanel(BaseModel, table=True):
pvs_are_connected_to_dwelling_electricity_meter: bool
percentage_of_external_roof_area_with_pvs: str
# property_description: Optional["PropertyDescription"] = Relationship(back_populates="photovoltaic_panel")
property_description: Optional["PropertyDescription"] = Relationship(back_populates="photovoltaic_panel")
# class WindTurbine(BaseModel, table=True):
# wind_turbine: bool
class WindTurbine(BaseModel, table=True):
wind_turbine: bool
# property_description: Optional["PropertyDescription"] = Relationship(back_populates="wind_turbine")
property_description: Optional["PropertyDescription"] = Relationship(back_populates="wind_turbine")
# class OtherDetails(BaseModel, table=True):
# electricity_meter_type: str
# main_gas_avalible: bool
class OtherDetails(BaseModel, table=True):
electricity_meter_type: str
main_gas_avalible: bool
# property_description: Optional["PropertyDescription"] = Relationship(back_populates="other_details")
property_description: Optional["PropertyDescription"] = Relationship(back_populates="other_details")
# class PropertyDescription(BaseModel, table=True):
# built_form: str
# detachment_or_position: str
# no_of_main_property: int
# no_of_extension_1: Optional[int] = 0
# no_of_extension_2: Optional[int] = 0
# no_of_extension_3: Optional[int] = 0
# no_of_extension_4: Optional[int] = 0
# no_of_habitable_rooms: int
# no_of_heated_rooms: int
# heated_basement: bool
# conservatory_type: str
# percentage_of_draught_proofed: int
# terrain_type: str
# conservatory: bool
class PropertyDescription(BaseModel, table=True):
built_form: str
detachment_or_position: str
no_of_main_property: int
no_of_extension_1: Optional[int] = 0
no_of_extension_2: Optional[int] = 0
no_of_extension_3: Optional[int] = 0
no_of_extension_4: Optional[int] = 0
no_of_habitable_rooms: int
no_of_heated_rooms: int
heated_basement: bool
conservatory_type: str
percentage_of_draught_proofed: int
terrain_type: str
conservatory: bool
# main_property_id: uuid.UUID = Field(foreign_key="propertydetail.id")
# ex1_property_id: Optional[uuid.UUID] = Field(default=None, foreign_key="propertydetail.id")
# ex2_property_id: Optional[uuid.UUID] = Field(default=None, foreign_key="propertydetail.id")
# ex3_property_id: Optional[uuid.UUID] = Field(default=None, foreign_key="propertydetail.id")
# ex4_property_id: Optional[uuid.UUID] = Field(default=None, foreign_key="propertydetail.id")
main_property_id: uuid.UUID = Field(foreign_key="propertydetail.id")
ex1_property_id: Optional[uuid.UUID] = Field(default=None, foreign_key="propertydetail.id")
ex2_property_id: Optional[uuid.UUID] = Field(default=None, foreign_key="propertydetail.id")
ex3_property_id: Optional[uuid.UUID] = Field(default=None, foreign_key="propertydetail.id")
ex4_property_id: Optional[uuid.UUID] = Field(default=None, foreign_key="propertydetail.id")
# door_id: Optional[uuid.UUID] = Field(default=None, foreign_key="door.id")
# ventilation_and_cooling_id: Optional[uuid.UUID] = Field(default=None, foreign_key="ventilationandcooling.id")
# lighting_id: Optional[uuid.UUID] = Field(default=None, foreign_key="lighting.id")
# water_heating_id: Optional[uuid.UUID] = Field(default=None, foreign_key="waterheating.id")
# hot_water_cylinder_id: Optional[uuid.UUID] = Field(default=None, foreign_key="hotwatercylinder.id")
# solar_water_heating_id: Optional[uuid.UUID] = Field(default=None, foreign_key="solarwaterheating.id")
# shower_and_baths_id: Optional[uuid.UUID] = Field(default=None, foreign_key="showerandbaths.id")
# flue_gas_heat_recovery_system_id: Optional[uuid.UUID] = Field(default=None, foreign_key="fluegasheatrecoverysystem.id")
# photovoltaic_panel_id: Optional[uuid.UUID] = Field(default=None, foreign_key="photovoltaicpanel.id")
# wind_turbine_id: Optional[uuid.UUID] = Field(default=None, foreign_key="windturbine.id")
# other_details_id: Optional[uuid.UUID] = Field(default=None, foreign_key="otherdetails.id")
# main_heating_id: Optional[uuid.UUID] = Field(default=None, foreign_key="heatingfrompresitenotes.id")
# main_heating2_id: Optional[uuid.UUID] = Field(default=None, foreign_key="heatingfrompresitenotes.id")
# secondary_heating_type_id: Optional[uuid.UUID] = Field(default=None, foreign_key="heatingtype.id")
door_id: Optional[uuid.UUID] = Field(default=None, foreign_key="door.id")
ventilation_and_cooling_id: Optional[uuid.UUID] = Field(default=None, foreign_key="ventilationandcooling.id")
lighting_id: Optional[uuid.UUID] = Field(default=None, foreign_key="lighting.id")
water_heating_id: Optional[uuid.UUID] = Field(default=None, foreign_key="waterheating.id")
hot_water_cylinder_id: Optional[uuid.UUID] = Field(default=None, foreign_key="hotwatercylinder.id")
solar_water_heating_id: Optional[uuid.UUID] = Field(default=None, foreign_key="solarwaterheating.id")
shower_and_baths_id: Optional[uuid.UUID] = Field(default=None, foreign_key="showerandbaths.id")
flue_gas_heat_recovery_system_id: Optional[uuid.UUID] = Field(default=None, foreign_key="fluegasheatrecoverysystem.id")
photovoltaic_panel_id: Optional[uuid.UUID] = Field(default=None, foreign_key="photovoltaicpanel.id")
wind_turbine_id: Optional[uuid.UUID] = Field(default=None, foreign_key="windturbine.id")
other_details_id: Optional[uuid.UUID] = Field(default=None, foreign_key="otherdetails.id")
main_heating_id: Optional[uuid.UUID] = Field(default=None, foreign_key="heatingfrompresitenotes.id")
main_heating2_id: Optional[uuid.UUID] = Field(default=None, foreign_key="heatingfrompresitenotes.id")
secondary_heating_type_id: Optional[uuid.UUID] = Field(default=None, foreign_key="heatingtype.id")
# # Relationships
# main_property: Optional["PropertyDetail"] = Relationship(sa_relationship_kwargs={"foreign_keys": "[PropertyDescription.main_property_id]"})
# ex1_property: Optional["PropertyDetail"] = Relationship(sa_relationship_kwargs={"foreign_keys": "[PropertyDescription.ex1_property_id]"})
# ex2_property: Optional["PropertyDetail"] = Relationship(sa_relationship_kwargs={"foreign_keys": "[PropertyDescription.ex2_property_id]"})
# ex3_property: Optional["PropertyDetail"] = Relationship(sa_relationship_kwargs={"foreign_keys": "[PropertyDescription.ex3_property_id]"})
# ex4_property: Optional["PropertyDetail"] = Relationship(sa_relationship_kwargs={"foreign_keys": "[PropertyDescription.ex4_property_id]"})
# Relationships
main_property: Optional["PropertyDetail"] = Relationship(sa_relationship_kwargs={"foreign_keys": "[PropertyDescription.main_property_id]"})
ex1_property: Optional["PropertyDetail"] = Relationship(sa_relationship_kwargs={"foreign_keys": "[PropertyDescription.ex1_property_id]"})
ex2_property: Optional["PropertyDetail"] = Relationship(sa_relationship_kwargs={"foreign_keys": "[PropertyDescription.ex2_property_id]"})
ex3_property: Optional["PropertyDetail"] = Relationship(sa_relationship_kwargs={"foreign_keys": "[PropertyDescription.ex3_property_id]"})
ex4_property: Optional["PropertyDetail"] = Relationship(sa_relationship_kwargs={"foreign_keys": "[PropertyDescription.ex4_property_id]"})
# # Related Models
# door: Optional["Door"] = Relationship(back_populates="property_description")
# ventilation_and_cooling: Optional["VentilationAndCooling"] = Relationship(back_populates="property_description")
# lighting: Optional["Lighting"] = Relationship(back_populates="property_description")
# water_heating: Optional["WaterHeating"] = Relationship(back_populates="property_description")
# hot_water_cylinder: Optional["HotWaterCylinder"] = Relationship(back_populates="property_description")
# solar_water_heating: Optional["SolarWaterHeating"] = Relationship(back_populates="property_description")
# shower_and_baths: Optional["ShowerAndBaths"] = Relationship(back_populates="property_description")
# flue_gas_heat_recovery_system: Optional["FlueGasHeatRecoverySystem"] = Relationship(back_populates="property_description")
# photovoltaic_panel: Optional["PhotovoltaicPanel"] = Relationship(back_populates="property_description")
# wind_turbine: Optional["WindTurbine"] = Relationship(back_populates="property_description")
# other_details: Optional["OtherDetails"] = Relationship(back_populates="property_description")
# main_heating: Optional["HeatingFromPreSiteNotes"] = Relationship(back_populates="property_description", sa_relationship_kwargs={"foreign_keys": "[PropertyDescription.main_heating_id]"})
# main_heating2: Optional["HeatingFromPreSiteNotes"] = Relationship(back_populates="property_description", sa_relationship_kwargs={"foreign_keys": "[PropertyDescription.main_heating2_id]"})
# secondary_heating_type: Optional["HeatingType"] = Relationship(back_populates="property_description")
# Related Models
door: Optional["Door"] = Relationship(back_populates="property_description")
ventilation_and_cooling: Optional["VentilationAndCooling"] = Relationship(back_populates="property_description")
lighting: Optional["Lighting"] = Relationship(back_populates="property_description")
water_heating: Optional["WaterHeating"] = Relationship(back_populates="property_description")
hot_water_cylinder: Optional["HotWaterCylinder"] = Relationship(back_populates="property_description")
solar_water_heating: Optional["SolarWaterHeating"] = Relationship(back_populates="property_description")
shower_and_baths: Optional["ShowerAndBaths"] = Relationship(back_populates="property_description")
flue_gas_heat_recovery_system: Optional["FlueGasHeatRecoverySystem"] = Relationship(back_populates="property_description")
photovoltaic_panel: Optional["PhotovoltaicPanel"] = Relationship(back_populates="property_description")
wind_turbine: Optional["WindTurbine"] = Relationship(back_populates="property_description")
other_details: Optional["OtherDetails"] = Relationship(back_populates="property_description")
main_heating: Optional["HeatingFromPreSiteNotes"] = Relationship(back_populates="property_description", sa_relationship_kwargs={"foreign_keys": "[PropertyDescription.main_heating_id]"})
main_heating2: Optional["HeatingFromPreSiteNotes"] = Relationship(back_populates="property_description", sa_relationship_kwargs={"foreign_keys": "[PropertyDescription.main_heating2_id]"})
secondary_heating_type: Optional["HeatingType"] = Relationship(back_populates="property_description")
# pre_site_notes: Optional["PreSiteNote"] = Relationship(back_populates="pre_site_note_description")
pre_site_notes: Optional["PreSiteNote"] = Relationship(back_populates="pre_site_note_description")
# class AssessorInfo(BaseModel, table=True):
# accreditation_number: str
# name: str
# phone_number: Optional[str] = None
# email_address: Optional[EmailStr] = None
class AssessorInfo(BaseModel, table=True):
accreditation_number: str
name: str
phone_number: Optional[str] = None
email_address: Optional[EmailStr] = None
# company_id: Optional[uuid.UUID] = Field(default=None, foreign_key="companyinfo.id")
# company: Optional["CompanyInfo"] = Relationship(back_populates="assessors")
# pre_site_notes: List["PreSiteNote"] = Relationship(back_populates="assessor")
# documents: List["Documents"] = Relationship(back_populates="author")
# class PreSiteNotesSummaryInfo(BaseModel, table=True):
# reference_number: str
# epc_language: str
# uprn: Optional[str] = ""
# postcode: str
# region: str
# address: str
# town: str
# county: Optional[str] = ""
# property_tenure: str
# transaction_type: str
# inspection_date: datetime
# current_sap: str
# potential_sap: str
# current_ei: str
# potential_ei: str
# current_annual_emissions: str
# current_annual_emission_including_0925_multiplayer: str
# current_annual_energy_costs: str
# pre_site_notes: List["PreSiteNote"] = Relationship(back_populates="summary_info")
# class CompanyInfo(BaseModel, table=True):
# address: str
# trading_name: str
# post_code: str
# fax_number: Optional[str] = None
# related_party_disclosure: Optional[str] = None
# assessors: List[AssessorInfo] = Relationship(back_populates="company")
# class Insulation(BaseModel, table=True):
# type: str
# PreSiteNote.update_forward_refs()
# AssessorInfo.update_forward_refs()
company_id: Optional[uuid.UUID] = Field(default=None, foreign_key="companyinfo.id")
company: Optional["CompanyInfo"] = Relationship(back_populates="assessors")
pre_site_notes: List["PreSiteNote"] = Relationship(back_populates="assessor")
documents: List["Documents"] = Relationship(back_populates="author")
class PreSiteNotesSummaryInfo(BaseModel, table=True):
reference_number: str
epc_language: str
uprn: Optional[str] = ""
postcode: str
region: str
address: str
town: str
county: Optional[str] = ""
property_tenure: str
transaction_type: str
inspection_date: datetime
current_sap: str
potential_sap: str
current_ei: str
potential_ei: str
current_annual_emissions: str
current_annual_emission_including_0925_multiplayer: str
current_annual_energy_costs: str
pre_site_notes: List["PreSiteNote"] = Relationship(back_populates="summary_info")
class CompanyInfo(BaseModel, table=True):
address: str
trading_name: str
post_code: str
fax_number: Optional[str] = None
related_party_disclosure: Optional[str] = None
assessors: List[AssessorInfo] = Relationship(back_populates="company")
class Insulation(BaseModel, table=True):
type: str
PreSiteNote.update_forward_refs()
AssessorInfo.update_forward_refs()

View file

@ -22,38 +22,38 @@ class BaseModel(SQLModel):
)
)
# class Buildings(BaseModel, table=True):
# address: str
# postcode: str
# UPRN: str
# landlord_id: str
# domna_id: str
class Buildings(BaseModel, table=True):
address: str
postcode: str
UPRN: str
landlord_id: str
domna_id: str
# documents: List["Documents"] = Relationship(back_populates="building")
documents: List["Documents"] = Relationship(back_populates="building")
# class Documents(BaseModel, table=True):
# assessor_id: uuid.UUID = Field(
# foreign_key="assessorinfo.id",
# nullable=False
# )
# author: Optional["AssessorInfo"] = Relationship(back_populates="documents")
# created_at: datetime
# document_type: ReportType
class Documents(BaseModel, table=True):
assessor_id: uuid.UUID = Field(
foreign_key="assessorinfo.id",
nullable=False
)
author: Optional["AssessorInfo"] = Relationship(back_populates="documents")
created_at: datetime
document_type: ReportType
# building_id: uuid.UUID = Field(foreign_key="buildings.id", nullable=False)
# building: Optional["Buildings"] = Relationship(back_populates="documents")
building_id: uuid.UUID = Field(foreign_key="buildings.id", nullable=False)
building: Optional["Buildings"] = Relationship(back_populates="documents")
# target_table: str
# target_id: uuid.UUID
target_table: str
target_id: uuid.UUID
# class ReportType(str, Enum):
# QUIDOS_PRESITE_NOTE = "QUIDOS_PRESITE_NOTE"
# CHARTED_SURVEYOR_REPORT = "CHARTED_SURVEYOR_REPORT"
# ENERGY_PERFORMANCE_REPORT = "ENERGY_PERFORMANCE_REPORT"
# U_VALUE_CALCULATOR_REPORT = "U_VALUE_CALCULATOR_REPORT"
# OVERWRITING_U_VALUE_DECLARATION_FORM = "OVERWRITING_U_VALUE_DECLARATION_FORM"
# OSMOSIS_CONDITION_PAS_2035_REPORT = "OSMOSIS_CONDITION_PAS_2035_REPORT"
# DOMNA_CONDITION_PAS_2035_REPORT = "DOMNA_CONDITION_PAS_2035_REPORT"
class ReportType(str, Enum):
QUIDOS_PRESITE_NOTE = "QUIDOS_PRESITE_NOTE"
CHARTED_SURVEYOR_REPORT = "CHARTED_SURVEYOR_REPORT"
ENERGY_PERFORMANCE_REPORT = "ENERGY_PERFORMANCE_REPORT"
U_VALUE_CALCULATOR_REPORT = "U_VALUE_CALCULATOR_REPORT"
OVERWRITING_U_VALUE_DECLARATION_FORM = "OVERWRITING_U_VALUE_DECLARATION_FORM"
OSMOSIS_CONDITION_PAS_2035_REPORT = "OSMOSIS_CONDITION_PAS_2035_REPORT"
DOMNA_CONDITION_PAS_2035_REPORT = "DOMNA_CONDITION_PAS_2035_REPORT"
class uploaded_files(BaseModel, table=True):
__tablename__ = "uploaded_files"

View file

@ -153,7 +153,7 @@ filtered_dfs.append(post_epc)
# POST EPR
post_epr = df[
df["post-epc status"].str.lower().isin(["post epr completed"])
df["post-epc status"].str.lower().isin(["post epr completed", "Post epr desktop based completed"])
].copy()
post_epr["job_type"] = "POST epr"
filtered_dfs.append(post_epr)

View file

@ -200,7 +200,7 @@ timestamp = datetime.datetime.now().strftime('%Y-%m-%d_%H-%M')
# Upload to sharepoint
attribute = ['address', 'client', 'job_type', 'rate']
master_folder_name = "Stonewater SHDF 2.0 - Operations"
master_folder_name = "Stonewater SHDF 3.0 - Operations"
file_name = f"{master_folder_name}_{timestamp}.xlsx"
combined_with_rates[attribute].to_excel(file_name, index=False)

View file

@ -212,6 +212,10 @@ master_folder_name = "NCHA SHDF Wave 3 On Hold"
file_name = f"{master_folder_name}_{timestamp}.xlsx"
combined_with_rates[attribute].to_excel(file_name, index=False)
print("hello world")
file_name = "WCHG SHDF 2.1 Mansard_2025-08-29_10-46.xlsx"
master_folder_name="junte was here"
file_path = os.path.abspath(file_name)
upload_to_month_end_folder(file_name, file_path, master_folder_name)

View file

@ -96,7 +96,7 @@ def get_df(df, column_name, success_critera, job_name=None):
# RA
ra = get_df(df, "ra", ["completed rdsap 9.9", "completed rdsap 10"], "RA")
ra = get_df(df, "ra", ["completed rdsap 9.9", "complete rdsap 10"], "RA")
filtered_dfs.append(ra)

View file

@ -24,11 +24,11 @@ class BoardID(Enum):
board_ids = [
"9349630181", # WCHG Walkups-Operations x
"8830772914", # "L&Q London" x
"9601691730", # Cardo Wales & West - Wave 3 x
"9660895490", # Northumberland County SHDF Wave 3 x
"9641491000", # Watford Warm Homes x
"9349630181", # WCHG Walkups-Operations
"8830772914", # "L&Q London"
"9601691730", # Cardo Wales & West - Wave 3
"9660895490", # Northumberland County SHDF Wave 3
"9641491000", # Watford Warm Homes
"9671463094", # Seddon
]
@ -244,11 +244,6 @@ for board, all_records in board_to_record.items():
if not v3.empty:
filtered_dfs.append(v3)
# Coordination stage 2 Please complete
cors2 = get_df(df, "rc stage 2", ["to invoice"], "Coordination Stage 2")
if not cors2.empty:
filtered_dfs.append(cors2)
# Design archetype complex
design = get_df(df, "design invoicing status", ["to invoice"])
design1 = get_df(design, "design invoice type", ["archetype (complex)"], "Design Archetype Complex")
@ -287,24 +282,25 @@ for board, all_records in board_to_record.items():
if not lodg2.empty:
filtered_dfs.append(lodg2)
# POST EPC
# POST EPC + Retrofit Evaluation
post_epc = get_df(df, "post epc & eval. invoicing status", ["epc to invoice"], "POST EPC")
if not post_epc.empty:
filtered_dfs.append(post_epc)
# POST EPR
post_epr = get_df(df, "post epc & eval. invoicing status", ["epr to invoice"], "POST EPR")
if not post_epr.empty:
filtered_dfs.append(post_epr)
# Retrofit Evaluation might be need for just EPR
# post att
post_att = get_df(df, "post att invoicing status", ["to invoice"], "POST ATT")
if not post_att.empty:
filtered_dfs.append(post_epc)
# Retrofit Evaluation
rc = get_df(df, "rc stage 2 invoicing status", ["to invoice"], "retrofit evaluation")
# Coordination Stage 2
rc = get_df(df, "rc stage 2 invoicing status", ["to invoice"], "Coordination Stage 2")
if not rc.empty:
filtered_dfs.append(rc)

Binary file not shown.

View file

@ -14,7 +14,7 @@ import time
osmosis = SharePointScraper(SharePointInstaller.OSMOSIS_WAVE_3)
parent_folder = "/Osmosis-ACD Projects/Cardo/Cardo (Wales & West)/2506 Cardo Property Folders"
parent_folder = "/Osmosis-ACD Projects/NCHA WHSHF Wave 3/Property Folders"
asset_list = pd.read_excel("osmosis_data/asset_list.xlsx", sheet_name="Sheet 1")
@ -66,23 +66,23 @@ for index, address in tqdm(asset_list.iterrows()):
new_asset_list.append(asset_data)
# Run this is you just want to get url
def just_url(asset_list):
new_asset_list = []
for index, address in tqdm(asset_list.iterrows()):
folder_name = address['Name'] + " " + address['Postcode']
webUrl = osmosis.create_dir(folder_name, parent_folder)
asset_data = {
"Name": address['Name'],
"Postcode": address['Postcode'],
"Sharepoint": webUrl,
}
print(asset_data)
# # Run this is you just want to get url
# def just_url(asset_list):
# new_asset_list = []
# for index, address in tqdm(asset_list.iterrows()):
# folder_name = address['Name'] + " " + address['Postcode']
# webUrl = osmosis.create_dir(folder_name, parent_folder)
# asset_data = {
# "Name": address['Name'],
# "Postcode": address['Postcode'],
# "Sharepoint": webUrl,
# }
# print(asset_data)
new_asset_list.append(asset_data)
return new_asset_list
# new_asset_list.append(asset_data)
# return new_asset_list
new_asset_list = just_url(asset_list=asset_list)
# new_asset_list = just_url(asset_list=asset_list)
df = pd.DataFrame(new_asset_list)
df.to_csv("output.csv", index=False)

View file

@ -28,6 +28,7 @@ class SharePointInstaller(Enum):
OSMOSIS_WAVE_3 = os.getenv("OSMOSIS_SHAREPOINT_ID", "350a3b48-8311-4506-8abb-69bafc280d6f")
OSMOSIS_WAVE_2 = os.getenv("OSMOSIS_SHAREPOINT_ID", "bc925a9a-ad0b-4de9-9a3c-e61014cc7489")
WARMFRONT = os.getenv("WARMFRONT_SHARPOINT_ID", "bea71c30-d366-454c-a484-ae4d6fd95bc4")
NEW_JJC = os.getenv("NEW JJC", "10d96eba-b4f9-4e30-804f-05a8b60507b0")
class SharePointScraper():
"""

View file

@ -0,0 +1,85 @@
import os
os.environ["SHAREPOINT_CLIENT_ID"] = "6832a4c5-fb8c-4082-a746-4f51e1020f0d"
os.environ["SHAREPOINT_CLIENT_SECRET"] = "xpC8Q~Frww48SM1V-D8lGy5iOY7P_cJ7FF3jgarQ"
os.environ["SHAREPOINT_TENANT_ID"] = "10d5af8b-2cfd-4882-9ccd-b96e4812dacf"
from etl.scraper.scraper import SharePointScraper, SharePointInstaller
from etl.fileReader.pdfReaderToText import pdfReaderToText
import pandas as pd
from MonthEndUploader import upload_to_nick_folder
sharepoint = SharePointScraper(SharePointInstaller.NEW_JJC)
file_paths = sharepoint.download_file_for_each_address()
master_path = "Baxter Kelly/Calico/CALICO-001"
address_files = sharepoint.get_folders_in_path(master_path)
def extract_rating(text: str) -> str:
# Remove label if present
text = text.strip()
if text.lower().startswith("potential sap rating:"):
text = text.split(":", 1)[1].strip()
# Remove spaces and make uppercase
return text.replace(" ", "").upper()
def files_to_download(files_to_download_sharepoint_info):
file_names_to_download = {}
only_pdf = [".pdf"]
for file in files_to_download_sharepoint_info['value']:
if 'file' in file:
if any(file["name"].endswith(ext) for ext in only_pdf):
file_names_to_download.update({file["name"]: file['@microsoft.graph.downloadUrl']})
return file_names_to_download
def download_to_local(download, scraper, address):
each_file = []
for file_name, url in download.items():
content = scraper.get_file_content(url)
path = scraper.create_temp_file(content, f"{address}/{file_name}")
each_file.append(path)
return {address: each_file}
all_address_to_work_on = []
if "value" in address_files:
for address in address_files["value"]:
name_of_address = address["name"]
to_download = sharepoint.get_folders_in_path(f"{master_path}/{name_of_address}")
download = files_to_download(to_download)
address_to_files = download_to_local(download,sharepoint, name_of_address)
all_address_to_work_on.append(address_to_files)
final_data = []
for eachaddress in all_address_to_work_on:
for addressName, files in eachaddress.items():
for file in files:
pdf = pdfReaderToText(file)
if "Summary Information".lower() == pdf.text_list[0].lower():
current_sap_rating = pdf.text_list[pdf.text_list.index("Current SAP rating:") + 1]
house_no = pdf.text_list[pdf.text_list.index("House No:") + 1]
street = pdf.text_list[pdf.text_list.index("Street:") + 1]
post_code = pdf.text_list[pdf.text_list.index("Postcode:") + 1]
address = f"{house_no} {street.title()}"
floor_area = pdf.text_list[pdf.text_list.index("Lowest Floor:") + 1]
fuel_bill = pdf.text_list[pdf.text_list.index("Fuel Bill:") + 1]
row = {
"address": address,
"postcode": post_code,
"current sap rating": current_sap_rating,
"floor_area ": floor_area,
"fuel_bill ": fuel_bill,
}
final_data.append(row)
df = pd.DataFrame(final_data)
file_name = "installer.xlsx"
df.to_excel(file_name, index=False)
# Get local path
file_path = os.path.abspath(file_name)
upload_to_nick_folder(file_name, file_path)
print(f"File saved at: {file_path}")