diff --git a/etl/find_my_epc/RetrieveFindMyEpc.py b/etl/find_my_epc/RetrieveFindMyEpc.py index d5a5134f..ac0e8235 100644 --- a/etl/find_my_epc/RetrieveFindMyEpc.py +++ b/etl/find_my_epc/RetrieveFindMyEpc.py @@ -126,6 +126,7 @@ class RetrieveFindMyEpc: # Find all h3 headers for each step and extract their related information step_headers = recommendations_div.find_all('h3', class_='govuk-heading-m') previous_sap_score = current_sap + previous_epc = current_rating.split(' ')[-6] for step_num, step_header in enumerate(step_headers, start=1): # Extract the step title (the measure) measure_title = step_header.text.strip().replace(f"Step {step_num}: ", "") @@ -138,7 +139,11 @@ class RetrieveFindMyEpc: # Check if the potential rating div is found if potential_rating_div: # Extract the rating text within the SVG text element - rating_text = potential_rating_div.find('text', class_='govuk-!-font-weight-bold').text.strip() + extracted_rating_text = potential_rating_div.find('text', class_='govuk-!-font-weight-bold') + if extracted_rating_text is not None: + rating_text = extracted_rating_text.text.strip() + else: + rating_text = " ".join([str(previous_sap_score), previous_epc]) # Parse the rating text to separate the numeric rating and EPC letter new_rating = int(rating_text.split()[0]) new_epc = rating_text.split()[1] @@ -152,6 +157,7 @@ class RetrieveFindMyEpc: "sap_points": new_rating - previous_sap_score }) previous_sap_score = new_rating + previous_epc = new_epc # Search for the assessment informaton assessment_information = address_res.find('div', {'id': 'information'}) @@ -270,6 +276,19 @@ class RetrieveFindMyEpc: "Solar photovoltaic (PV) panels": ["solar_pv"], "Party wall insulation": ["party_wall_insulation"], 'Draught proofing': ["draught_proofing"], + "Roof insulation recommendation": [], + "Cavity wall insulation recommendation": [], + "Windows draught proofing": [], + "Low energy lighting for all fixed outlets": ["low_energy_lighting"], + "Cylinder thermostat recommendation": [], + "Heating controls recommendation": [], + "Replace boiler with Band A condensing boiler": [], + "Solar panel recommendation": [], + "Double glazing recommendation": [], + "Solid wall insulation recommendation": [], + "Fuel change recommendation": [], + "PV Cells recommendation": [], + "Replacement glazing units": ["double_glazing"], } survey = True diff --git a/etl/route_march_data_pull/app.py b/etl/route_march_data_pull/app.py index 1e478b0c..80caefc9 100644 --- a/etl/route_march_data_pull/app.py +++ b/etl/route_march_data_pull/app.py @@ -1,5 +1,6 @@ import os import time +from idlelib.iomenu import errors import pandas as pd import numpy as np @@ -21,6 +22,8 @@ EPC_AUTH_TOKEN = os.getenv("EPC_AUTH_TOKEN") def get_data(asset_list, fulladdress_column, address1_column, postcode_column): + home = asset_list[asset_list["row_id"].isin(errors)].head(1).tail(1).squeeze() + epc_data = [] errors = [] no_epc = [] @@ -56,10 +59,21 @@ def get_data(asset_list, fulladdress_column, address1_column, postcode_column): property_recommendations = {"rows": []} # Retrieve data from FindMyEPC - find_epc_searcher = RetrieveFindMyEpc( - address=searcher.newest_epc["address"], postcode=searcher.newest_epc["postcode"] - ) - find_epc_data = find_epc_searcher.retrieve_newest_find_my_epc_data() + try: + find_epc_searcher = RetrieveFindMyEpc( + address=searcher.newest_epc["address"], postcode=searcher.newest_epc["postcode"] + ) + find_epc_data = find_epc_searcher.retrieve_newest_find_my_epc_data() + except ValueError as e: + if "No EPC found" in str(e): + find_epc_searcher = RetrieveFindMyEpc( + address=searcher.newest_epc["address1"], postcode=searcher.newest_epc["postcode"] + ) + find_epc_data = find_epc_searcher.retrieve_newest_find_my_epc_data() + else: + find_epc_data = {} + except Exception as e: + raise Exception(f"Error retrieving FindMyEPC data: {e}") time.sleep(np.random.uniform(0.1, 1)) epc = {