mirror of
https://github.com/Hestia-Homes/survey-extraction.git
synced 2026-07-12 13:29:08 +00:00
326 lines
13 KiB
Python
326 lines
13 KiB
Python
import hubspot
|
|
from enum import Enum
|
|
from hubspot.crm.deals import PublicObjectSearchRequest
|
|
from hubspot.crm.deals.models import SimplePublicObjectInput
|
|
from etl.hubSpotClient.types import SubmissionInfoFromDeal
|
|
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):
|
|
SURVEYED_COMPLETE_NEEDS_SIGN_OFF = "1617223914"
|
|
SURVEYED_NO_ACCESS_NEED_SIGN_OFF = "1617223915"
|
|
CUSTOMER_CONTACTED = "888730834"
|
|
SURVEYED_COMPLETED_SIGNED_OFF = "1617223916"
|
|
FILES_MISSING_FROM_ASSESSOR = "1887736000"
|
|
|
|
class HubSpotClient():
|
|
def __init__(self):
|
|
self.access_token = "pat-eu1-064f7f5c-a7d8-4d93-a9b2-b604da6164a6"
|
|
self.client = hubspot.Client.create(access_token=self.access_token)
|
|
self.logger = Logger(name='HubSpotClient', level=logging.INFO).get_logger()
|
|
|
|
def get_all_deals(self):
|
|
return self.client.crm.deals.get_all()
|
|
|
|
def get_owner_name_from_id(self, owner_id):
|
|
owner = self.client.crm.owners.owners_api.get_by_id(owner_id)
|
|
time.sleep(1)
|
|
first_name = owner.first_name or ""
|
|
last_name = owner.last_name or ""
|
|
return f"{first_name} {last_name}".strip()
|
|
|
|
def get_deal_name_by_id(self, deal_id):
|
|
try:
|
|
deal = self.client.crm.deals.basic_api.get_by_id(deal_id)
|
|
time.sleep(1)
|
|
return deal.properties.get("dealname", "No deal name")
|
|
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 = []
|
|
after = None
|
|
while True:
|
|
# Correct filter for notes associated with the given deal ID
|
|
search_request = PublicObjectSearchRequest(
|
|
filter_groups=[{
|
|
"filters": [{
|
|
"propertyName": "associations.deal", # Filter by association to the deal
|
|
"operator": "EQ",
|
|
"value": deals_id,
|
|
}]
|
|
}],
|
|
properties=["domna_property_id", "owner_property_id", 'national_uprn'], # Properties of the note you need
|
|
limit=200,
|
|
after=after,
|
|
)
|
|
# Call the search API
|
|
response = self.client.crm.objects.search_api.do_search(object_type="0-420", public_object_search_request=search_request)
|
|
time.sleep(1)
|
|
|
|
# Add the results to the found_notes list
|
|
found_notes.extend(response.results)
|
|
|
|
# Handle pagination if more results are available
|
|
if not response.paging or not response.paging.next:
|
|
break
|
|
after = response.paging.next.after
|
|
|
|
if found_notes:
|
|
return found_notes[0]
|
|
return None
|
|
|
|
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.get('national_uprn', '') or ''
|
|
|
|
def get_notes_from_deals_id(self, deals_id):
|
|
from hubspot.crm.objects import PublicObjectSearchRequest
|
|
found_notes = []
|
|
after = None
|
|
while True:
|
|
# Correct filter for notes associated with the given deal ID
|
|
search_request = PublicObjectSearchRequest(
|
|
filter_groups=[{
|
|
"filters": [{
|
|
"propertyName": "associations.deal", # Filter by association to the deal
|
|
"operator": "EQ",
|
|
"value": deals_id,
|
|
}]
|
|
}],
|
|
properties=["hs_note_body", "hubspot_owner_id"], # Properties of the note you need
|
|
limit=200,
|
|
after=after,
|
|
)
|
|
# Call the search API
|
|
response = self.client.crm.objects.search_api.do_search(object_type="notes", public_object_search_request=search_request)
|
|
time.sleep(1)
|
|
|
|
# Add the results to the found_notes list
|
|
found_notes.extend(response.results)
|
|
|
|
# Handle pagination if more results are available
|
|
if not response.paging or not response.paging.next:
|
|
break
|
|
after = response.paging.next.after
|
|
|
|
all_notes = []
|
|
for note in found_notes:
|
|
# Extract note content and author information
|
|
note_body = note.properties.get("hs_note_body", "No content")
|
|
|
|
# Collect note details in a dictionary
|
|
all_notes.append({
|
|
"note_id": note.id,
|
|
"note": note_body,
|
|
"created_at": note.created_at.strftime("%Y-%m-%d %H:%M:%S"),
|
|
})
|
|
return all_notes
|
|
|
|
|
|
def get_all_deals_from_stage_id(self, stage_id):
|
|
found_deals = []
|
|
after = None
|
|
while True:
|
|
search_request = PublicObjectSearchRequest(
|
|
filter_groups=[{
|
|
"filters": [{
|
|
"propertyName": "dealstage",
|
|
"operator": "EQ",
|
|
"value": stage_id,
|
|
}]
|
|
}],
|
|
properties=[
|
|
"dealname",
|
|
"amount",
|
|
"hubspot_owner_id",
|
|
],
|
|
limit=200,
|
|
after=after,
|
|
)
|
|
response = self.client.crm.deals.search_api.do_search(search_request)
|
|
time.sleep(1)
|
|
found_deals.extend(response.results)
|
|
if not response.paging or not response.paging.next:
|
|
break
|
|
after = response.paging.next.after
|
|
|
|
all_deals = []
|
|
for deal in found_deals:
|
|
all_deals.append({
|
|
"deal_id": deal.id,
|
|
"value": deal.properties["amount"],
|
|
"deal_owner": deal.properties.get("hubspot_owner_id"),
|
|
})
|
|
return all_deals
|
|
|
|
def get_associations_for_deal(self, deal_id, to_object_type):
|
|
"""
|
|
Returns a list of associated object IDs of type `to_object_type`
|
|
(e.g. "contacts", "companies", "notes", etc.)
|
|
"""
|
|
assoc_resp = self.client.crm.deals.associations_api.get_all(
|
|
deal_id=deal_id,
|
|
to_object_type=to_object_type
|
|
)
|
|
return [assoc.id for assoc in assoc_resp.results]
|
|
|
|
def get_deals_from_deal_stage(self, deal_stage: DealStage):
|
|
found_deals = []
|
|
after = None
|
|
while True:
|
|
search_request = PublicObjectSearchRequest(
|
|
filter_groups=[{
|
|
"filters": [{
|
|
"propertyName": "dealstage",
|
|
"operator": "EQ",
|
|
"value": deal_stage.value,
|
|
}]
|
|
}],
|
|
properties=[
|
|
"dealname",
|
|
"number_of_wet_rooms_needing_ventilation",
|
|
"work_type",
|
|
"property_needs_trickle_vents",
|
|
"domna_survey_post_sap",
|
|
"existing_wall_insulation",
|
|
"installer",
|
|
"submission_folder",
|
|
],
|
|
limit=200,
|
|
after=after,
|
|
)
|
|
response = self.client.crm.deals.search_api.do_search(search_request)
|
|
found_deals.extend(response.results)
|
|
if not response.paging or not response.paging.next:
|
|
break
|
|
after = response.paging.next.after
|
|
|
|
all_deals = []
|
|
for i,deal in enumerate(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"],
|
|
work_type=deal.properties["work_type"],
|
|
needs_trickle_ventilation=True if deal.properties.get("property_needs_trickle_vents", "NO").upper() == "YES" else False,
|
|
post_sap_score=int(deal.properties["domna_survey_post_sap"]),
|
|
existing_wall_insulation=deal.properties.get("existing_wall_insulation") if deal.properties.get("existing_wall_insulation") else "None",
|
|
no_of_wet_rooms=int(deal.properties["number_of_wet_rooms_needing_ventilation"]),
|
|
installer=deal.properties["installer"],
|
|
submission_folder_path = deal.properties["submission_folder"],
|
|
landlord_id = landlord_id,
|
|
domna_id = domna_id,
|
|
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" - <b>Message:</b> {msg}<br>"
|
|
f" - <b>Type:</b> {error_type}<br>"
|
|
f" - <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']
|
|
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.FILES_MISSING_FROM_ASSESSOR.value)
|
|
return all_deals
|
|
|
|
def print_all_pipeline_ids(self):
|
|
pipelines = self.client.crm.pipelines.pipelines_api.get_all(object_type="deals")
|
|
for pipeline in pipelines.results:
|
|
print(f"Pipeline: {pipeline.label}")
|
|
for stage in pipeline.stages:
|
|
print(f" - Label: {stage.label}")
|
|
print(f" ID: {stage.id}")
|
|
|
|
def move_deals_to_different_stage(self, list_of_deals_id, to_stage_id):
|
|
deal_properties = SimplePublicObjectInput(
|
|
properties={
|
|
"dealstage": to_stage_id
|
|
}
|
|
)
|
|
for deal_id in list_of_deals_id:
|
|
self.client.crm.deals.basic_api.update(
|
|
deal_id,
|
|
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)
|