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