mirror of
https://github.com/Hestia-Homes/Model.git
synced 2026-06-30 13:10:47 +00:00
289 lines
9.4 KiB
Python
289 lines
9.4 KiB
Python
import pandas as pd
|
|
from tqdm import tqdm
|
|
from backend.SearchEpc import SearchEpc
|
|
import numpy as np
|
|
|
|
contact_list = pd.read_csv(
|
|
"/Users/khalimconn-kowlessar/Documents/hestia/Customers/Bromford/Solar Programme Hubspot Upload/Bromford - Solar "
|
|
"PV address list - second wave KLD - PP.csv"
|
|
)
|
|
contact_list["house_no"] = contact_list.apply(lambda x: SearchEpc.get_house_number(
|
|
address=str(x["Address 1: Street 1"]).strip(),
|
|
postcode=str(x["Postal Code"]).strip(),
|
|
), axis=1)
|
|
|
|
asset_list = pd.read_excel(
|
|
"/Users/khalimconn-kowlessar/Documents/hestia/Customers/Bromford/Solar Programme Hubspot Upload/asset_list - "
|
|
"Standardised (1).xlsx",
|
|
sheet_name="Standardised Asset List"
|
|
)
|
|
|
|
lookup = []
|
|
missed = []
|
|
for _, x in tqdm(contact_list.iterrows(), total=len(contact_list)):
|
|
|
|
if x["Address 1: Street 1"] == '1 The Beck':
|
|
lookup.append(
|
|
{
|
|
"UPRN": x["UPRN"],
|
|
"landlord_property_id": 40692,
|
|
}
|
|
)
|
|
continue
|
|
|
|
if x["Address 1: Street 1"] == '3 The Beck ':
|
|
lookup.append(
|
|
{
|
|
"UPRN": x["UPRN"],
|
|
"landlord_property_id": 40693,
|
|
}
|
|
)
|
|
continue
|
|
|
|
if x["Address 1: Street 1"] == '2 Orchard Close ':
|
|
lookup.append(
|
|
{
|
|
"UPRN": x["UPRN"],
|
|
"landlord_property_id": 7924,
|
|
}
|
|
)
|
|
continue
|
|
|
|
if x["Address 1: Street 1"] == '2 Orchard Close ':
|
|
lookup.append(
|
|
{
|
|
"UPRN": x["UPRN"],
|
|
"landlord_property_id": 7924,
|
|
}
|
|
)
|
|
continue
|
|
|
|
if x["Address 1: Street 1"] == '3 Croxall Road':
|
|
lookup.append(
|
|
{
|
|
"UPRN": x["UPRN"],
|
|
"landlord_property_id": 40650,
|
|
}
|
|
)
|
|
continue
|
|
|
|
if x["Address 1: Street 1"] == '4 Ward Road ':
|
|
lookup.append(
|
|
{
|
|
"UPRN": x["UPRN"],
|
|
"landlord_property_id": 33175,
|
|
}
|
|
)
|
|
continue
|
|
|
|
df = asset_list[
|
|
asset_list["domna_full_address"].str.replace(",", "").str.contains(x["Address 1: Street 1"].strip()) &
|
|
asset_list["domna_postcode"].str.contains(x["Postal Code"].strip())
|
|
]
|
|
|
|
if df.shape[0] != 1:
|
|
df = asset_list[
|
|
asset_list["domna_full_address"].str.replace(",", "") == x["Address 1: Street 1"].strip() &
|
|
asset_list["domna_postcode"].str.contains(x["Postal Code"].strip())
|
|
]
|
|
|
|
if df.shape[0] != 1:
|
|
df = asset_list[
|
|
(asset_list["domna_address_1"].astype(str) == str(x["house_no"])) &
|
|
(asset_list["domna_postcode"].str.contains(x["Postal Code"].strip()) == True)
|
|
]
|
|
|
|
if df.shape[0] != 1:
|
|
missed.append(x["UPRN"])
|
|
continue
|
|
|
|
lookup.append(
|
|
{
|
|
"UPRN": x["UPRN"],
|
|
"landlord_property_id": df["landlord_property_id"].values[0],
|
|
}
|
|
)
|
|
|
|
lookup = pd.DataFrame(lookup)
|
|
|
|
contact_list = contact_list.merge(lookup, how="left", on="UPRN")
|
|
# Store
|
|
contact_list.to_csv(
|
|
"/Users/khalimconn-kowlessar/Documents/hestia/Customers/Bromford/Solar Programme Hubspot Upload/Bromford - Solar "
|
|
"PV address list - second wave KLD - PP with landlord_property_id.csv",
|
|
index=False
|
|
)
|
|
|
|
# I manually completed the lookup for the missed ones. We now read it back in and pull in the properties for the
|
|
# stndardised asset list
|
|
contacts_complete = pd.read_csv(
|
|
"/Users/khalimconn-kowlessar/Documents/hestia/Customers/Bromford/Solar Programme Hubspot Upload/Bromford - Solar "
|
|
"PV address list - second wave KLD - PP with landlord_property_id.csv"
|
|
)
|
|
|
|
new_data = pd.read_excel(
|
|
"/Users/khalimconn-kowlessar/Documents/hestia/Customers/Bromford/Solar Programme Hubspot Upload/Master Sheet "
|
|
"Solar PV installs.xlsx",
|
|
sheet_name="Sheet1"
|
|
)
|
|
|
|
contact_list = contact_list.merge(
|
|
new_data,
|
|
how="left",
|
|
left_on="UPRN",
|
|
right_on="CE UPRN"
|
|
)
|
|
route = asset_list[
|
|
asset_list["landlord_property_id"].isin(contact_list["Legacy UPRN"].astype("Int64").astype(str))
|
|
].copy()
|
|
|
|
# Add the new heating data
|
|
contact_list["Legacy UPRN"] = contact_list["Legacy UPRN"].astype("Int64").astype(str)
|
|
route2 = contact_list.merge(
|
|
route,
|
|
how="left",
|
|
right_on="landlord_property_id",
|
|
left_on="Legacy UPRN"
|
|
)
|
|
|
|
# Because I did a data pull, we can fill the other bits of information
|
|
missed = contact_list[~contact_list["Legacy UPRN"].isin(route["landlord_property_id"].astype(int))]
|
|
|
|
# Store both the route and missed
|
|
route2.to_csv(
|
|
"/Users/khalimconn-kowlessar/Documents/hestia/Customers/Bromford/Solar Programme Hubspot Upload/route.csv",
|
|
index=False
|
|
)
|
|
|
|
# Add on phone number
|
|
contact_details_filepath = ("/Users/khalimconn-kowlessar/Documents/hestia/Customers/Bromford/Solar Programme "
|
|
"Hubspot Upload/Hubspot/Bromford - Solar PV address list - second wave KLD - PP with "
|
|
"landlord_property_id.xlsx")
|
|
|
|
contacts_filenames = [
|
|
"/Users/khalimconn-kowlessar/Documents/hestia/Customers/Bromford/Solar Programme Hubspot Upload/contact "
|
|
"details/FAO Paul Contact Details-Table 1.csv",
|
|
"/Users/khalimconn-kowlessar/Documents/hestia/Customers/Bromford/Solar Programme Hubspot Upload/contact "
|
|
"details/Green Contact Details-Table 1.csv",
|
|
"/Users/khalimconn-kowlessar/Documents/hestia/Customers/Bromford/Solar Programme Hubspot Upload/contact "
|
|
"details/Main Contact Details-Table 1.csv",
|
|
]
|
|
|
|
merge_to = pd.read_excel(contact_details_filepath)
|
|
|
|
lookup = []
|
|
for fn in contacts_filenames:
|
|
df = pd.read_csv(fn, encoding="utf-8-sig")
|
|
# Merge on phone
|
|
details = df[
|
|
df["Property Reference Number (Main Address) (Property)"].isin(merge_to["UPRN"].astype(str))
|
|
][[
|
|
"Property Reference Number (Main Address) (Property)", "Landline", "Mobile Phone", "Email Address",
|
|
"First Name", "Last Name"
|
|
]]
|
|
|
|
lookup.append(details)
|
|
|
|
lookup = pd.concat(lookup)
|
|
|
|
# Drop entries where landline, mobile and email are all NaN
|
|
lookup = lookup.dropna(subset=["Landline", "Mobile Phone", "Email Address"], how="all")
|
|
lookup = lookup.drop_duplicates(["Landline", "Mobile Phone", "Email Address"])
|
|
# Sort so email is first, then landline, then mobile
|
|
lookup = lookup.sort_values(
|
|
["Property Reference Number (Main Address) (Property)", "Email Address", "Landline", "Mobile Phone"],
|
|
ascending=[True, True, True, True]
|
|
)
|
|
|
|
# Store
|
|
lookup.to_csv(
|
|
"/Users/khalimconn-kowlessar/Documents/hestia/Customers/Bromford/Solar Programme Hubspot Upload/Hubspot/contact "
|
|
"details.csv",
|
|
index=False
|
|
)
|
|
|
|
lookup2 = []
|
|
for _, x in lookup.groupby("Property Reference Number (Main Address) (Property)"):
|
|
|
|
# We any entries have an email, we take that
|
|
if x["Email Address"].notna().any():
|
|
x = x[x["Email Address"].notna()]
|
|
# We then take the entry with a phone number
|
|
if x["Landline"].notna().any() or x["Mobile Phone"].notna().any():
|
|
x = x[x["Landline"].notna() | x["Mobile Phone"].notna()]
|
|
|
|
# Take the first entry
|
|
x = x.iloc[0]
|
|
lookup2.append(x)
|
|
|
|
lookup2 = pd.DataFrame(lookup2)
|
|
|
|
import pandas as pd
|
|
|
|
# Sample structure based on your columns
|
|
columns = ['Property Reference Number (Main Address) (Property)', 'Landline', 'Mobile Phone', 'Email Address']
|
|
|
|
# Simulating example input DataFrame
|
|
# In practice, you would use: lookup = pd.read_csv(...) or similar
|
|
lookup = pd.DataFrame(columns=columns)
|
|
|
|
# Grouping and transforming
|
|
results = []
|
|
|
|
for prop_id, group in lookup.groupby("Property Reference Number (Main Address) (Property)"):
|
|
# Filter rows with any contact information
|
|
filtered = group[
|
|
group["Email Address"].notna() &
|
|
(group["Landline"].notna() | group["Mobile Phone"].notna())
|
|
]
|
|
|
|
if filtered.empty:
|
|
continue
|
|
|
|
# Sort by presence of phone numbers (prioritize those with both)
|
|
filtered["contact_score"] = (
|
|
filtered["Landline"].notna().astype(int) +
|
|
filtered["Mobile Phone"].notna().astype(int)
|
|
)
|
|
filtered = filtered.sort_values("contact_score", ascending=False)
|
|
|
|
primary = filtered.iloc[0]
|
|
# Make sure secondary is not the same as primary
|
|
if not pd.isnull(primary["Mobile Phone"]):
|
|
secondary = filtered[
|
|
(filtered["Mobile Phone"] != primary["Mobile Phone"])
|
|
]
|
|
elif not pd.isnull(primary["Landline"]):
|
|
secondary = filtered[
|
|
(filtered["Landline"] != primary["Landline"])
|
|
]
|
|
else:
|
|
raise Exception("Look at me")
|
|
|
|
secondary = filtered.iloc[1] if len(filtered) > 1 else None
|
|
|
|
results.append({
|
|
"Property ID": prop_id,
|
|
"Primary Email": primary["Email Address"],
|
|
"Primary Phone": primary["Mobile Phone"] or primary["Landline"],
|
|
"Secondary Email": secondary["Email Address"] if secondary is not None else None,
|
|
"Secondary Phone": secondary["Mobile Phone"] or secondary["Landline"] if secondary is not None else None,
|
|
})
|
|
|
|
final_df = pd.DataFrame(results)
|
|
|
|
import ace_tools as tools;
|
|
|
|
tools.display_dataframe_to_user(name="Cleaned Contact Lookup", dataframe=final_df)
|
|
|
|
# We set up primary and secondary phone numbers. We use mobile as the primary
|
|
|
|
|
|
# We have duplicates, we prioritise entries, by ID, that have a email
|
|
lookup2 = lookup.sort_values("Property Reference Number (Main Address) (Property)").drop_duplicates(
|
|
"Property Reference Number (Main Address) (Property)", keep="last"
|
|
)
|
|
|
|
# TODO: Get into the standardised asset list format
|
|
# TODO: Add the deal postcode to Hubspot
|
|
# TODO: Upload the deal postcode
|