junte/survey-extraction
1
0
Fork 0
mirror of https://github.com/Hestia-Homes/survey-extraction.git synced 2026-07-12 13:29:08 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions

Merge pull request #65 from Hestia-Homes/feature/hubspot_to_verification

Browse source 
Feature/hubspot to verification
This commit is contained in:
Jun-te Kim 2025-07-14 11:14:37 +01:00 • committed by GitHub
commit 90da471f46
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
39 changed files with 2273 additions and 155 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

6
.github/workflows/hubspot_deal_notes.yml → .github/workflows/hubspot_surveyed_needs_sign_off.yml vendored
View file

@ -1,7 +1,7 @@
name: Deal Notes From HubSpot Scraper
name: Daily Surved
on:
schedule:
- cron: '0 19 * * 0'
- cron: '0 17 * * 1-5'
workflow_dispatch:
jobs:
@ -24,6 +24,6 @@ jobs:
run: |
pwd
ls -la
poetry run python etl/dimitra_hubspot_notes_gather.py
poetry run python etl/hubspot_surveyed_needs_sign_off.py
env:
PYTHONPATH: ${{ github.workspace }}

1
.vscode/settings.json vendored
View file

@ -3,6 +3,7 @@
"python.REPL.sendToNativeREPL": true,
"notebook.output.scrolling": true,
"terminal.integrated.defaultProfile.linux": "bash",
"editor.rulers": [67],
"terminal.integrated.profiles.linux": {
"bash": {
"path": "/bin/bash"

5
alembic/env.py
View file

@ -5,8 +5,9 @@ from sqlalchemy import pool
from alembic import context
from sqlmodel import SQLModel
from etl.load.topLevel import *
from etl.load.preSiteNoteTypes import *
from etl.models.topLevel import *
from etl.models.preSiteNoteTypes import *
from etl.models.conditionReport import *
import os

19
alembic/things_alembic_doesn't_automate_for.txt Normal file
View file

@ -0,0 +1,19 @@
Source: https://alembic.sqlalchemy.org/en/latest/autogenerate.html
Autogenerate can not detect:
Changes of table name. These will come out as an add/drop of two different tables, and should be hand-edited into a name change instead.
Changes of column name. Like table name changes, these are detected as a column add/drop pair, which is not at all the same as a name change.
Anonymously named constraints. Give your constraints a name, e.g. UniqueConstraint('col1', 'col2', name="my_name"). See the section The Importance of Naming Constraints for background on how to configure automatic naming schemes for constraints.
Special SQLAlchemy types such as Enum when generated on a backend which doesn’t support ENUM directly - this because the representation of such a type in the non-supporting database, i.e. a CHAR+ CHECK constraint, could be any kind of CHAR+CHECK. For SQLAlchemy to determine that this is actually an ENUM would only be a guess, something that’s generally a bad idea. To implement your own “guessing” function here, use the sqlalchemy.events.DDLEvents.column_reflect() event to detect when a CHAR (or whatever the target type is) is reflected, and change it to an ENUM (or whatever type is desired) if it is known that that’s the intent of the type. The sqlalchemy.events.DDLEvents.after_parent_attach() can be used within the autogenerate process to intercept and un-attach unwanted CHECK constraints.
Autogenerate can’t currently, but will eventually detect:
Some free-standing constraint additions and removals may not be supported, including PRIMARY KEY, EXCLUDE, CHECK; these are not necessarily implemented within the autogenerate detection system and also may not be supported by the supporting SQLAlchemy dialect.
Sequence additions, removals - not yet implemented.

166
alembic/versions/42a19efef660_condition_table_from_osmosis_added_as_i_.py Normal file
View file

@ -0,0 +1,166 @@
"""condition table from osmosis added as i didn't add
Revision ID: 42a19efef660
Revises: 956c4ea39b29
Create Date: 2025-06-17 12:45:14.976944
"""
from typing import Sequence, Union
from alembic import op
import sqlalchemy as sa
import sqlmodel
# revision identifiers, used by Alembic.
revision: str = '42a19efef660'
down_revision: Union[str, None] = '956c4ea39b29'
branch_labels: Union[str, Sequence[str], None] = None
depends_on: Union[str, Sequence[str], None] = None
def upgrade() -> None:
"""Upgrade schema."""
# ### commands auto generated by Alembic - please adjust! ###
op.create_table('assessordetails',
sa.Column('id', sa.Uuid(), nullable=False),
sa.Column('assessor_name_and_id', sqlmodel.sql.sqltypes.AutoString(), nullable=False),
sa.Column('elmhurst_id', sqlmodel.sql.sqltypes.AutoString(), nullable=False),
sa.PrimaryKeyConstraint('id')
)
op.create_table('conservatoryoroutbuilding',
sa.Column('id', sa.Uuid(), nullable=False),
sa.Column('is_there_a_conservatory', sa.Boolean(), nullable=False),
sa.Column('is_there_a_cellar_present', sa.Boolean(), nullable=False),
sa.Column('is_there_an_outbuilding', sa.Boolean(), nullable=False),
sa.PrimaryKeyConstraint('id')
)
op.create_table('elevation',
sa.Column('id', sa.Uuid(), nullable=False),
sa.Column('protected_conservatory_or_aonb', sa.Boolean(), nullable=False),
sa.Column('material_type', sqlmodel.sql.sqltypes.AutoString(), nullable=False),
sa.Column('are_there_any_visible_signs_of_existing_wall_insulation', sqlmodel.sql.sqltypes.AutoString(), nullable=False),
sa.Column('does_the_existing_ground_level_on_any_elevation_bridge_the_dpc', sa.Boolean(), nullable=False),
sa.PrimaryKeyConstraint('id')
)
op.create_table('elevationinfo',
sa.Column('id', sa.Uuid(), nullable=False),
sa.Column('elevation_type', sqlmodel.sql.sqltypes.AutoString(), nullable=False),
sa.Column('cavity_wall_depth', sqlmodel.sql.sqltypes.AutoString(), nullable=False),
sa.Column('is_insulation_present', sa.Boolean(), nullable=False),
sa.Column('insulation_type', sqlmodel.sql.sqltypes.AutoString(), nullable=False),
sa.PrimaryKeyConstraint('id')
)
op.create_table('externalelevation',
sa.Column('id', sa.Uuid(), nullable=False),
sa.Column('structural_defects_of_elevation', sqlmodel.sql.sqltypes.AutoString(), nullable=False),
sa.Column('does_any_structural_defect_need_resolving_before_retrofit', sa.Boolean(), nullable=False),
sa.Column('are_there_any_signs_of_water_penetration_caused_by_failed_rainwater_goods_or_pipework', sa.Boolean(), nullable=False),
sa.Column('are_there_any_visible_signs_of_movement', sa.Boolean(), nullable=False),
sa.Column('are_there_any_visible_signs_of_cracking_to_the_existing_external_finish', sa.Boolean(), nullable=False),
sa.PrimaryKeyConstraint('id')
)
op.create_table('generalconditionheatingsystem',
sa.Column('id', sa.Uuid(), nullable=False),
sa.Column('is_the_heating_system_in_working_order', sa.Boolean(), nullable=False),
sa.Column('does_the_occupant_have_a_smart_meter', sa.Boolean(), nullable=False),
sa.Column('are_there_any_smart_monitoring_devices', sa.Boolean(), nullable=False),
sa.Column('are_the_gas_and_electricity_meters_accessible', sa.Boolean(), nullable=False),
sa.Column('dual_or_single_electric_meter', sqlmodel.sql.sqltypes.AutoString(), nullable=False),
sa.PrimaryKeyConstraint('id')
)
op.create_table('heating_from_condition_report',
sa.Column('id', sa.Uuid(), nullable=False),
sa.Column('room_stat_in_temperature_in_celsius', sqlmodel.sql.sqltypes.AutoString(), nullable=True),
sa.Column('room_stat_location', sqlmodel.sql.sqltypes.AutoString(), nullable=True),
sa.Column('is_the_heating_pattern_known', sqlmodel.sql.sqltypes.AutoString(), nullable=True),
sa.PrimaryKeyConstraint('id')
)
op.create_table('inspectionandproject',
sa.Column('id', sa.Uuid(), nullable=False),
sa.Column('inspection_date', sqlmodel.sql.sqltypes.AutoString(), nullable=False),
sa.PrimaryKeyConstraint('id')
)
op.create_table('mainheatingone',
sa.Column('id', sa.Uuid(), nullable=False),
sa.Column('as_defined_by', sqlmodel.sql.sqltypes.AutoString(), nullable=False),
sa.Column('fuel', sqlmodel.sql.sqltypes.AutoString(), nullable=False),
sa.Column('type', sqlmodel.sql.sqltypes.AutoString(), nullable=False),
sa.PrimaryKeyConstraint('id')
)
op.create_table('mainheatingtwo',
sa.Column('id', sa.Uuid(), nullable=False),
sa.Column('is_there_a_main_heating_two', sa.Boolean(), nullable=False),
sa.PrimaryKeyConstraint('id')
)
op.create_table('propertyaccess',
sa.Column('id', sa.Uuid(), nullable=False),
sa.Column('are_there_any_road_restriction_in_the_locality', sa.Boolean(), nullable=False),
sa.Column('is_on_street_parking_available', sa.Boolean(), nullable=False),
sa.Column('are_there_any_overhead_wires_or_cables', sa.Boolean(), nullable=False),
sa.Column('is_the_access_gated', sa.Boolean(), nullable=False),
sa.Column('is_there_restricted_space_for_contractors_to_access_the_wall_area', sa.Boolean(), nullable=False),
sa.Column('is_there_restricted_space_for_contractors_to_access_the_roof_area', sa.Boolean(), nullable=False),
sa.Column('is_there_more_than_1_point_5_meters_in_width_to_fence_or_neighbouring_boundary_along_the_full_gable_elevation', sa.Boolean(), nullable=False),
sa.Column('is_access_to_the_rear_provided_by_use_of_a_ginnel', sa.Boolean(), nullable=False),
sa.Column('is_access_to_the_rear_provided_by_use_of_a_secured_alleyway', sa.Boolean(), nullable=False),
sa.PrimaryKeyConstraint('id')
)
op.create_table('secondaryheating',
sa.Column('id', sa.Uuid(), nullable=False),
sa.Column('is_there_a_secondary_heating', sa.Boolean(), nullable=False),
sa.Column('fuel', sqlmodel.sql.sqltypes.AutoString(), nullable=False),
sa.Column('electric_heating_type', sqlmodel.sql.sqltypes.AutoString(), nullable=False),
sa.Column('gas_heating_type', sqlmodel.sql.sqltypes.AutoString(), nullable=False),
sa.PrimaryKeyConstraint('id')
)
op.create_table('theproperty',
sa.Column('id', sa.Uuid(), nullable=False),
sa.Column('house_type', sqlmodel.sql.sqltypes.AutoString(), nullable=False),
sa.Column('on_which_floor_is_the_flat_located', sqlmodel.sql.sqltypes.AutoString(), nullable=False),
sa.Column('is_there_a_corridor', sa.Boolean(), nullable=False),
sa.Column('is_it_heated', sa.Boolean(), nullable=False),
sa.Column('it_there_a_balcony', sa.Boolean(), nullable=False),
sa.Column('classification_type', sqlmodel.sql.sqltypes.AutoString(), nullable=False),
sa.Column('orientation_front_elevation', sqlmodel.sql.sqltypes.AutoString(), nullable=False),
sa.Column('orientation_in_degrees_front_elevation', sqlmodel.sql.sqltypes.AutoString(), nullable=False),
sa.Column('exposure_zone', sqlmodel.sql.sqltypes.AutoString(), nullable=False),
sa.Column('main_wall_construction', sqlmodel.sql.sqltypes.AutoString(), nullable=False),
sa.PrimaryKeyConstraint('id')
)
op.create_table('conditionreport',
sa.Column('id', sa.Uuid(), nullable=False),
sa.Column('project_site_name', sqlmodel.sql.sqltypes.AutoString(), nullable=False),
sa.Column('property_reference_code', sqlmodel.sql.sqltypes.AutoString(), nullable=False),
sa.Column('property_address', sqlmodel.sql.sqltypes.AutoString(), nullable=False),
sa.Column('postcode', sqlmodel.sql.sqltypes.AutoString(), nullable=False),
sa.Column('assessor_details_id', sa.Uuid(), nullable=True),
sa.Column('inspection_id', sa.Uuid(), nullable=True),
sa.Column('property_id', sa.Uuid(), nullable=True),
sa.Column('general_condition_id', sa.Uuid(), nullable=True),
sa.ForeignKeyConstraint(['assessor_details_id'], ['assessordetails.id'], ),
sa.ForeignKeyConstraint(['general_condition_id'], ['generalconditionheatingsystem.id'], ),
sa.ForeignKeyConstraint(['inspection_id'], ['inspectionandproject.id'], ),
sa.ForeignKeyConstraint(['property_id'], ['theproperty.id'], ),
sa.PrimaryKeyConstraint('id')
)
# ### end Alembic commands ###
def downgrade() -> None:
"""Downgrade schema."""
# ### commands auto generated by Alembic - please adjust! ###
op.drop_table('conditionreport')
op.drop_table('theproperty')
op.drop_table('secondaryheating')
op.drop_table('propertyaccess')
op.drop_table('mainheatingtwo')
op.drop_table('mainheatingone')
op.drop_table('inspectionandproject')
op.drop_table('heating_from_condition_report')
op.drop_table('generalconditionheatingsystem')
op.drop_table('externalelevation')
op.drop_table('elevationinfo')
op.drop_table('elevation')
op.drop_table('conservatoryoroutbuilding')
op.drop_table('assessordetails')
# ### end Alembic commands ###

132
alembic/versions/544f060f1eb8_add_elevation.py Normal file
View file

@ -0,0 +1,132 @@
"""add elevation
Revision ID: 544f060f1eb8
Revises: 881142a89edb
Create Date: 2025-06-19 09:54:43.770068
"""
from typing import Sequence, Union
from alembic import op
import sqlalchemy as sa
# revision identifiers, used by Alembic.
revision: str = '544f060f1eb8'
down_revision: Union[str, None] = '881142a89edb'
branch_labels: Union[str, Sequence[str], None] = None
depends_on: Union[str, Sequence[str], None] = None
def upgrade() -> None:
"""Upgrade schema."""
# ### commands auto generated by Alembic - please adjust! ###
op.drop_table('secondaryheating')
op.drop_table('heating_from_condition_report')
op.drop_table('generalconditionheatingsystem')
op.drop_table('conditionreport')
op.drop_table('mainheatingtwo')
op.drop_table('conservatoryoroutbuilding')
op.drop_table('mainheatingone')
op.drop_table('elevation')
op.drop_table('generalinformation')
# ### end Alembic commands ###
def downgrade() -> None:
"""Downgrade schema."""
# ### commands auto generated by Alembic - please adjust! ###
op.create_table('elevation',
sa.Column('id', sa.UUID(), autoincrement=False, nullable=False),
sa.Column('protected_conservatory_or_aonb', sa.BOOLEAN(), autoincrement=False, nullable=False),
sa.Column('material_type', sa.VARCHAR(), autoincrement=False, nullable=False),
sa.Column('are_there_any_visible_signs_of_existing_wall_insulation', sa.VARCHAR(), autoincrement=False, nullable=False),
sa.Column('does_the_existing_ground_level_on_any_elevation_bridge_the_dpc', sa.BOOLEAN(), autoincrement=False, nullable=False),
sa.PrimaryKeyConstraint('id', name='elevation_pkey')
)
op.create_table('mainheatingone',
sa.Column('id', sa.UUID(), autoincrement=False, nullable=False),
sa.Column('as_defined_by', sa.VARCHAR(), autoincrement=False, nullable=False),
sa.Column('fuel', sa.VARCHAR(), autoincrement=False, nullable=False),
sa.Column('type', sa.VARCHAR(), autoincrement=False, nullable=False),
sa.PrimaryKeyConstraint('id', name='mainheatingone_pkey')
)
op.create_table('conservatoryoroutbuilding',
sa.Column('id', sa.UUID(), autoincrement=False, nullable=False),
sa.Column('is_there_a_conservatory', sa.BOOLEAN(), autoincrement=False, nullable=False),
sa.Column('is_there_a_cellar_present', sa.BOOLEAN(), autoincrement=False, nullable=False),
sa.Column('is_there_an_outbuilding', sa.BOOLEAN(), autoincrement=False, nullable=False),
sa.PrimaryKeyConstraint('id', name='conservatoryoroutbuilding_pkey')
)
op.create_table('mainheatingtwo',
sa.Column('id', sa.UUID(), autoincrement=False, nullable=False),
sa.Column('is_there_a_main_heating_two', sa.BOOLEAN(), autoincrement=False, nullable=False),
sa.PrimaryKeyConstraint('id', name='mainheatingtwo_pkey')
)
op.create_table('conditionreport',
sa.Column('id', sa.UUID(), autoincrement=False, nullable=False),
sa.Column('project_site_name', sa.VARCHAR(), autoincrement=False, nullable=False),
sa.Column('property_reference_code', sa.VARCHAR(), autoincrement=False, nullable=False),
sa.Column('property_address', sa.VARCHAR(), autoincrement=False, nullable=False),
sa.Column('postcode', sa.VARCHAR(), autoincrement=False, nullable=False),
sa.Column('general_information_id', sa.UUID(), autoincrement=False, nullable=True),
sa.ForeignKeyConstraint(['general_information_id'], ['generalinformation.id'], name='conditionreport_general_information_id_fkey'),
sa.PrimaryKeyConstraint('id', name='conditionreport_pkey')
)
op.create_table('generalconditionheatingsystem',
sa.Column('id', sa.UUID(), autoincrement=False, nullable=False),
sa.Column('is_the_heating_system_in_working_order', sa.BOOLEAN(), autoincrement=False, nullable=False),
sa.Column('does_the_occupant_have_a_smart_meter', sa.BOOLEAN(), autoincrement=False, nullable=False),
sa.Column('are_there_any_smart_monitoring_devices', sa.BOOLEAN(), autoincrement=False, nullable=False),
sa.Column('are_the_gas_and_electricity_meters_accessible', sa.BOOLEAN(), autoincrement=False, nullable=False),
sa.Column('dual_or_single_electric_meter', sa.VARCHAR(), autoincrement=False, nullable=False),
sa.PrimaryKeyConstraint('id', name='generalconditionheatingsystem_pkey')
)
op.create_table('heating_from_condition_report',
sa.Column('id', sa.UUID(), autoincrement=False, nullable=False),
sa.Column('room_stat_in_temperature_in_celsius', sa.VARCHAR(), autoincrement=False, nullable=True),
sa.Column('room_stat_location', sa.VARCHAR(), autoincrement=False, nullable=True),
sa.Column('is_the_heating_pattern_known', sa.VARCHAR(), autoincrement=False, nullable=True),
sa.PrimaryKeyConstraint('id', name='heating_from_condition_report_pkey')
)
op.create_table('externalelevation',
sa.Column('id', sa.UUID(), autoincrement=False, nullable=False),
sa.Column('structural_defects_of_elevation', sa.VARCHAR(), autoincrement=False, nullable=False),
sa.Column('does_any_structural_defect_need_resolving_before_retrofit', sa.BOOLEAN(), autoincrement=False, nullable=False),
sa.Column('are_there_any_signs_of_water_penetration_caused_by_failed_rainw', sa.BOOLEAN(), autoincrement=False, nullable=False),
sa.Column('are_there_any_visible_signs_of_movement', sa.BOOLEAN(), autoincrement=False, nullable=False),
sa.Column('are_there_any_visible_signs_of_cracking_to_the_existing_externa', sa.BOOLEAN(), autoincrement=False, nullable=False),
sa.PrimaryKeyConstraint('id', name='externalelevation_pkey')
)
op.create_table('generalinformation',
sa.Column('id', sa.UUID(), autoincrement=False, nullable=False),
sa.Column('assessor_details_id', sa.UUID(), autoincrement=False, nullable=False),
sa.Column('inspection_and_project_id', sa.UUID(), autoincrement=False, nullable=False),
sa.Column('the_property_id', sa.UUID(), autoincrement=False, nullable=False),
sa.ForeignKeyConstraint(['assessor_details_id'], ['assessordetails.id'], name='generalinformation_assessor_details_id_fkey'),
sa.ForeignKeyConstraint(['inspection_and_project_id'], ['inspectionandproject.id'], name='generalinformation_inspection_and_project_id_fkey'),
sa.ForeignKeyConstraint(['the_property_id'], ['theproperty.id'], name='generalinformation_the_property_id_fkey'),
sa.PrimaryKeyConstraint('id', name='generalinformation_pkey')
)
op.create_table('secondaryheating',
sa.Column('id', sa.UUID(), autoincrement=False, nullable=False),
sa.Column('is_there_a_secondary_heating', sa.BOOLEAN(), autoincrement=False, nullable=False),
sa.Column('fuel', sa.VARCHAR(), autoincrement=False, nullable=False),
sa.Column('electric_heating_type', sa.VARCHAR(), autoincrement=False, nullable=False),
sa.Column('gas_heating_type', sa.VARCHAR(), autoincrement=False, nullable=False),
sa.PrimaryKeyConstraint('id', name='secondaryheating_pkey')
)
op.create_table('propertyaccess',
sa.Column('id', sa.UUID(), autoincrement=False, nullable=False),
sa.Column('are_there_any_road_restriction_in_the_locality', sa.BOOLEAN(), autoincrement=False, nullable=False),
sa.Column('is_on_street_parking_available', sa.BOOLEAN(), autoincrement=False, nullable=False),
sa.Column('are_there_any_overhead_wires_or_cables', sa.BOOLEAN(), autoincrement=False, nullable=False),
sa.Column('is_the_access_gated', sa.BOOLEAN(), autoincrement=False, nullable=False),
sa.Column('is_there_restricted_space_for_contractors_to_access_the_wall_ar', sa.BOOLEAN(), autoincrement=False, nullable=False),
sa.Column('is_there_restricted_space_for_contractors_to_access_the_roof_ar', sa.BOOLEAN(), autoincrement=False, nullable=False),
sa.Column('is_there_more_than_1_point_5_meters_in_width_to_fence_or_neighb', sa.BOOLEAN(), autoincrement=False, nullable=False),
sa.Column('is_access_to_the_rear_provided_by_use_of_a_ginnel', sa.BOOLEAN(), autoincrement=False, nullable=False),
sa.Column('is_access_to_the_rear_provided_by_use_of_a_secured_alleyway', sa.BOOLEAN(), autoincrement=False, nullable=False),
sa.PrimaryKeyConstraint('id', name='propertyaccess_pkey')
)
# ### end Alembic commands ###

73
alembic/versions/771a2bc48ac2_add_general_information_class.py Normal file
View file

@ -0,0 +1,73 @@
"""add general information class
Revision ID: 771a2bc48ac2
Revises: f2f205448dff
Create Date: 2025-06-18 14:56:08.762223
"""
from typing import Sequence, Union
from alembic import op
import sqlalchemy as sa
# revision identifiers, used by Alembic.
revision: str = '771a2bc48ac2'
down_revision: Union[str, None] = 'f2f205448dff'
branch_labels: Union[str, Sequence[str], None] = None
depends_on: Union[str, Sequence[str], None] = None
def upgrade() -> None:
"""Upgrade schema."""
# ### commands auto generated by Alembic - please adjust! ###
op.create_table('generalinformation',
sa.Column('id', sa.Uuid(), nullable=False),
sa.Column('assessor_details_id', sa.Uuid(), nullable=False),
sa.Column('inspection_and_project_id', sa.Uuid(), nullable=False),
sa.Column('the_property_id', sa.Uuid(), nullable=False),
sa.ForeignKeyConstraint(['assessor_details_id'], ['assessordetails.id'], ),
sa.ForeignKeyConstraint(['inspection_and_project_id'], ['inspectionandproject.id'], ),
sa.ForeignKeyConstraint(['the_property_id'], ['theproperty.id'], ),
sa.PrimaryKeyConstraint('id')
)
op.add_column('conditionreport', sa.Column('general_information_id', sa.Uuid(), nullable=True))
op.drop_constraint('conditionreport_general_condition_id_fkey', 'conditionreport', type_='foreignkey')
op.drop_constraint('conditionreport_assessor_details_id_fkey', 'conditionreport', type_='foreignkey')
op.create_foreign_key(None, 'conditionreport', 'generalinformation', ['general_information_id'], ['id'])
op.drop_column('conditionreport', 'general_condition_id')
op.drop_column('conditionreport', 'assessor_details_id')
op.add_column('externalelevation', sa.Column('are_there_any_signs_of_water_penetration_caused_by_failed_rainwater_goods_or_pipework', sa.Boolean(), nullable=False))
op.add_column('externalelevation', sa.Column('are_there_any_visible_signs_of_cracking_to_the_existing_external_finish', sa.Boolean(), nullable=False))
op.drop_column('externalelevation', 'water_penetration_from_failed_rainwater_goods')
op.drop_column('externalelevation', 'cracking_to_external_finish_visible')
op.add_column('propertyaccess', sa.Column('is_there_restricted_space_for_contractors_to_access_the_wall_area', sa.Boolean(), nullable=False))
op.add_column('propertyaccess', sa.Column('is_there_restricted_space_for_contractors_to_access_the_roof_area', sa.Boolean(), nullable=False))
op.add_column('propertyaccess', sa.Column('is_there_more_than_1_point_5_meters_in_width_to_fence_or_neighbouring_boundary_along_the_full_gable_elevation', sa.Boolean(), nullable=False))
op.drop_column('propertyaccess', 'sufficient_boundary_width')
op.drop_column('propertyaccess', 'restricted_wall_access')
op.drop_column('propertyaccess', 'restricted_roof_access')
# ### end Alembic commands ###
def downgrade() -> None:
"""Downgrade schema."""
# ### commands auto generated by Alembic - please adjust! ###
op.add_column('propertyaccess', sa.Column('restricted_roof_access', sa.BOOLEAN(), autoincrement=False, nullable=False))
op.add_column('propertyaccess', sa.Column('restricted_wall_access', sa.BOOLEAN(), autoincrement=False, nullable=False))
op.add_column('propertyaccess', sa.Column('sufficient_boundary_width', sa.BOOLEAN(), autoincrement=False, nullable=False))
op.drop_column('propertyaccess', 'is_there_more_than_1_point_5_meters_in_width_to_fence_or_neighbouring_boundary_along_the_full_gable_elevation')
op.drop_column('propertyaccess', 'is_there_restricted_space_for_contractors_to_access_the_roof_area')
op.drop_column('propertyaccess', 'is_there_restricted_space_for_contractors_to_access_the_wall_area')
op.add_column('externalelevation', sa.Column('cracking_to_external_finish_visible', sa.BOOLEAN(), autoincrement=False, nullable=False))
op.add_column('externalelevation', sa.Column('water_penetration_from_failed_rainwater_goods', sa.BOOLEAN(), autoincrement=False, nullable=False))
op.drop_column('externalelevation', 'are_there_any_visible_signs_of_cracking_to_the_existing_external_finish')
op.drop_column('externalelevation', 'are_there_any_signs_of_water_penetration_caused_by_failed_rainwater_goods_or_pipework')
op.add_column('conditionreport', sa.Column('assessor_details_id', sa.UUID(), autoincrement=False, nullable=True))
op.add_column('conditionreport', sa.Column('general_condition_id', sa.UUID(), autoincrement=False, nullable=True))
op.drop_constraint(None, 'conditionreport', type_='foreignkey')
op.create_foreign_key('conditionreport_assessor_details_id_fkey', 'conditionreport', 'assessordetails', ['assessor_details_id'], ['id'])
op.create_foreign_key('conditionreport_general_condition_id_fkey', 'conditionreport', 'generalconditionheatingsystem', ['general_condition_id'], ['id'])
op.drop_column('conditionreport', 'general_information_id')
op.drop_table('generalinformation')
# ### end Alembic commands ###

176
alembic/versions/881142a89edb_delete_everythign_and_start_again.py Normal file
View file

@ -0,0 +1,176 @@
"""delete everythign and start again
Revision ID: 881142a89edb
Revises: 771a2bc48ac2
Create Date: 2025-06-18 15:38:53.955796
"""
from typing import Sequence, Union
from alembic import op
import sqlalchemy as sa
# revision identifiers, used by Alembic.
revision: str = '881142a89edb'
down_revision: Union[str, None] = '771a2bc48ac2'
branch_labels: Union[str, Sequence[str], None] = None
depends_on: Union[str, Sequence[str], None] = None
def upgrade() -> None:
"""Upgrade schema."""
# ### commands auto generated by Alembic - please adjust! ###
op.execute("DROP TABLE conditionreport CASCADE")
op.execute("DROP TABLE generalinformation CASCADE")
op.execute("DROP TABLE theproperty CASCADE")
op.execute("DROP TABLE assessordetails CASCADE")
op.execute("DROP TABLE inspectionandproject CASCADE")
op.execute("DROP TABLE propertyaccess CASCADE")
op.execute("DROP TABLE externalelevation CASCADE")
op.execute("DROP TABLE generalconditionheatingsystem CASCADE")
op.execute("DROP TABLE heating_from_condition_report CASCADE")
op.execute("DROP TABLE mainheatingone CASCADE")
op.execute("DROP TABLE mainheatingtwo CASCADE")
op.execute("DROP TABLE secondaryheating CASCADE")
op.execute("DROP TABLE conservatoryoroutbuilding CASCADE")
op.execute("DROP TABLE elevation CASCADE")
op.execute("DROP TABLE elevationinfo CASCADE")
def downgrade() -> None:
"""Downgrade schema."""
# ### commands auto generated by Alembic - please adjust! ###
op.create_table('inspectionandproject',
sa.Column('id', sa.UUID(), autoincrement=False, nullable=False),
sa.Column('inspection_date', sa.VARCHAR(), autoincrement=False, nullable=False),
sa.PrimaryKeyConstraint('id', name='inspectionandproject_pkey'),
postgresql_ignore_search_path=False
)
op.create_table('elevation',
sa.Column('id', sa.UUID(), autoincrement=False, nullable=False),
sa.Column('protected_conservatory_or_aonb', sa.BOOLEAN(), autoincrement=False, nullable=False),
sa.Column('material_type', sa.VARCHAR(), autoincrement=False, nullable=False),
sa.Column('are_there_any_visible_signs_of_existing_wall_insulation', sa.VARCHAR(), autoincrement=False, nullable=False),
sa.Column('does_the_existing_ground_level_on_any_elevation_bridge_the_dpc', sa.BOOLEAN(), autoincrement=False, nullable=False),
sa.PrimaryKeyConstraint('id', name='elevation_pkey')
)
op.create_table('mainheatingone',
sa.Column('id', sa.UUID(), autoincrement=False, nullable=False),
sa.Column('as_defined_by', sa.VARCHAR(), autoincrement=False, nullable=False),
sa.Column('fuel', sa.VARCHAR(), autoincrement=False, nullable=False),
sa.Column('type', sa.VARCHAR(), autoincrement=False, nullable=False),
sa.PrimaryKeyConstraint('id', name='mainheatingone_pkey')
)
op.create_table('assessordetails',
sa.Column('id', sa.UUID(), autoincrement=False, nullable=False),
sa.Column('assessor_name_and_id', sa.VARCHAR(), autoincrement=False, nullable=False),
sa.Column('elmhurst_id', sa.VARCHAR(), autoincrement=False, nullable=False),
sa.PrimaryKeyConstraint('id', name='assessordetails_pkey'),
postgresql_ignore_search_path=False
)
op.create_table('elevationinfo',
sa.Column('id', sa.UUID(), autoincrement=False, nullable=False),
sa.Column('elevation_type', sa.VARCHAR(), autoincrement=False, nullable=False),
sa.Column('cavity_wall_depth', sa.VARCHAR(), autoincrement=False, nullable=False),
sa.Column('is_insulation_present', sa.BOOLEAN(), autoincrement=False, nullable=False),
sa.Column('insulation_type', sa.VARCHAR(), autoincrement=False, nullable=False),
sa.PrimaryKeyConstraint('id', name='elevationinfo_pkey')
)
op.create_table('externalelevation',
sa.Column('id', sa.UUID(), autoincrement=False, nullable=False),
sa.Column('structural_defects_of_elevation', sa.VARCHAR(), autoincrement=False, nullable=False),
sa.Column('does_any_structural_defect_need_resolving_before_retrofit', sa.BOOLEAN(), autoincrement=False, nullable=False),
sa.Column('are_there_any_signs_of_water_penetration_caused_by_failed_rainw', sa.BOOLEAN(), autoincrement=False, nullable=False),
sa.Column('are_there_any_visible_signs_of_movement', sa.BOOLEAN(), autoincrement=False, nullable=False),
sa.Column('are_there_any_visible_signs_of_cracking_to_the_existing_externa', sa.BOOLEAN(), autoincrement=False, nullable=False),
sa.PrimaryKeyConstraint('id', name='externalelevation_pkey')
)
op.create_table('conditionreport',
sa.Column('id', sa.UUID(), autoincrement=False, nullable=False),
sa.Column('project_site_name', sa.VARCHAR(), autoincrement=False, nullable=False),
sa.Column('property_reference_code', sa.VARCHAR(), autoincrement=False, nullable=False),
sa.Column('property_address', sa.VARCHAR(), autoincrement=False, nullable=False),
sa.Column('postcode', sa.VARCHAR(), autoincrement=False, nullable=False),
sa.Column('inspection_id', sa.UUID(), autoincrement=False, nullable=True),
sa.Column('property_id', sa.UUID(), autoincrement=False, nullable=True),
sa.Column('general_information_id', sa.UUID(), autoincrement=False, nullable=True),
sa.ForeignKeyConstraint(['inspection_id'], ['inspectionandproject.id'], name='conditionreport_inspection_id_fkey'),
sa.ForeignKeyConstraint(['property_id'], ['theproperty.id'], name='conditionreport_property_id_fkey'),
sa.PrimaryKeyConstraint('id', name='conditionreport_pkey')
)
op.create_table('theproperty',
sa.Column('id', sa.UUID(), autoincrement=False, nullable=False),
sa.Column('house_type', sa.VARCHAR(), autoincrement=False, nullable=False),
sa.Column('on_which_floor_is_the_flat_located', sa.VARCHAR(), autoincrement=False, nullable=False),
sa.Column('is_there_a_corridor', sa.BOOLEAN(), autoincrement=False, nullable=False),
sa.Column('is_it_heated', sa.BOOLEAN(), autoincrement=False, nullable=False),
sa.Column('it_there_a_balcony', sa.BOOLEAN(), autoincrement=False, nullable=False),
sa.Column('classification_type', sa.VARCHAR(), autoincrement=False, nullable=False),
sa.Column('orientation_front_elevation', sa.VARCHAR(), autoincrement=False, nullable=False),
sa.Column('orientation_in_degrees_front_elevation', sa.VARCHAR(), autoincrement=False, nullable=False),
sa.Column('exposure_zone', sa.VARCHAR(), autoincrement=False, nullable=False),
sa.Column('main_wall_construction', sa.VARCHAR(), autoincrement=False, nullable=False),
sa.PrimaryKeyConstraint('id', name='theproperty_pkey'),
postgresql_ignore_search_path=False
)
op.create_table('generalconditionheatingsystem',
sa.Column('id', sa.UUID(), autoincrement=False, nullable=False),
sa.Column('is_the_heating_system_in_working_order', sa.BOOLEAN(), autoincrement=False, nullable=False),
sa.Column('does_the_occupant_have_a_smart_meter', sa.BOOLEAN(), autoincrement=False, nullable=False),
sa.Column('are_there_any_smart_monitoring_devices', sa.BOOLEAN(), autoincrement=False, nullable=False),
sa.Column('are_the_gas_and_electricity_meters_accessible', sa.BOOLEAN(), autoincrement=False, nullable=False),
sa.Column('dual_or_single_electric_meter', sa.VARCHAR(), autoincrement=False, nullable=False),
sa.PrimaryKeyConstraint('id', name='generalconditionheatingsystem_pkey')
)
op.create_table('mainheatingtwo',
sa.Column('id', sa.UUID(), autoincrement=False, nullable=False),
sa.Column('is_there_a_main_heating_two', sa.BOOLEAN(), autoincrement=False, nullable=False),
sa.PrimaryKeyConstraint('id', name='mainheatingtwo_pkey')
)
op.create_table('generalinformation',
sa.Column('id', sa.UUID(), autoincrement=False, nullable=False),
sa.Column('assessor_details_id', sa.UUID(), autoincrement=False, nullable=False),
sa.Column('inspection_and_project_id', sa.UUID(), autoincrement=False, nullable=False),
sa.Column('the_property_id', sa.UUID(), autoincrement=False, nullable=False),
sa.ForeignKeyConstraint(['assessor_details_id'], ['assessordetails.id'], name='generalinformation_assessor_details_id_fkey'),
sa.ForeignKeyConstraint(['inspection_and_project_id'], ['inspectionandproject.id'], name='generalinformation_inspection_and_project_id_fkey'),
sa.ForeignKeyConstraint(['the_property_id'], ['theproperty.id'], name='generalinformation_the_property_id_fkey'),
sa.PrimaryKeyConstraint('id', name='generalinformation_pkey')
)
op.create_table('heating_from_condition_report',
sa.Column('id', sa.UUID(), autoincrement=False, nullable=False),
sa.Column('room_stat_in_temperature_in_celsius', sa.VARCHAR(), autoincrement=False, nullable=True),
sa.Column('room_stat_location', sa.VARCHAR(), autoincrement=False, nullable=True),
sa.Column('is_the_heating_pattern_known', sa.VARCHAR(), autoincrement=False, nullable=True),
sa.PrimaryKeyConstraint('id', name='heating_from_condition_report_pkey')
)
op.create_table('propertyaccess',
sa.Column('id', sa.UUID(), autoincrement=False, nullable=False),
sa.Column('are_there_any_road_restriction_in_the_locality', sa.BOOLEAN(), autoincrement=False, nullable=False),
sa.Column('is_on_street_parking_available', sa.BOOLEAN(), autoincrement=False, nullable=False),
sa.Column('are_there_any_overhead_wires_or_cables', sa.BOOLEAN(), autoincrement=False, nullable=False),
sa.Column('is_the_access_gated', sa.BOOLEAN(), autoincrement=False, nullable=False),
sa.Column('is_there_restricted_space_for_contractors_to_access_the_wall_ar', sa.BOOLEAN(), autoincrement=False, nullable=False),
sa.Column('is_there_restricted_space_for_contractors_to_access_the_roof_ar', sa.BOOLEAN(), autoincrement=False, nullable=False),
sa.Column('is_there_more_than_1_point_5_meters_in_width_to_fence_or_neighb', sa.BOOLEAN(), autoincrement=False, nullable=False),
sa.Column('is_access_to_the_rear_provided_by_use_of_a_ginnel', sa.BOOLEAN(), autoincrement=False, nullable=False),
sa.Column('is_access_to_the_rear_provided_by_use_of_a_secured_alleyway', sa.BOOLEAN(), autoincrement=False, nullable=False),
sa.PrimaryKeyConstraint('id', name='propertyaccess_pkey')
)
op.create_table('secondaryheating',
sa.Column('id', sa.UUID(), autoincrement=False, nullable=False),
sa.Column('is_there_a_secondary_heating', sa.BOOLEAN(), autoincrement=False, nullable=False),
sa.Column('fuel', sa.VARCHAR(), autoincrement=False, nullable=False),
sa.Column('electric_heating_type', sa.VARCHAR(), autoincrement=False, nullable=False),
sa.Column('gas_heating_type', sa.VARCHAR(), autoincrement=False, nullable=False),
sa.PrimaryKeyConstraint('id', name='secondaryheating_pkey')
)
op.create_table('conservatoryoroutbuilding',
sa.Column('id', sa.UUID(), autoincrement=False, nullable=False),
sa.Column('is_there_a_conservatory', sa.BOOLEAN(), autoincrement=False, nullable=False),
sa.Column('is_there_a_cellar_present', sa.BOOLEAN(), autoincrement=False, nullable=False),
sa.Column('is_there_an_outbuilding', sa.BOOLEAN(), autoincrement=False, nullable=False),
sa.PrimaryKeyConstraint('id', name='conservatoryoroutbuilding_pkey')
)
# ### end Alembic commands ###

34
alembic/versions/956c4ea39b29_condition_table_from_osmosis.py Normal file
View file

@ -0,0 +1,34 @@
"""condition table from osmosis
Revision ID: 956c4ea39b29
Revises: 427e65da69c1
Create Date: 2025-06-17 10:49:55.213111
"""
from typing import Sequence, Union
from alembic import op
import sqlalchemy as sa
# revision identifiers, used by Alembic.
revision: str = '956c4ea39b29'
down_revision: Union[str, None] = '427e65da69c1'
branch_labels: Union[str, Sequence[str], None] = None
depends_on: Union[str, Sequence[str], None] = None
def upgrade() -> None:
"""Upgrade schema."""
# ### commands auto generated by Alembic - please adjust! ###
op.create_foreign_key(None, 'documents', 'assessorinfo', ['assessor_id'], ['id'])
op.create_foreign_key(None, 'presitenote', 'assessorinfo', ['assessor_id'], ['id'])
# ### end Alembic commands ###
def downgrade() -> None:
"""Downgrade schema."""
# ### commands auto generated by Alembic - please adjust! ###
op.drop_constraint(None, 'presitenote', type_='foreignkey')
op.drop_constraint(None, 'documents', type_='foreignkey')
# ### end Alembic commands ###

38
alembic/versions/c373a9a083f1_is_there_an_fourth_external_elevation.py Normal file
View file

@ -0,0 +1,38 @@
"""is there an fourth external elevation
Revision ID: c373a9a083f1
Revises: ceffde5b28ac
Create Date: 2025-06-20 15:08:31.752580
"""
from typing import Sequence, Union
from alembic import op
import sqlalchemy as sa
# revision identifiers, used by Alembic.
revision: str = 'c373a9a083f1'
down_revision: Union[str, None] = 'ceffde5b28ac'
branch_labels: Union[str, Sequence[str], None] = None
depends_on: Union[str, Sequence[str], None] = None
def upgrade() -> None:
"""Upgrade schema."""
# ### commands auto generated by Alembic - please adjust! ###
op.add_column('externalelevationgabletwo', sa.Column('is_there_a_fourth_external_elevation', sa.Boolean(), nullable=False))
op.drop_column('externalelevationgabletwo', 'do_all_answers_for_the_front_elevation_apply_to_this_wall')
# ### end Alembic commands ###
def downgrade() -> None:
"""Downgrade schema."""
# ### commands auto generated by Alembic - please adjust! ###
op.add_column('externalelevationgabletwo', sa.Column('do_all_answers_for_the_front_elevation_apply_to_this_wall', sa.BOOLEAN(), autoincrement=False, nullable=False))
op.drop_column('externalelevationgabletwo', 'is_there_a_fourth_external_elevation')
op.add_column('externalelevation', sa.Column('are_there_any_visible_signs_of_cracking_to_the_existing_externa', sa.BOOLEAN(), autoincrement=False, nullable=False))
op.add_column('externalelevation', sa.Column('any_signs_of_water_penetration_caused_by_failed_rainwater_goods', sa.BOOLEAN(), autoincrement=False, nullable=False))
op.drop_column('externalelevation', 'are_there_any_visible_signs_of_cracking_to_the_existing_external_finish')
op.drop_column('externalelevation', 'any_signs_of_water_penetration_caused_by_failed_rainwater_goods_or_pipework')
# ### end Alembic commands ###

43
alembic/versions/ceffde5b28ac_added_new_column.py Normal file
View file

@ -0,0 +1,43 @@
"""added new column
Revision ID: ceffde5b28ac
Revises: 544f060f1eb8
Create Date: 2025-06-19 10:23:37.567361
"""
from typing import Sequence, Union
from alembic import op
import sqlalchemy as sa
import sqlmodel
# revision identifiers, used by Alembic.
revision: str = 'ceffde5b28ac'
down_revision: Union[str, None] = '544f060f1eb8'
branch_labels: Union[str, Sequence[str], None] = None
depends_on: Union[str, Sequence[str], None] = None
def upgrade() -> None:
"""Upgrade schema."""
# ### commands auto generated by Alembic - please adjust! ###
op.create_table('elevation',
sa.Column('id', sa.Uuid(), nullable=False),
sa.Column('protected_conservatory_or_aonb', sa.Boolean(), nullable=False),
sa.Column('material_type', sqlmodel.sql.sqltypes.AutoString(), nullable=False),
sa.Column('visible_signs_of_existing_wall_insulation', sqlmodel.sql.sqltypes.AutoString(), nullable=False),
sa.Column('ground_level_bridge_the_dpc', sa.Boolean(), nullable=False),
sa.PrimaryKeyConstraint('id')
)
op.add_column('elevationinfo', sa.Column('elevation_id', sa.Uuid(), nullable=True))
op.create_foreign_key(None, 'elevationinfo', 'elevation', ['elevation_id'], ['id'])
# ### end Alembic commands ###
def downgrade() -> None:
"""Downgrade schema."""
# ### commands auto generated by Alembic - please adjust! ###
op.drop_constraint(None, 'elevationinfo', type_='foreignkey')
op.drop_column('elevationinfo', 'elevation_id')
op.drop_table('elevation')
# ### end Alembic commands ###

78
alembic/versions/f2f205448dff_rename_ntables.py Normal file
View file

@ -0,0 +1,78 @@
"""rename ntables
Revision ID: f2f205448dff
Revises: 42a19efef660
Create Date: 2025-06-18 10:09:39.973162
"""
from typing import Sequence, Union
from alembic import op
import sqlalchemy as sa
import sqlmodel
# revision identifiers, used by Alembic.
revision: str = 'f2f205448dff'
down_revision: Union[str, None] = '42a19efef660'
branch_labels: Union[str, Sequence[str], None] = None
depends_on: Union[str, Sequence[str], None] = None
def upgrade() -> None:
"""Upgrade schema."""
# Drop foreign keys referencing 'heating'
op.drop_constraint('propertydescription_main_heating_id_fkey', 'propertydescription', type_='foreignkey')
op.drop_constraint('propertydescription_main_heating2_id_fkey', 'propertydescription', type_='foreignkey')
# Rename 'heating' table to 'heatingfrompresitenotes'
op.rename_table('heating', 'heatingfrompresitenotes')
# Rename/re-add cleaned up columns (ensure you're within 63 char limit!)
op.drop_column('externalelevation', 'are_there_any_visible_signs_of_cracking_to_the_existing_externa')
op.drop_column('externalelevation', 'are_there_any_signs_of_water_penetration_caused_by_failed_rainw')
op.add_column('externalelevation', sa.Column(
'water_penetration_from_failed_rainwater_goods', sa.Boolean(), nullable=False))
op.add_column('externalelevation', sa.Column(
'cracking_to_external_finish_visible', sa.Boolean(), nullable=False))
# Update propertyaccess fields
op.drop_column('propertyaccess', 'is_there_restricted_space_for_contractors_to_access_the_wall_ar')
op.drop_column('propertyaccess', 'is_there_restricted_space_for_contractors_to_access_the_roof_ar')
op.drop_column('propertyaccess', 'is_there_more_than_1_point_5_meters_in_width_to_fence_or_neighb')
op.add_column('propertyaccess', sa.Column(
'restricted_wall_access', sa.Boolean(), nullable=False))
op.add_column('propertyaccess', sa.Column(
'restricted_roof_access', sa.Boolean(), nullable=False))
op.add_column('propertyaccess', sa.Column(
'sufficient_boundary_width', sa.Boolean(), nullable=False))
# Create new foreign keys to heatingfrompresitenotes
op.create_foreign_key('propertydescription_main_heating_id_fkey', 'propertydescription', 'heatingfrompresitenotes', ['main_heating_id'], ['id'])
op.create_foreign_key('propertydescription_main_heating2_id_fkey', 'propertydescription', 'heatingfrompresitenotes', ['main_heating2_id'], ['id'])
def downgrade() -> None:
"""Downgrade schema."""
# Drop new foreign keys
op.drop_constraint('propertydescription_main_heating_id_fkey', 'propertydescription', type_='foreignkey')
op.drop_constraint('propertydescription_main_heating2_id_fkey', 'propertydescription', type_='foreignkey')
# Rename 'heatingfrompresitenotes' back to 'heating'
op.rename_table('heatingfrompresitenotes', 'heating')
# Revert column changes
op.drop_column('externalelevation', 'water_penetration_from_failed_rainwater_goods')
op.drop_column('externalelevation', 'cracking_to_external_finish_visible')
op.add_column('externalelevation', sa.Column('are_there_any_signs_of_water_penetration_caused_by_failed_rainw', sa.Boolean(), nullable=False))
op.add_column('externalelevation', sa.Column('are_there_any_visible_signs_of_cracking_to_the_existing_externa', sa.Boolean(), nullable=False))
op.drop_column('propertyaccess', 'restricted_wall_access')
op.drop_column('propertyaccess', 'restricted_roof_access')
op.drop_column('propertyaccess', 'sufficient_boundary_width')
op.add_column('propertyaccess', sa.Column('is_there_restricted_space_for_contractors_to_access_the_wall_ar', sa.Boolean(), nullable=False))
op.add_column('propertyaccess', sa.Column('is_there_restricted_space_for_contractors_to_access_the_roof_ar', sa.Boolean(), nullable=False))
op.add_column('propertyaccess', sa.Column('is_there_more_than_1_point_5_meters_in_width_to_fence_or_neighb', sa.Boolean(), nullable=False))
# Restore original foreign keys
op.create_foreign_key('propertydescription_main_heating_id_fkey', 'propertydescription', 'heating', ['main_heating_id'], ['id'])
op.create_foreign_key('propertydescription_main_heating2_id_fkey', 'propertydescription', 'heating', ['main_heating2_id'], ['id'])

10
etl/condition_report_etl.py
View file

@ -1,6 +1,14 @@
from pprint import pprint
from etl.db.db import get_db_session, init_db
from etl.surveyedData.surveryedData import surveyedDataProcessor
condition_report_file_path = "/workspaces/survey-extractor/etl/files/osmosis_condition_report.pdf"
sdp = surveyedDataProcessor("123 Fake Street", [condition_report_file_path])
pprint(sdp.condition_report.master_obj)
init_db()
with get_db_session() as db_session:
sdp.load_condition_report(db_session)
# TODO: add the ability to add document type, and sharepoint or s3 link so we can process access it again
# TODO: add the ability to add document type, and sharepoint or s3 link so we can process access it again
# Terraform lambda set up to start this job from a s3 link

2
etl/daily_script.py
View file

@ -1,5 +1,5 @@
import os
from pdfReader.pdfReaderToText import pdfReaderToText
from fileReader.pdfReaderToText import pdfReaderToText
from etl.scraper.scraper import SharePointScraper, SharePointInstaller, WEEK_COMMENCING
from pprint import pprint, pformat
import logging

2
etl/development.py
View file

@ -1,6 +1,6 @@
from etl.scraper.scraper import SharePointScraper, SharePointInstaller
from pprint import pformat
from etl.pdfReader.pdfReaderToText import pdfReaderToText
from etl.fileReader.pdfReaderToText import pdfReaderToText
from etl.surveyedData.surveryedData import surveyedDataProcessor
import pandas as pd

0
etl/pdfReader/__init__.py → etl/fileReader/__init__.py
View file

19
etl/pdfReader/pdfReaderToText.py → etl/fileReader/pdfReaderToText.py
View file

@ -1,8 +1,8 @@
from etl.utils.logger import Logger
import logging
import pymupdf
from etl.pdfReader.sitenotes import QuidosSiteNotesExtractor, CSR, ConditionReport
from etl.pdfReader.reportType import ReportType
from etl.fileReader.sitenotes import QuidosSiteNotesExtractor, CSR, WarmHomesConditionReport, ECOConditionReport, RDSAPEnergyReport
from etl.fileReader.reportType import ReportType
class pdfReaderToText():
@ -24,6 +24,7 @@ class pdfReaderToText():
self.all_text += text
self.text_list = self.all_text.split('\n')
print(self.text_list)
def get_list_of_text(self):
return self.text_list
@ -41,7 +42,11 @@ class pdfReaderToText():
elif "Chartered Surveyor Report: Recommending Extraction of Defective Cavity Wall Insulation " in self.text_list:
self.type = ReportType.CHARTED_SURVEYOR_REPORT
elif "Osmosis ACD NEW PAS 2035 Condition Report".lower() in self.text_list[0].lower():
self.type = ReportType.OSMOSIS_CONDITION_PAS_2035_REPORT
self.type = ReportType.WARM_HOMES_CONDITION_REPORT
elif "Domna NEW PAS 2035 ECO Condition Report".lower() in self.text_list[0].lower():
self.type = ReportType.ECO_CONDITION_REPORT
elif "ENERGY REPORT".lower() == self.text_list[0].lower():
self.type = ReportType.RDSAP_ENERGY_REPORT
else:
pass
return self.type
@ -53,6 +58,10 @@ class pdfReaderToText():
return QuidosSiteNotesExtractor(self.text_list)
elif self.type == ReportType.CHARTED_SURVEYOR_REPORT:
return CSR(self.text_list)
elif self.type == ReportType.OSMOSIS_CONDITION_PAS_2035_REPORT:
return ConditionReport(self.text_list)
elif self.type == ReportType.WARM_HOMES_CONDITION_REPORT:
return WarmHomesConditionReport(self.text_list)
elif self.type == ReportType.ECO_CONDITION_REPORT:
return ECOConditionReport(self.text_list)
elif self.type == ReportType.RDSAP_ENERGY_REPORT:
return RDSAPEnergyReport(self.text_list)

15
etl/fileReader/reportType.py Normal file
View file

@ -0,0 +1,15 @@
from enum import Enum
class ReportType(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"
ECO_CONDITION_REPORT = "osmosis_condition_pas_2035_report"
WARM_HOMES_CONDITION_REPORT = "warm_homes_condition_pas_2035_report"
RDSAP_ENERGY_REPORT = "rdsap_energy_report"
LIG_XML = "lodgement_xml_needed_for_lodgement_to_like_trademark"
RDSAP_XML = "reduce_xml_needed_to_generate_full_sap_xml"
FULLSAP_XML = "full_xml_needed_for_co_ordination"

81
etl/pdfReader/sitenotes.py → etl/fileReader/sitenotes.py
View file

@ -1,11 +1,11 @@
from etl.pdfReader.reportType import ReportType
from etl.fileReader.reportType import ReportType
from etl.transform.preSiteNoteTypes import (
CompanyInfo, PreSiteNotesSummaryInfo, AssessorInfo,
PropertyDescription, PropertyDetail, Dimension,
Walls, Roofs, Floors, Door, VentilationAndCooling,
Lighting, WaterHeating, HotWaterCylinder, SolarWaterHeating,
ShowerAndBaths, FlueGasHeatRecoverySystem, PhotovoltaicPanel,
WindTurbine, OtherDetails, Windows, Heating, HeatingSystemControls,
WindTurbine, OtherDetails, Windows, HeatingFromPreSiteNotes, HeatingSystemControls,
HeatingType, Insulation
)
from etl.transform.conditionReportTypes import (
@ -14,7 +14,8 @@ from etl.transform.conditionReportTypes import (
ExternalElevationGableOne, ExternalElevationGableTwo, ExternalElevationRear, ConservatoryOrOutbuilding,
AccessAndElevations, Hallway, RoomInfo, WindowsInfo, VentilationInfo, LivingRoom, DiningRoom, Kitchen, Rooms,
Utility, WC, Landing, Bedroom, Bathroom, LoftSpace, RoomInRoof, HeatingSystem, GeneralConditionHeatingSystem,
MainHeatingOne, MainHeatingTwo, SecondaryHeating, HeatingByRoom, Renewables, Occupant, EnergyUse, Heating, ShowerAndBath, FridgeAndFreezers, Cooker, TumbleDryer
MainHeatingOne, MainHeatingTwo, SecondaryHeating, HeatingByRoom, Renewables, Occupant, EnergyUse, HeatingFromConditionReport, ShowerAndBath, FridgeAndFreezers, Cooker, TumbleDryer,
GeneralInformation, OccupantAssessment
)
from datetime import datetime
from pprint import pprint
@ -87,30 +88,52 @@ class CSR(SiteNotesExtractor):
type=dict_.get('detailed_description_of_existing_cavity_wall_insulation_', "")
) if dict_ is not None else None
class ConditionReport(SiteNotesExtractor):
class RDSAPEnergyReport(SiteNotesExtractor):
def __init__(self, data_list):
super().__init__(data_list)
self.type = ReportType.OSMOSIS_CONDITION_PAS_2035_REPORT
self.setup_condition_report()
self.type = ReportType.RDSAP_ENERGY_REPORT
self.master_obj = self.setup_energy_report()
def setup_energy_report(self):
pass
class ECOConditionReport(SiteNotesExtractor):
def __init__(self, data_list):
super().__init__(data_list)
self.type = ReportType.ECO_CONDITION_REPORT
self.master_obj = self.setup_condition_report()
def setup_condition_report(self):
assesor_details, inspection_and_project, the_property, main_elevation, elevations = self.get_section_1()
pass
class WarmHomesConditionReport(SiteNotesExtractor):
def __init__(self, data_list):
super().__init__(data_list)
self.type = ReportType.WARM_HOMES_CONDITION_REPORT
self.master_obj = self.setup_condition_report()
def setup_condition_report(self):
general_information = self.get_section_1()
access_and_elevations = self.get_section_2()
rooms = self.get_section_3()
general_condition_of_heating_system, main_heating_one, main_heating_two, secondary_heating, heating_by_room, renewables = self.get_section_4()
occupants, energy_use, heating, shower_and_bath, appliances, fridge_and_freezers, cooker, tumble_dryer = self.get_section_5()
heating_system = self.get_section_4()
occupant_assessment = self.get_section_5()
site_name, reference_code, address, postcode = self.get_section_0()
return ConditionReportModel(
project_site_name=site_name,
property_reference_code=reference_code,
property_address=address,
postcode=postcode,
general_information=general_information,
access_and_elevations=access_and_elevations,
rooms=rooms,
heating_system=heating_system,
occupancy_assessment=occupant_assessment,
)
def get_section_0(self):
data = self.get_data_between("Project Site Name", "1. General Information")
site_name = self.get_next_value(data, "Project Site Name")
@ -127,7 +150,14 @@ class ConditionReport(SiteNotesExtractor):
the_property = self.get_the_property()
main_elevation = self.get_main_elevation()
elevations = self.get_all_elevations()
return assessor_details, inspection_and_project, the_property, main_elevation, elevations
return GeneralInformation(
assessor_details=assessor_details,
inspection_and_project=inspection_and_project,
the_property=the_property,
main_elevation=main_elevation,
elevations=elevations
)
def get_assessor_details(self):
data = self.get_data_between("1.1 Assessor details","1.2 Inspection & Project")
@ -501,7 +531,15 @@ class ConditionReport(SiteNotesExtractor):
secondary_heating = self.get_secondary_heating()
heating_by_room = self.get_heating_by_room()
renewables = self.get_renewables()
return general_condition_of_heating_system, main_heating_one, main_heating_two, secondary_heating, heating_by_room, renewables
return HeatingSystem(
general_condition=general_condition_of_heating_system,
main_heating_one=main_heating_one,
main_heating_two=main_heating_two,
secondary_heating=secondary_heating,
heating_by_room=heating_by_room,
renewables=renewables
)
def get_main_heating_one(self):
data = self.get_data_between("Main Heating 1", "Main Heating 2")
@ -589,7 +627,16 @@ class ConditionReport(SiteNotesExtractor):
fridge_and_freezers = self.get_fridge_and_freezers()
cooker = self.get_cooker()
tumble_dryer = self.get_tumble_dryer()
return occupants, energy_use, heating, shower_and_bath, appliances, fridge_and_freezers, cooker, tumble_dryer
return OccupantAssessment(
occupant=occupants,
energy_use=energy_use,
heating=heating,
shower_and_bath=shower_and_bath,
appliances=appliances,
fridge_and_freezers=fridge_and_freezers,
cooker=cooker,
tumble_dryer=tumble_dryer
)
def get_occupants(self):
data = self.get_data_between("Occupants", "Energy use")
@ -616,7 +663,7 @@ class ConditionReport(SiteNotesExtractor):
def get_heating(self):
data = self.get_data_between("Heating", "Shower & bath")
return Heating(
return HeatingFromConditionReport(
room_stat_in_temperature_in_celsius=self.get_next_value(data, "Room Stat Temperature (in °C)", avoid=["Room Stat Location", '\xa0']),
room_stat_location=self.get_next_value(data, "Room Stat Location", avoid = ["Is the heating pattern known?", '\xa0']),
is_the_heating_pattern_known=self.get_next_value(data, "Is the heating pattern known?", avoid=["Shower & bath", '\xa0']),

43
etl/fileReader/xmlReader.py Normal file
View file

@ -0,0 +1,43 @@
from etl.utils.logger import Logger
import logging
from xml.dom.minidom import parse
import os
from etl.fileReader.reportType import ReportType
class xmlReader():
def __init__(self, file_path):
self.source_path = file_path
self.logger = Logger(name='xmlReader', level=logging.INFO).get_logger()
self.xml_obj = None
self.type = None
self.get_xml_obj()
def get_xml_obj(self):
try:
if not os.path.exists(self.source_path):
self.logger.error(f"File not found: {self.source_path}")
return None
with open(self.source_path, 'r', encoding='utf-8') as file:
self.xml_obj = parse(file)
self.get_type()
return self.xml_obj
except Exception as e:
self.logger.error(f"Failed to parse XML file {self.source_path}: {e}")
self.xml_obj = None
return self.xml_obj
def get_type(self):
xmlHeaderName = self.xml_obj.documentElement.tagName
xmlHeaderName = xmlHeaderName.lower()
if xmlHeaderName == 'RdSap-Report'.lower():
self.type = ReportType.LIG_XML
elif xmlHeaderName == "SurveyRec".lower():
self.type = ReportType.RDSAP_XML
elif xmlHeaderName == "ImportExportRecord".lower():
self.type = ReportType.FULLSAP_XML
else:
pass
return self.type

92
etl/hubSpotClient/hubspot.py
View file

@ -7,7 +7,10 @@ import time
from pydantic import ValidationError
from etl.utils.logger import Logger
import logging
import traceback
from hubspot.crm.objects.notes import SimplePublicObjectInput as NoteInput
from hubspot.crm.associations import BatchInputPublicAssociation, PublicAssociation
import time
class DealStage(Enum):
@ -41,6 +44,7 @@ class HubSpotClient():
except Exception as e:
return "Unknown Deal" # Fallback if the deal name is not found
def get_listings_from_deals_id(self, deals_id):
from hubspot.crm.objects import PublicObjectSearchRequest
found_notes = []
@ -77,7 +81,7 @@ class HubSpotClient():
def get_domna_and_landlord_id(self, deals_id):
data = self.get_listings_from_deals_id(deals_id)
return data.properties['domna_property_id'], data.properties['owner_property_id'], data.properties['national_uprn']
return data.properties['domna_property_id'], data.properties['owner_property_id'], data.properties.get('national_uprn', '') or ''
def get_notes_from_deals_id(self, deals_id):
from hubspot.crm.objects import PublicObjectSearchRequest
@ -205,6 +209,9 @@ class HubSpotClient():
for deal in found_deals:
domna_id, landlord_id, uprn = self.get_domna_and_landlord_id(deal.id)
try:
deal_name = deal.properties['dealname']
self.logger.info(f"Validating <{deal_name}>")
# input(f"Press enter to verfiy <{deal_name}>")
all_deals.append(SubmissionInfoFromDeal(
deal_id= deal.properties["hs_object_id"],
deal_name=deal.properties["dealname"],
@ -220,14 +227,46 @@ class HubSpotClient():
uprn = uprn,
))
except Exception as e:
def format_error_note(e):
note_text = "⚠️ <b>Automated Verification Failed:</b><br><br>"
if hasattr(e, "errors") and callable(e.errors):
note_text += "❌ <b>Validation Errors:</b><br>"
for error in e.errors():
loc = error.get('loc', 'N/A')
msg = error.get('msg', 'N/A')
error_type = error.get('type', 'N/A')
error_input = error.get('input', 'N/A')
note_text += (
f"• <b>Field:</b> <code>{loc}</code><br>"
f"&nbsp;&nbsp;- <b>Message:</b> {msg}<br>"
f"&nbsp;&nbsp;- <b>Type:</b> {error_type}<br>"
f"&nbsp;&nbsp;- <b>Input:</b> {error_input}<br><br>"
)
else:
note_text += (
"❗ <b>Non-validation error:</b><br>"
f"<pre>{str(e)}</pre><br>"
)
note_text += (
"🛠️ Please review this error and take necessary actions.<br>"
"Contact <b>Jun-te</b> if help is needed: <b>+44 7519 530 549</b> or via Teams."
)
return note_text
deal_id = deal.properties['hs_object_id']
self.logger.info(f"Deal <{deal_id}> not valid")
if hasattr(e, "errors"):
for error in e.errors():
self.add_note_to_deal(deal_id, format_error_note(e))
else:
self.logger.error(f"Non-validation error occurred: {str(e)}", exc_info=True)
self.logger.info(f"Deal name <{deal_name}> moving to 'needs additional information'")
self.move_deals_to_different_stage([deal_id], DealStage.NEEDS_ADDITIONAL_INFORMATION_FROM_ASSESSOR.value)
return all_deals
def print_all_pipeline_ids(self):
@ -250,3 +289,40 @@ class HubSpotClient():
simple_public_object_input=deal_properties
)
self.logger.info(f"Deal {deal_id} moved to stage with ID {to_stage_id}.")
def add_note_to_deal(self, deal_id, note_text):
try:
# Generate current time in milliseconds since epoch
hs_timestamp = int(time.time() * 1000)
# Step 1: Create the note with hs_timestamp
note = NoteInput(
properties={
"hs_note_body": note_text,
"hs_timestamp": hs_timestamp # Required field in your HubSpot setup
}
)
created_note = self.client.crm.objects.notes.basic_api.create(note)
note_id = created_note.id
# Step 2: Associate the note to the deal
association = PublicAssociation(
_from=note_id,
to=deal_id,
type="note_to_deal"
)
self.client.crm.associations.batch_api.create(
'notes',
'deals',
batch_input_public_association=BatchInputPublicAssociation(
inputs=[association]
)
)
self.logger.info(f"📝 Note added to deal {deal_id}: {note_text}")
except Exception as e:
self.logger.error(f"❌ Failed to add note to deal {deal_id}: {e}", exc_info=True)

73
etl/hubSpotClient/types.py
View file

@ -2,9 +2,14 @@ from sqlmodel import Field, SQLModel
from sqlalchemy import Column
from sqlalchemy.dialects.postgresql import UUID
import uuid
from pydantic import Field, field_validator, ValidationError, model_validator
from pydantic import Field, field_validator, model_validator
from etl.utils.utils import get_sharepoint_path
from etl.scraper.scraper import SharePointScraper, SharePointInstaller
from etl.surveyedData.surveryedData import surveyedDataProcessor
def string_to_installer(installer):
if installer.upper() == "J & J CRUMP":
@ -12,7 +17,7 @@ def string_to_installer(installer):
elif installer.upper() == "SCIS":
return SharePointInstaller.SOUTH_COAST_INSULATION
elif installer.upper() == "SGEC":
return SharePointInstaller.SGEC
return SharePointInstaller.JJC
else:
return None
@ -36,23 +41,63 @@ class SubmissionInfoFromDeal(BaseModel):
submission_folder_path: str = Field(..., min_length=1)
landlord_id: str = Field(..., min_length=1)
domna_id: str = Field(..., min_length=1)
uprn: str = Field(..., min_length=1)
uprn: str
@field_validator('post_sap_score', 'no_of_wet_rooms')
@classmethod
def must_be_non_negative(cls, v):
def must_be_non_negative(cls, v, info):
if v < 0:
raise ValidationError("Must be non-negative for Post Sap Score")
raise ValueError(f"{info.field_name} must be non-negative")
return v
@model_validator(mode="after")
def check_submission_folder_path(self):
path = get_sharepoint_path(self.submission_folder_path)
installer = string_to_installer(self.installer)
sp = SharePointScraper(installer)
files = sp.get_folders_in_path(path)
if "value" in files:
if len(files["value"]) > 0:
return self
def check_sharepoint_link_and_contents(self):
try:
path = get_sharepoint_path(self.submission_folder_path)
installer = string_to_installer(self.installer)
sp = SharePointScraper(installer)
except Exception as e:
raise ValueError(f"Error accessing SharePoint path: {self.submission_folder_path}. Error: {str(e)}")
try:
# Check if sharepoint link is reachable and has any contents
files = sp.get_folders_in_path(path)
if "value" in files and len(files["value"]) > 0:
pass
else:
raise ValueError(f"SharePoint folder is empty: {self.submission_folder_path}")
except Exception as e:
raise ValueError(str(e))
# download files in url and check files are there:
try:
files = sp.download_files_from_path(path)
print(files)
sdp = surveyedDataProcessor("fake address", files)
missing_items = []
if sdp.condition_report is None:
missing_items.append("Condition Report")
if sdp.energy_report is None:
missing_items.append("Energy Report PDF")
if sdp.rd_sap_xml is None:
missing_items.append("RDSAP XML")
if sdp.lig_sap_xml is None:
missing_items.append("LIG SAP XML")
if missing_items:
raise ValueError(f"Missing required items: {', '.join(missing_items)}")
except Exception as e:
raise ValueError(str(e))
raise RuntimeError("Sharepoint URL invalid")
return self

25
etl/hubspot_surveyed_needs_sign_off.py Normal file
View file

@ -0,0 +1,25 @@
"""
This is the script that runs when we are at the 'surveyed-needs sign off' stage within hubspot
"""
import os
from pprint import pprint
os.environ["SHAREPOINT_CLIENT_ID"] = "895e3b77-b1d7-43ec-b18f-dcfe07cdfeaf"
os.environ["SHAREPOINT_CLIENT_SECRET"] = "SOf8Q~-is4wdQiqvEEm9FlJQRAY9ELGaj5Qz-a6E"
os.environ["SHAREPOINT_TENANT_ID"] = "c3f7519c-2719-4547-af04-6da6cbfd8f8f"
os.environ["SOUTH_COAST_INSULATION_SERVICE_SHAREPOINT_ID"] = "b5a51507-9427-4ee0-b03e-90ec7681e2d3"
os.environ["JJC_SERVICE_SHAREPOINT_ID"] = "7fdd0485-bbf3-4b29-b30f-98c81c2a6284"
from etl.hubSpotClient.hubspot import DealStage, HubSpotClient
# Local development
os.environ["DATABASE_URL"] = "postgresql://postgres:makingwarmhomes@db:5432/postgres"
hubspotClient = HubSpotClient()
deals = hubspotClient.get_deals_from_deal_stage(DealStage.SURVEYED_COMPLETE_NEEDS_SIGN_OFF)
for deal in deals:
hubspotClient.move_deals_to_different_stage([deal.deal_id], DealStage.SURVEYED_COMPLETED_SIGNED_OFF.value)
# TODO load when we are at 'ready to co-ordination' - script!

2
etl/jjc_old_lewis_manual_way_.py
View file

@ -5,7 +5,7 @@ os.environ["SHAREPOINT_TENANT_ID"] = "c3f7519c-2719-4547-af04-6da6cbfd8f8f"
os.environ["JJC_SERVICE_SHAREPOINT_ID"] = "7fdd0485-bbf3-4b29-b30f-98c81c2a6284"
from etl.scraper.scraper import SharePointScraper, SharePointInstaller, WEEK_COMMENCING
from pprint import pformat
from etl.pdfReader.pdfReaderToText import pdfReaderToText
from etl.fileReader.pdfReaderToText import pdfReaderToText
from etl.surveyedData.surveryedData import surveyedDataProcessor
import pandas as pd
import math

386
etl/models/conditionReport.py Normal file
View file

@ -0,0 +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
class AssessorDetails(BaseModel, table=True):
assessor_name_and_id: str
elmhurst_id: 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 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")
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")
#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
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")
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 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 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 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 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()
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 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: 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 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 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 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 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 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()
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()
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()
kitchen_id: uuid.UUID = Field(foreign_key="kitchen.id")
kitchen: Kitchen = 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()
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()
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")
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 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 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 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_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()
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()
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 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 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 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()
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()
shower_and_bath_id: uuid.UUID = Field(foreign_key="showerandbath.id")
shower_and_bath: ShowerAndBath = Relationship()
# appliances: Optional[Appliances]
# appliances_id
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()
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
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()
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()
occupancy_assessment_id: uuid.UUID = Field(foreign_key="occupantassessment.id")
occupancy_assessment: OccupantAssessment = Relationship()

13
etl/load/preSiteNoteTypes.py → etl/models/preSiteNoteTypes.py
View file

@ -5,7 +5,7 @@ from datetime import datetime
from pydantic import EmailStr
from sqlalchemy import Column
from sqlalchemy.dialects.postgresql import UUID
from etl.load.topLevel import BaseModel, Documents
from etl.models.topLevel import BaseModel, Documents
class PreSiteNote(BaseModel, table=True):
@ -127,7 +127,7 @@ class HeatingSystemControls(BaseModel, table=True):
mains_gas_available: Optional[bool] = False
class Heating(BaseModel, table=True):
class HeatingFromPreSiteNotes(BaseModel, table=True):
type: str
heating_source: str
efficiency_source: str
@ -252,8 +252,8 @@ class PropertyDescription(BaseModel, table=True):
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="heating.id")
main_heating2_id: Optional[uuid.UUID] = Field(default=None, foreign_key="heating.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
@ -275,8 +275,8 @@ class PropertyDescription(BaseModel, table=True):
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["Heating"] = Relationship(back_populates="property_description", sa_relationship_kwargs={"foreign_keys": "[PropertyDescription.main_heating_id]"})
main_heating2: Optional["Heating"] = Relationship(back_populates="property_description", sa_relationship_kwargs={"foreign_keys": "[PropertyDescription.main_heating2_id]"})
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")
@ -326,7 +326,6 @@ class CompanyInfo(BaseModel, table=True):
assessors: List[AssessorInfo] = Relationship(back_populates="company")
class Insulation(BaseModel, table=True):
type: str

2
etl/load/topLevel.py → etl/models/topLevel.py
View file

@ -5,7 +5,7 @@ from datetime import datetime
from pydantic import EmailStr
from sqlalchemy import Column
from sqlalchemy.dialects.postgresql import UUID
from etl.pdfReader.reportType import ReportType
from etl.fileReader.reportType import ReportType
class BaseModel(SQLModel):
id: uuid.UUID = Field(default_factory=uuid.uuid4, primary_key=True)

BIN
etl/osmosis_data/asset_list.xlsx
View file

Binary file not shown.

2
etl/osmosis_monday_to_sharepoint_automation.py
View file

@ -13,7 +13,7 @@ from tqdm import tqdm
osmosis = SharePointScraper(SharePointInstaller.OSMOSIS_WAVE_2)
parent_folder = "/Osmosis ACD/Osmosis ACD Projects/Stonewater/Stonewater Property ID Folders/12. Decent Homes"
parent_folder = "/Osmosis ACD/Osmosis ACD Projects/WCHG/WCHG Walkups/Property Folders"
asset_list = pd.read_excel("osmosis_data/asset_list.xlsx", sheet_name="Sheet1")

10
etl/pdfReader/reportType.py
View file

@ -1,10 +0,0 @@
from enum import Enum
class ReportType(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"

2
etl/scis_invoice.py
View file

@ -1,6 +1,6 @@
from etl.scraper.scraper import SharePointScraper, SharePointInstaller, WEEK_COMMENCING
from pprint import pformat
from etl.pdfReader.pdfReaderToText import pdfReaderToText
from etl.fileReader.pdfReaderToText import pdfReaderToText
from etl.surveyedData.surveryedData import surveyedDataProcessor
import pandas as pd

38
etl/scraper/scraper.py
View file

@ -13,7 +13,6 @@ from datetime import datetime, timedelta
def previous_monday():
today = datetime.today()
print(f"Todays date is {today}")
monday = today - timedelta(days=today.weekday()) # weekday() = 0 for Monday
return f"W.C. {monday.strftime('%d.%m.%Y')}"
@ -302,13 +301,46 @@ class SharePointScraper():
file_names_to_download.update({file["name"]: file['@microsoft.graph.downloadUrl']})
each_file = []
for file_name, url in file_names_to_download.items():
self.logger.info(f"Downloading {file_name} from {url}")
self.logger.debug(f"Downloading {file_name} from {url}")
content = self.get_file_content(url)
file_path = self.create_temp_file(content, f"{name}/{WEEK_COMMENCING}/{house_ass}/{address}/{file_name}")
each_file.append(file_path)
address_paths.update({address: each_file})
paths.append(address_paths)
return paths
def download_files_from_path(self, path):
"""
Download all non-media files from a list of root paths.
Args:
root_paths (List[str]): List of full folder paths to start from.
Returns:
List[Dict[str, List[str]]]: A list of dictionaries mapping address folder names to downloaded file paths.
"""
avoid = [".jpg", ".mov", ".JPG", ".heic", ".HEIC", ".png", ".PNG", ".jpeg", ".JPEG", ".mp4", ".MP4"]
files_info = self.get_folders_in_path(path)
if 'value' not in files_info:
raise RuntimeError(f"Failed to get files from {path}")
file_names_to_download = {
file["name"]: file["@microsoft.graph.downloadUrl"]
for file in files_info['value']
if 'file' in file and not any(file["name"].endswith(ext) for ext in avoid)
}
downloaded_files = []
for file_name, url in file_names_to_download.items():
self.logger.info(f"Downloading {file_name} from {url}")
content = self.get_file_content(url)
file_path = self.create_temp_file(content, f"{path}/{file_name}")
downloaded_files.append(file_path)
return downloaded_files
def create_temp_file(self, content, path):
# Ensure the path is under /tmp/
@ -321,6 +353,6 @@ class SharePointScraper():
with open(path, 'wb+') as temp_file:
temp_file.write(content.getvalue())
self.logger.info(f"Temporary file created at: {path}")
self.logger.debug(f"Temporary file created at: {path}")
return path

2
etl/sgec_invoice.py
View file

@ -1,6 +1,6 @@
from etl.scraper.scraper import SharePointScraper, SharePointInstaller, WEEK_COMMENCING
from pprint import pformat
from etl.pdfReader.pdfReaderToText import pdfReaderToText
from etl.fileReader.pdfReaderToText import pdfReaderToText
from etl.surveyedData.surveryedData import surveyedDataProcessor
import pandas as pd

732
etl/surveyedData/surveryedData.py
View file

@ -1,16 +1,34 @@
from etl.pdfReader.pdfReaderToText import pdfReaderToText
from etl.pdfReader.reportType import ReportType
from etl.fileReader.pdfReaderToText import pdfReaderToText
from etl.fileReader.xmlReader import xmlReader
from etl.fileReader.reportType import ReportType
import math
from etl.load.preSiteNoteTypes import (
from xml.dom.minidom import parseString
from etl.models.preSiteNoteTypes import (
AssessorInfo, CompanyInfo,
PreSiteNotesSummaryInfo,
PreSiteNote,
PropertyDescription, Dimension, HeatingType, Heating, HeatingSystemControls,
PropertyDescription, Dimension, HeatingType, HeatingFromPreSiteNotes, HeatingSystemControls,
OtherDetails, WindTurbine, PhotovoltaicPanel, FlueGasHeatRecoverySystem, ShowerAndBaths,
SolarWaterHeating, HotWaterCylinder, WaterHeating, Lighting, VentilationAndCooling,
Door, Walls, Roofs, Floors, PropertyDetail, Windows
)
from etl.load.topLevel import(
from pprint import pprint
from etl.models.conditionReport import (
AssessorDetails, InspectionAndProject, TheProperty, ElevationInfo,
MainElevation, Elevation,GeneralInformation,
PropertyAccess, ExternalElevation,
ExternalElevationFront, ExternalElevationRear, ExternalElevationGableOne,
ExternalElevationGableTwo, ConservatoryOrOutbuilding, AccessAndElevations,Hallway,
VentilationInfo,WindowsInfo,RoomInfo,LivingRoom, DiningRoom, Kitchen, Utility, WC,
Landing,LoftSpace,RoomInRoof, Bedroom,Bathroom, Rooms,
GeneralConditionHeatingSystem, MainHeatingOne, MainHeatingTwo,
SecondaryHeating, HeatingByRoom, Renewables, HeatingSystem, Occupant, EnergyUse,
HeatingFromConditionReport, ShowerAndBath, FridgeAndFreezers, Cooker, TumbleDryer,
OccupantAssessment, ConditionReportModel
)
from etl.models.topLevel import(
Buildings, Documents
)
import uuid
@ -22,21 +40,663 @@ class surveyedDataProcessor():
self.pre_site_note = None
self.csr = None
self.condition_report = None
self.identify_files()
self.hubspot_deal_id = None
self.energy_report = None
self.full_sap_xml = None
self.lig_sap_xml = None
self.rd_sap_xml = None
self.identify_files()
def identify_files(self):
for file in self.files:
pdf = pdfReaderToText(file)
if pdf:
if pdf.type == ReportType.QUIDOS_PRESITE_NOTE:
self.pre_site_note = pdf.get_reader()
self.address = self.pre_site_note.survey_information.address
elif pdf.type == ReportType.CHARTED_SURVEYOR_REPORT:
self.csr = pdf.get_reader()
elif pdf.type == ReportType.OSMOSIS_CONDITION_PAS_2035_REPORT:
self.condition_report = pdf.get_reader()
if file.lower().endswith('.pdf'):
pdf = pdfReaderToText(file)
if pdf:
if pdf.type == ReportType.QUIDOS_PRESITE_NOTE:
self.pre_site_note = pdf.get_reader()
self.address = self.pre_site_note.survey_information.address
elif pdf.type == ReportType.CHARTED_SURVEYOR_REPORT:
self.csr = pdf.get_reader()
elif pdf.type == ReportType.WARM_HOMES_CONDITION_REPORT:
self.condition_report = pdf.get_reader()
elif pdf.type == ReportType.ECO_CONDITION_REPORT:
self.condition_report = pdf.get_reader()
elif pdf.type == ReportType.RDSAP_ENERGY_REPORT:
self.energy_report = pdf.get_reader()
elif file.lower().endswith('.xml'):
xml = xmlReader(file)
if xml:
if xml.type is ReportType.FULLSAP_XML:
self.full_sap_xml = xml.xml_obj
elif xml.type is ReportType.LIG_XML:
self.lig_sap_xml = xml.xml_obj
elif xml.type is ReportType.RDSAP_XML:
self.rd_sap_xml = xml.xml_obj
def load_condition_report(self, db_session):
general_information = self.load_general_information_from_condition_report(db_session)
access_and_elevations = self.load_access_and_elevations_from_condition_report(db_session)
rooms = self.load_rooms_from_condition_report(db_session)
heating_system = self.load_heating_system_from_condition_report(db_session)
occupant_assessment = self.load_occupant_info_from_condition_report(db_session)
condition_report = self.get_attribute_and_load(
self.condition_report,
"master_obj",
ConditionReportModel,
db_session,
additional_fields={
"general_information_id": general_information.id,
"access_and_elevations_id": access_and_elevations.id,
"rooms_id": rooms.id,
"heating_system_id": heating_system.id,
"occupancy_assessment_id": occupant_assessment.id,
},
)
# Create building table
data = {
"address": self.pre_site_note.survey_information.address,
"postcode": self.pre_site_note.survey_information.postcode,
"UPRN": self.pre_site_note.survey_information.uprn,
"landlord_id": landlord_id,
"domna_id": domna_id
}
building = self.upsert_record(
db_session=db_session,
model_class=Buildings,
data_dict=data,
lookup_field="UPRN",
)
return building
# Create document table
def load_occupant_info_from_condition_report(self, db_session):
occupant_info = self.load_occupant_information_from_condition_report(db_session)
energy_use = self.load_energy_use_from_condition_report(db_session)
heating = self.load_heating_from_condition_report(db_session)
showerAndBath = self.load_shower_and_bath_from_condition_report(db_session)
# The example I (Jun-te) had, had no appliances so skipped until I have one that needs this information
appliances = self.load_appliances_from_condition_report(db_session)
fridgeAndFreezers = self.load_fridge_and_freezer_from_condition_report(db_session)
cooker = self.load_cooker_from_condition_report(db_session)
tumble_dryer = self.load_tumble_dryer_from_condition_report(db_session)
return self.get_attribute_and_load(
self.condition_report.master_obj,
"occupancy_assessment",
OccupantAssessment,
db_session,
additional_fields={
"occupant_id": occupant_info.id,
"energy_use_id": energy_use.id,
"heating_id": heating.id,
"shower_and_bath_id": showerAndBath.id,
"fridge_and_freezers_id": fridgeAndFreezers.id,
"cooker_id": cooker.id,
"tumble_dryer_id": tumble_dryer.id
}
)
def load_tumble_dryer_from_condition_report(self, db_session):
return self.get_attribute_and_load(
self.condition_report.master_obj.occupancy_assessment,
"tumble_dryer",
TumbleDryer,
db_session,
)
def load_cooker_from_condition_report(self, db_session):
return self.get_attribute_and_load(
self.condition_report.master_obj.occupancy_assessment,
"cooker",
Cooker,
db_session,
)
def load_fridge_and_freezer_from_condition_report(self, db_session):
return self.get_attribute_and_load(
self.condition_report.master_obj.occupancy_assessment,
"fridge_and_freezers",
FridgeAndFreezers,
db_session,
)
def load_appliances_from_condition_report(self, db_session):
print("Appliances was not done, please contact jun-te to complete this!")
return None
def load_shower_and_bath_from_condition_report(self, db_session):
return self.get_attribute_and_load(
self.condition_report.master_obj.occupancy_assessment,
"shower_and_bath",
ShowerAndBath,
db_session
)
def load_heating_from_condition_report(self, db_session):
return self.get_attribute_and_load(
self.condition_report.master_obj.occupancy_assessment,
"heating",
HeatingFromConditionReport,
db_session
)
def load_energy_use_from_condition_report(self, db_session):
return self.get_attribute_and_load(
self.condition_report.master_obj.occupancy_assessment,
"energy_use",
EnergyUse,
db_session,
)
def load_occupant_information_from_condition_report(self, db_session):
return self.get_attribute_and_load(
self.condition_report.master_obj.occupancy_assessment,
"occupant",
Occupant,
db_session,
additional_fields={
"landlord_wrote_that_the_tenent_agrees_assessment_been_supplied":self.condition_report.master_obj.occupancy_assessment.occupant.landlord_has_written_confirmation_that_the_tenent_agrees_to_the_assessment_been_supplied
}
)
def load_heating_system_from_condition_report(self, db_session):
general_condition = self.load_general_condition_of_heating_system_from_condition_report(db_session)
main_heating_one = self.load_main_heating_one_from_condition_report(db_session)
main_heating_two = self.load_main_heating_two_from_condition_report(db_session)
secondary_heating = self.load_secondary_heating_from_condition_report(db_session)
heating_by_room = self.load_heating_by_room_from_condition_report(db_session)
renwables = self.load_renewables_from_condition_report(db_session)
return self.get_attribute_and_load(
self.condition_report.master_obj,
"heating_system",
HeatingSystem,
db_session,
additional_fields={
"general_condition_id": general_condition.id,
"main_heating_one_id": main_heating_one.id,
"main_heating_two_id": main_heating_two.id,
"secondary_heating_id": secondary_heating.id,
"heating_by_room_id": heating_by_room.id,
"renewables_id": renwables.id
},
)
def load_renewables_from_condition_report(self, db_session):
return self.get_attribute_and_load(
self.condition_report.master_obj.heating_system,
"renewables",
Renewables,
db_session,
)
def load_heating_by_room_from_condition_report(self, db_session):
return self.get_attribute_and_load(
self.condition_report.master_obj.heating_system,
"heating_by_room",
HeatingByRoom,
db_session,
exclude_list=False
)
def load_secondary_heating_from_condition_report(self, db_session):
return self.get_attribute_and_load(
self.condition_report.master_obj.heating_system,
"secondary_heating",
SecondaryHeating,
db_session,
)
def load_main_heating_two_from_condition_report(self, db_session):
return self.get_attribute_and_load(
self.condition_report.master_obj.heating_system,
"main_heating_two",
MainHeatingTwo,
db_session,
)
def load_main_heating_one_from_condition_report(self, db_session):
return self.get_attribute_and_load(
self.condition_report.master_obj.heating_system,
"main_heating_one",
MainHeatingOne,
db_session,
)
def load_general_condition_of_heating_system_from_condition_report(self, db_session):
return self.get_attribute_and_load(
self.condition_report.master_obj.heating_system,
"general_condition",
GeneralConditionHeatingSystem,
db_session,
)
def load_rooms_from_condition_report(self, db_session):
hallway = self.load_hallway_from_condition_report(db_session)
living_room = self.load_living_room_from_condition_report(db_session)
dining_room = self.load_dining_room_from_condition_report(db_session)
kitchen = self.load_kitchen_from_condition_report(db_session)
utility_room = self.load_utility_room_from_condition_report(db_session)
wash_chamber = self.load_wash_chamber_from_condition_report(db_session)
landing = self.load_landing_from_condition_report(db_session)
loft_space = self.load_loft_space_from_condition_report(db_session)
rir = self.load_room_in_roof_from_condition_report(db_session)
rooms = self.upsert_record(
model_class=Rooms,
data_dict=self.condition_report.master_obj.rooms.model_dump(),
db_session=db_session,
additional_fields={
"hallway_id": hallway.id,
"living_room_id": living_room.id,
"dining_room_id": dining_room.id,
"kitchen_id": kitchen.id,
"utility_id": utility_room.id,
"wash_chamber_id": wash_chamber.id,
"landing_id": landing.id,
"loft_space_id": loft_space.id,
"room_in_roof_id": rir.id,
}
)
bedrooms = self.load_bedrooms_from_condition_report(db_session, rooms.id)
bathrooms = self.load_bathrooms_from_condition_report(db_session, rooms.id)
return rooms
def load_bathrooms_from_condition_report(self, db_session, rooms_id):
bathroom_ids = []
for i, bathrooms in enumerate(self.condition_report.master_obj.rooms.bathrooms):
room_info = self.load_room_info_from_condition_report(
db_session,
self.condition_report.master_obj.rooms.bathrooms[i]
)
bathroom = self.upsert_record(
db_session=db_session,
model_class=Bathroom,
data_dict=bathrooms.model_dump(),
additional_fields={
"room_info_id": room_info.id,
"rooms_id": rooms_id,
}
)
bathroom_ids.append(bathroom.id)
return bathroom_ids
def load_bedrooms_from_condition_report(self, db_session, rooms_id):
bedroom_ids = []
for i, bedrooms in enumerate(self.condition_report.master_obj.rooms.bedrooms):
room_info = self.load_room_info_from_condition_report(
db_session,
bedrooms
)
bedroom = self.upsert_record(
db_session=db_session,
model_class=Bedroom,
data_dict=bedrooms.model_dump(),
additional_fields={
"room_info_id": room_info.id,
"rooms_id": rooms_id,
}
)
bedroom_ids.append(bedroom.id)
return bedroom_ids
def load_room_in_roof_from_condition_report(self, db_session):
room_info = None
if self.condition_report.master_obj.rooms.room_in_roof.is_there_a_room_in_roof:
room_info = self.load_room_info_from_condition_report(
db_session,
self.condition_report.master_obj.rooms.room_in_roof,
)
return self.get_attribute_and_load(
self.condition_report.master_obj.rooms,
"room_in_roof",
RoomInRoof,
db_session,
additional_fields={
"room_info_id": room_info.id if room_info else None,
}
)
def load_landing_from_condition_report(self, db_session):
room_info = None
if self.condition_report.master_obj.rooms.landing.is_there_a_landing:
room_info = self.load_room_info_from_condition_report(
db_session,
self.condition_report.master_obj.rooms.landing,
)
return self.get_attribute_and_load(
self.condition_report.master_obj.rooms,
"landing",
Landing,
db_session,
additional_fields={
"room_info_id": room_info.id if room_info else None,
}
)
def load_loft_space_from_condition_report(self, db_session):
return self.get_attribute_and_load(
self.condition_report.master_obj.rooms,
"loft_space",
LoftSpace,
db_session,
)
def load_wash_chamber_from_condition_report(self, db_session):
room_info = None
if self.condition_report.master_obj.rooms.wash_chamber.is_there_a_seperated_wc:
room_info = self.load_room_info_from_condition_report(
db_session,
self.condition_report.master_obj.rooms.wash_chamber,
)
return self.get_attribute_and_load(
self.condition_report.master_obj.rooms,
"wash_chamber",
WC,
db_session,
additional_fields={
"room_info_id": room_info.id if room_info else None,
}
)
def load_utility_room_from_condition_report(self, db_session):
room_info = None
if self.condition_report.master_obj.rooms.utility.is_there_a_utility_room:
room_info = self.load_room_info_from_condition_report(
db_session,
self.condition_report.master_obj.rooms.utility
)
return self.get_attribute_and_load(
self.condition_report.master_obj.rooms,
"utility",
Utility,
db_session,
additional_fields={
"room_info_id": room_info.id if room_info else None,
}
)
def load_kitchen_from_condition_report(self, db_session):
room_info = self.load_room_info_from_condition_report(
db_session,
self.condition_report.master_obj.rooms.kitchen,
)
return self.get_attribute_and_load(
self.condition_report.master_obj.rooms,
"kitchen",
pydanticModel=Kitchen,
db_session=db_session,
additional_fields={
"room_info_id": room_info.id if room_info else None,
}
)
def load_dining_room_from_condition_report(self, db_session):
room_info = None
if self.condition_report.master_obj.rooms.dining_room.is_there_a_dining_room:
room_info = self.load_room_info_from_condition_report(
db_session,
self.condition_report.master_obj.rooms.dining_room
)
return self.get_attribute_and_load(
self.condition_report.master_obj.rooms,
"dining_room",
pydanticModel=DiningRoom,
db_session=db_session,
additional_fields={
"room_info_id": room_info.id if room_info else None,
}
)
def load_living_room_from_condition_report(self, db_session):
room_info = self.load_room_info_from_condition_report(
db_session,
self.condition_report.master_obj.rooms.living_room,
)
return self.get_attribute_and_load(
self.condition_report.master_obj.rooms,
"living_room",
pydanticModel=LivingRoom,
db_session=db_session,
additional_fields={
"room_info_id": room_info.id if room_info else None,
}
)
def load_hallway_from_condition_report(self, db_session):
room_info = None
if self.condition_report.master_obj.rooms.hallway.is_there_a_hallway:
room_info = self.load_room_info_from_condition_report(
db_session,
self.condition_report.master_obj.rooms.hallway
)
return self.get_attribute_and_load(
self.condition_report.master_obj.rooms,
"hallway",
pydanticModel=Hallway,
db_session=db_session,
additional_fields={
"is_there_a_hallway": self.condition_report.master_obj.rooms.hallway.is_there_a_hallway,
"room_info_id": room_info.id if room_info else None,
}
)
def load_room_info_from_condition_report(self, db_session, obj):
room_info = obj.room_info
ventilation_info = room_info.ventilation_info
ventilation = self.get_attribute_and_load(
room_info,
"ventilation_info",
VentilationInfo,
db_session,
additional_fields={
"any_damp_mould_or_excessive_condensation_within_the_room": ventilation_info.are_there_any_visible_or_reported_signs_of_damp_mould_or_excessive_condensation_within_the_room
}
)
windows = self.get_attribute_and_load(
room_info,
"windows_info",
WindowsInfo,
db_session,
)
room_info = self.get_attribute_and_load(
obj,
"room_info",
RoomInfo,
db_session,
additional_fields={
"windows_info_id": windows.id,
"ventilation_info_id": ventilation.id,
}
)
return room_info
def load_ventilation_info_from_condition_report(self, db_session, obj):
self.get_attribute_and_load(
obj
)
def load_access_and_elevations_from_condition_report(self, db_session):
pa = self.get_attribute_and_load(
self.condition_report.master_obj.access_and_elevations,
"property_access",
PropertyAccess,
db_session=db_session,
additional_fields={
"more_than_1_5_meters_in_width_to_fence_or__along_the_full_gable_elevation": self.condition_report.master_obj.access_and_elevations.property_access.is_there_more_than_1_point_5_meters_in_width_to_fence_or_neighbouring_boundary_along_the_full_gable_elevation
}
)
print(f"hello junte {pa.id}")
# Front
obj = self.condition_report.master_obj.access_and_elevations.external_elevation_front
external_elevation = self.get_attribute_and_load(
obj,
"external_elevation",
ExternalElevation,
db_session=db_session,
additional_fields={
"any_signs_of_water_penetration_caused_by_failed_rainwater_goods_or_pipework": obj.external_elevation.are_there_any_signs_of_water_penetration_caused_by_failed_rainwater_goods_or_pipework,
},
)
external_elevation_front = self.upsert_record(
db_session=db_session,
model_class=ExternalElevationFront,
data_dict={
"external_elevation_id": external_elevation.id
},
lookup_field=None,
)
def external_elevation_load_for_side(obj, pydanticModel, different="do_all_answers_for_the_front_elevation_apply_to_this_wall"):
external_elevation = None
data_dict={
different: getattr(obj, different),
}
if getattr(obj, different) is False:
if getattr(obj, "external_elevation") is not None:
external_elevation = self.get_attribute_and_load(
obj,
"external_elevation",
ExternalElevation,
db_session=db_session,
additional_fields={
"any_signs_of_water_penetration_caused_by_failed_rainwater_goods_or_pipework": obj.external_elevation.are_there_any_signs_of_water_penetration_caused_by_failed_rainwater_goods_or_pipework,
},
)
data_dict.update({"external_elevation_id": external_elevation.id})
return self.upsert_record(
db_session=db_session,
model_class=pydanticModel,
data_dict=data_dict,
)
# Back
back = external_elevation_load_for_side(self.condition_report.master_obj.access_and_elevations.external_elevation_back, ExternalElevationRear)
# One
gable_one = external_elevation_load_for_side(self.condition_report.master_obj.access_and_elevations.external_elevation_gable_one, ExternalElevationGableOne)
# Two
gable_two = external_elevation_load_for_side(self.condition_report.master_obj.access_and_elevations.external_elevation_gable_two, ExternalElevationGableTwo, different="is_there_a_fourth_external_elevation")
# CO
co = self.get_attribute_and_load(
self.condition_report.master_obj.access_and_elevations,
"conservatory_or_out_building",
ConservatoryOrOutbuilding,
db_session=db_session,
)
return self.upsert_record(
db_session=db_session,
model_class=AccessAndElevations,
data_dict={
"property_access_id": pa.id,
"external_elevation_front_id": external_elevation_front.id,
"external_elevation_back_id": back.id,
"external_elevation_gable_one_id": gable_one.id,
"external_elevation_gable_two_id": gable_two.id,
"conservatory_or_out_building_id": co.id,
}
)
def load_general_information_from_condition_report(self, db_session):
assessors_data = self.condition_report.master_obj.general_information.assessor_details.model_dump()
assessor_details = self.upsert_record(
db_session=db_session,
model_class=AssessorDetails,
data_dict=assessors_data,
lookup_field="elmhurst_id",
)
inspection_data = self.condition_report.master_obj.general_information.inspection_and_project.model_dump()
inspection_and_project = self.upsert_record(
db_session=db_session,
model_class=InspectionAndProject,
data_dict=inspection_data,
lookup_field=None
)
property_data = self.condition_report.master_obj.general_information.the_property.model_dump()
the_property = self.upsert_record(
db_session=db_session,
model_class=TheProperty,
data_dict=property_data,
lookup_field=None
)
# Main elevation
def upload_elevation_and_return_elevation(data):
return self.upsert_record(
db_session=db_session,
model_class=ElevationInfo,
data_dict=data,
lookup_field=None
)
elevation_data = self.condition_report.master_obj.general_information.main_elevation.elevation_info.model_dump()
elevation_info = upload_elevation_and_return_elevation(elevation_data)
main_elevation = self.upsert_record(
db_session=db_session,
model_class=MainElevation,
data_dict={
"elevation_info_id": elevation_info.id
},
lookup_field=None
)
# elevations multiple, elevation definition first then elevationInfo
elevations_data = self.condition_report.master_obj.general_information.elevations.model_dump()
elevations = self.upsert_record(
db_session=db_session,
model_class=Elevation,
data_dict={
"protected_conservatory_or_aonb": elevations_data.get("protected_conservatory_or_aonb"),
"material_type": elevations_data.get("material_type"),
"visible_signs_of_existing_wall_insulation": elevations_data.get("are_there_any_visible_signs_of_existing_wall_insulation"),
"ground_level_bridge_the_dpc": elevations_data.get("does_the_existing_ground_level_on_any_elevation_bridge_the_dpc"),
},
lookup_field=None,
)
for i, elevation in enumerate(self.condition_report.master_obj.general_information.elevations.info):
data = elevation.model_dump()
data.update({"elevation_id": elevations.id})
elevation_info = upload_elevation_and_return_elevation(data)
return self.upsert_record(
db_session=db_session,
model_class=GeneralInformation,
data_dict={
"assessor_detail_id": assessor_details.id,
"inspection_and_project_id": inspection_and_project.id,
"the_property_id": the_property.id,
"main_elevation_id": main_elevation.id,
"elevations_id": elevations.id,
},
lookup_field=None,
)
def load_pre_site_notes_summary_table(self, db_session):
summary_data = self.pre_site_note.survey_information.model_dump()
@ -47,31 +707,20 @@ class surveyedDataProcessor():
lookup_field="reference_number"
)
def create_building_table(self, db_session):
return self.upsert_record(
db_session=db_session,
model_class=Buildings,
data_dict={
"address":"foo",
"potcode": "foobar",
"UPRN": self.pre_site_note.survey_information.uprn,
"landlord_id": "landlord_id",
"domna_id": "landlord_id",
},
lookup_field="UPRN",
)
def get_attribute_and_load(self, obj, attr_string, pydanticModel, db_session):
def get_attribute_and_load(self, obj, attr_string, pydanticModel, db_session, lookup_field=None, additional_fields={}, exclude_list=True):
found = getattr(obj, attr_string, None)
if found:
print(f"Uploading to data base {found}")
print(f"Uploaded to database with this dict {found.model_dump()}")
print(f"Pydantic model {pydanticModel}")
if found.model_dump():
db = self.upsert_record(
db_session=db_session,
model_class=pydanticModel,
data_dict=found.model_dump(),
lookup_field=None
lookup_field=lookup_field,
additional_fields=additional_fields,
exclude_list=exclude_list,
)
return db
return None
@ -103,7 +752,7 @@ class surveyedDataProcessor():
if data:
mainheating2 = self.upsert_record(
db_session=db_session,
model_class=Heating,
model_class=HeatingFromPreSiteNotes,
data_dict=data,
lookup_field=None,
additional_fields= {"controls_id": mainheating2controls.id},
@ -120,7 +769,7 @@ class surveyedDataProcessor():
if data:
mainheating = self.upsert_record(
db_session=db_session,
model_class=Heating,
model_class=HeatingFromPreSiteNotes,
data_dict=data,
lookup_field=None,
additional_fields={"controls_id": mainheatingcontrols.id}
@ -398,11 +1047,18 @@ class surveyedDataProcessor():
db_session,
model_class,
data_dict,
lookup_field,
lookup_field=None,
update_if_exists: bool = False,
additional_fields: dict = None
additional_fields: dict = None,
exclude_list=True
):
clean_data = data_dict
def remove_nested_dicts_and_lists(data: dict):
if exclude_list:
return {k: v for k, v in data.items() if not isinstance(v, (dict, list))}
else:
return {k: v for k, v in data.items() if not isinstance(v, (dict))}
clean_data = remove_nested_dicts_and_lists(data_dict)
# Merge additional fields if provided
if additional_fields:
@ -428,6 +1084,7 @@ class surveyedDataProcessor():
# Filter out invalid fields that don't exist in the model class
valid_fields = [field for field in clean_data if hasattr(model_class, field)]
print(f"valid fields {valid_fields}")
clean_data = {field: clean_data[field] for field in valid_fields}
print(f'clean data is {clean_data}')
@ -441,9 +1098,6 @@ class surveyedDataProcessor():
return new_record
def load_assessor_table(self, db_session):
company = self.load_company_table(db_session)
assessor_data = self.pre_site_note.assessor_information.model_dump()

28
etl/transform/conditionReportTypes.py
View file

@ -51,7 +51,6 @@ class PropertyAccess(BaseModel):
is_there_restricted_space_for_contractors_to_access_the_roof_area: bool
is_there_more_than_1_point_5_meters_in_width_to_fence_or_neighbouring_boundary_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_ginnel: bool
is_access_to_the_rear_provided_by_use_of_a_secured_alleyway: bool
class ExternalElevation(BaseModel):
@ -154,17 +153,17 @@ class RoomInRoof(BaseModel):
room_info: Optional[RoomInfo]
class Rooms(BaseModel):
hallway: Hallway
living_room: LivingRoom
dining_room: DiningRoom
kitchen: Kitchen
utility: Utility
wash_chamber: WC
landing: Landing
bedrooms: List[Bedroom]
bathrooms: List[Bathroom]
loft_space: LoftSpace
room_in_roof: RoomInRoof
hallway: Hallway #loaded
living_room: LivingRoom # loaded
dining_room: DiningRoom # loaded
kitchen: Kitchen # loaded
utility: Utility # loaded
wash_chamber: WC # loaded
landing: Landing # loadded
loft_space: LoftSpace #loaded
room_in_roof: RoomInRoof # loaded
bedrooms: List[Bedroom] # loaded
bathrooms: List[Bathroom] # loaded
class GeneralConditionHeatingSystem(BaseModel):
is_the_heating_system_in_working_order: bool
@ -250,7 +249,8 @@ class ShowerAndBath(BaseModel):
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 Heating(BaseModel):
class HeatingFromConditionReport(BaseModel):
# TODO find one with an example of this one
room_stat_in_temperature_in_celsius: Optional[str] = None
room_stat_location: Optional[str] = None
@ -264,7 +264,7 @@ class EnergyUse(BaseModel):
class OccupantAssessment(BaseModel):
occupant: Occupant
energy_use: EnergyUse
heating: Heating
heating: HeatingFromConditionReport
shower_and_bath: ShowerAndBath
appliances: Optional[Appliances]
fridge_and_freezers: FridgeAndFreezers

6
etl/transform/preSiteNoteTypes.py
View file

@ -101,7 +101,7 @@ class HeatingSystemControls(BaseModel):
electricity_meter_type: Optional[str] = ""
mains_gas_available: Optional[bool] = False
class Heating(BaseModel):
class HeatingFromPreSiteNotes(BaseModel):
type: str
heating_source: str
efficiency_source: str
@ -198,8 +198,8 @@ class PropertyDescription(BaseModel):
photovoltaicPanel: Optional[PhotovoltaicPanel]
windTurbine: Optional[WindTurbine]
otherDetails: Optional[OtherDetails]
mainHeating: Optional[Heating]
mainHeating2: Optional[Heating]
mainHeating: Optional[HeatingFromPreSiteNotes]
mainHeating2: Optional[HeatingFromPreSiteNotes]
secondaryHeatingType: Optional[HeatingType]
# class PropertyReport():

55
etl/utils/utils.py
View file

@ -1,23 +1,40 @@
from urllib.parse import unquote
class SharePointURLError(ValueError):
"""Custom error for SharePoint URL parsing issues."""
pass
def get_sharepoint_path(url):
url_parts = url.split('/')
# Find the index of 'Forms'
forms_index = url_parts.index('Forms')
# Get the part after 'Forms'
after_forms = url_parts[forms_index + 1]
# Find 'id=' and extract after it
if 'id=' in after_forms:
id_part = after_forms.split('id=')[1]
# Only keep the path before '&' (to ignore other parameters)
id_path = id_part.split('&')[0]
# Decode the path
decoded_path = unquote(id_path)
# Now, remove the leading '/sites/xxx/Shared Documents/' part
parts = decoded_path.split('Shared Documents')
if len(parts) > 1:
final_path = parts[1].strip('/')
return final_path
try:
url_parts = url.split('/')
# Find the index of 'Forms'
forms_index = url_parts.index('Forms')
# Get the part after 'Forms'
after_forms = url_parts[forms_index + 1]
# Find 'id=' and extract after it
if 'id=' in after_forms:
id_part = after_forms.split('id=')[1]
# Only keep the path before '&' (to ignore other parameters)
id_path = id_part.split('&')[0]
# Decode the path
decoded_path = unquote(id_path)
# Remove the leading '/sites/xxx/Shared Documents/' part
parts = decoded_path.split('Shared Documents')
if len(parts) > 1:
final_path = parts[1].strip('/')
return final_path
else:
return decoded_path.strip('/')
else:
return decoded_path.strip('/')
raise SharePointURLError(f"The URL does not contain 'id=' parameter. URL: {url}")
except (IndexError, ValueError) as e:
raise SharePointURLError(f"Error with SharePoint URL, please check {url}. Reason: {e}")

4
etl/validator/validator.py
View file

@ -2,8 +2,8 @@ import os
import logging
from etl.utils.logger import Logger
import re
from etl.pdfReader.pdfReaderToText import pdfReaderToText
from etl.pdfReader.reportType import ReportType
from etl.fileReader.pdfReaderToText import pdfReaderToText
from etl.fileReader.reportType import ReportType
class DomnaSharePointValidator():

13
migration_db.sh
View file

@ -1,3 +1,14 @@
#poetry run alembic revision --autogenerate -m "Initial table"
#poetry run alembic revision --autogenerate -m "add utility"
poetry run alembic upgrade head
# See which hash I'm at
#poetry run alembic current
# downgrade one version
#poetry run alembic upgrade head
#poetry run alembic downgrade -1