mirror of
https://github.com/Hestia-Homes/Model.git
synced 2026-07-12 13:29:04 +00:00
399 lines
15 KiB
Python
399 lines
15 KiB
Python
import os
|
|
from typing import Any, Dict, List, Optional
|
|
|
|
from backend.app.db.models.hubspot_deal_data import HubspotDealData
|
|
from etl.hubspot.project_data import ProjectData
|
|
from etl.hubspot.utils import parse_hs_bool, parse_hs_date, parse_hs_int
|
|
|
|
|
|
class HubspotDealDiffer:
|
|
COORDINATION_COMPLETE: List[str] = [
|
|
"(v1) ioe/mtp complete",
|
|
"(v2) ioe/mtp complete",
|
|
"(v3) ioe/mtp complete",
|
|
]
|
|
RETROFIT_DESIGN_COMPLETE = "uploaded"
|
|
LODGEMENT_COMPLETE: List[str] = ["lodgement complete", "measures lodged"]
|
|
ABANDONMENT_ATTEMPTS_THRESHOLD = 3
|
|
NEGATIVE_OUTCOMES: List[str] = [
|
|
"no answer",
|
|
"cancelled",
|
|
"no show",
|
|
"tenant refusal",
|
|
"not viable",
|
|
]
|
|
# Verified against the portal (2026-07-07): Abri stock-condition deals
|
|
# carry no project association, so the project_code match is the branch
|
|
# that fires in production; the id below is the project object id
|
|
# embedded at the end of that code, kept for when associations appear.
|
|
ABRI_CONDITION_ASSOCIATED_PROJECT_ID = "1247536318670"
|
|
ABRI_CONDITION_PROJECT_CODE = (
|
|
"[Abri Stock Condition - Privately Funded - 062026 - 1247536318670]"
|
|
)
|
|
|
|
@staticmethod
|
|
def check_for_db_update_trigger(
|
|
new_deal: Dict[str, str],
|
|
new_company: Optional[str],
|
|
new_listing: Optional[Dict[str, str]],
|
|
old_deal: HubspotDealData,
|
|
) -> bool:
|
|
"""
|
|
Returns True if ANY difference exists between HubSpot data and DB.
|
|
Returns False if everything matches (i.e. no update needed).
|
|
"""
|
|
|
|
# --- Deal ID ---
|
|
if str(old_deal.deal_id) != str(new_deal.get("hs_object_id")):
|
|
return True
|
|
|
|
# --- Company ---
|
|
if new_company is not None:
|
|
if old_deal.company_id != new_company:
|
|
return True
|
|
|
|
# --- Listing ---
|
|
hs_listing = new_listing or {}
|
|
|
|
if old_deal.listing_id != hs_listing.get("listing_id"):
|
|
return True
|
|
|
|
if old_deal.landlord_property_id != hs_listing.get("owner_property_id"):
|
|
return True
|
|
|
|
if old_deal.uprn != hs_listing.get("national_uprn"):
|
|
return True
|
|
|
|
# --- Field mappings ---
|
|
FIELD_MAP = {
|
|
"outcome": "outcome",
|
|
"number_of_attempts": "number_of_attempts",
|
|
"dealstage": "dealstage",
|
|
"dealname": "dealname",
|
|
"project_code": "project_code",
|
|
"outcome_notes": "outcome_notes",
|
|
"booking_status": "booking_status",
|
|
"major_condition_issue_description": "major_condition_issue_description",
|
|
"major_condition_issue_photos": "major_condition_issue_photos",
|
|
"coordination_status__stage_1_": "coordination_status",
|
|
"coordination_comments": "coordination_comments",
|
|
"retrofit_design_status": "design_status",
|
|
"pashub_link": "pashub_link",
|
|
"sharepoint_link": "sharepoint_link",
|
|
"dampmould_growth": "dampmould_growth",
|
|
"damp_mould_and_repairs_comments": "damp_mould_and_repairs_comments",
|
|
"pre_sap_score_dropdown": "pre_sap",
|
|
"batch": "batch",
|
|
"batch_description": "batch_description",
|
|
"block_reference": "block_reference",
|
|
"nonfunded_measures": "nonfunded_measures",
|
|
"epc_prn": "epc_prn",
|
|
"potential_post_sap_score_dropdown": "potential_post_sap_score_dropdown",
|
|
"ei_score": "ei_score",
|
|
"ei_score__potential_": "ei_score__potential_",
|
|
"epc_sap_score": "epc_sap_score",
|
|
"epc_sap_score__potential_": "epc_sap_score__potential_",
|
|
"coordinator_user": "coordinator",
|
|
"proposed_measures_dropdown": "proposed_measures",
|
|
"approved_package": "approved_package",
|
|
"designer_user": "designer",
|
|
"actual_measures_installed": "actual_measures_installed",
|
|
"installer": "installer",
|
|
"installer_handover": "installer_handover",
|
|
"lodgement_status": "lodgement_status",
|
|
"design_type": "design_type",
|
|
"surveyor": "surveyor",
|
|
"confirmed_survey_time": "confirmed_survey_time",
|
|
"client_booking_reference": "client_booking_reference",
|
|
"survey_type": "survey_type",
|
|
"measures_for_pibi_ordered": "measures_for_pibi_ordered",
|
|
"property_halted_reason": "property_halted_reason",
|
|
"technical_approved_measures_for_install": "technical_approved_measures_for_install",
|
|
}
|
|
|
|
for hs_field, db_field in FIELD_MAP.items():
|
|
old_value = getattr(old_deal, db_field)
|
|
new_value = new_deal.get(hs_field)
|
|
|
|
if old_value != new_value:
|
|
return True
|
|
|
|
# --- Date fields ---
|
|
date_fields = [
|
|
("mtp_completion_date", "mtp_completion_date"),
|
|
("mtp_re_model_completion_date", "mtp_re_model_completion_date"),
|
|
("ioe_v3_completion_date", "ioe_v3_completion_date"),
|
|
("design_completion_date", "design_completion_date"),
|
|
("measures_lodgement_date", "measures_lodgement_date"),
|
|
("lodgement_date", "lodgement_date"),
|
|
("expected_commencement_date", "expected_commencement_date"),
|
|
("confirmed_survey_date", "confirmed_survey_date"),
|
|
("surveyed_date", "surveyed_date"),
|
|
("pibi_order_date", "pibi_order_date"),
|
|
("pibi_completed_date", "pibi_completed_date"),
|
|
("property_halted_date", "property_halted_date"),
|
|
("sent_to_iw_for_pricing", "sent_to_installer_for_pricing"),
|
|
("osmosis_survey_date", "domna_survey_date"),
|
|
("date_booking_made", "date_booking_made"),
|
|
("last_contact_date", "last_contact_date"),
|
|
("last_outbound_call", "last_outbound_call"),
|
|
("last_outbound_email", "last_outbound_email"),
|
|
("last_submission_date", "last_submission_date"),
|
|
]
|
|
|
|
for hs_field, db_field in date_fields:
|
|
old_value = getattr(old_deal, db_field)
|
|
new_value = parse_hs_date(new_deal.get(hs_field))
|
|
|
|
if old_value != new_value:
|
|
return True
|
|
|
|
# --- Boolean fields ---
|
|
bool_fields = [
|
|
("osmosis_survey_required", "domna_survey_required"),
|
|
]
|
|
|
|
for hs_field, db_field in bool_fields:
|
|
old_value = getattr(old_deal, db_field)
|
|
new_value = parse_hs_bool(new_deal.get(hs_field))
|
|
|
|
if old_value != new_value:
|
|
return True
|
|
|
|
# --- Time field ---
|
|
if old_deal.confirmed_survey_time != new_deal.get("confirmed_survey_time"):
|
|
return True
|
|
|
|
# No differences found
|
|
return False
|
|
|
|
@staticmethod
|
|
def check_for_pashub_trigger(
|
|
new_deal: Dict[str, str], old_deal: HubspotDealData
|
|
) -> bool:
|
|
new_pashub_link: str = new_deal.get("pashub_link", "")
|
|
|
|
if not HubspotDealDiffer._has_valid_pashub_link(new_pashub_link):
|
|
return False
|
|
|
|
if HubspotDealDiffer._new_or_updated_pashub_link(new_pashub_link, old_deal):
|
|
return True
|
|
|
|
if HubspotDealDiffer._coordination_completed(new_deal, old_deal):
|
|
return True
|
|
|
|
if HubspotDealDiffer._design_completed(new_deal, old_deal):
|
|
return True
|
|
|
|
if HubspotDealDiffer._lodgement_completed(new_deal, old_deal):
|
|
return True
|
|
|
|
return False
|
|
|
|
@staticmethod
|
|
def check_abri_triggers_and_construct_message(
|
|
hubspot_deal_id: str,
|
|
new_deal: Dict[str, str],
|
|
new_project: Optional[ProjectData],
|
|
new_listing: Optional[Dict[str, str]],
|
|
old_deal: HubspotDealData,
|
|
) -> Optional[Dict[str, Any]]:
|
|
flows: List[str] = []
|
|
if HubspotDealDiffer.check_for_abri_job_amendment(
|
|
new_deal=new_deal, new_project=new_project, old_deal=old_deal
|
|
):
|
|
flows.append("amend_job")
|
|
if HubspotDealDiffer.check_for_abri_job_logging(
|
|
new_deal=new_deal, new_project=new_project, old_deal=old_deal
|
|
):
|
|
flows.append("log_job")
|
|
if HubspotDealDiffer.check_for_abri_tenant_data_fetch(
|
|
new_deal=new_deal, new_project=new_project, old_deal=old_deal
|
|
):
|
|
flows.append("sync_tenant_data")
|
|
if HubspotDealDiffer.check_for_abri_deal_abandonment(
|
|
new_deal=new_deal, new_project=new_project, old_deal=old_deal
|
|
):
|
|
flows.append("abandon_job")
|
|
|
|
if not flows:
|
|
return None
|
|
|
|
confirmed_survey_date = parse_hs_date(new_deal.get("confirmed_survey_date"))
|
|
last_submission_date = parse_hs_date(new_deal.get("last_submission_date"))
|
|
listing = new_listing or {}
|
|
return {
|
|
"hubspot_deal_id": hubspot_deal_id,
|
|
"flows": flows,
|
|
"place_ref": listing.get("owner_property_id"),
|
|
"deal_name": new_deal.get("dealname"),
|
|
"confirmed_survey_date": (
|
|
confirmed_survey_date.date().isoformat()
|
|
if confirmed_survey_date is not None
|
|
else None
|
|
),
|
|
"confirmed_survey_time": new_deal.get("confirmed_survey_time"),
|
|
# Carried for the abandon flow: it dates the cancel to the confirmed
|
|
# survey date, falling back to the last submission date.
|
|
"last_submission_date": (
|
|
last_submission_date.date().isoformat()
|
|
if last_submission_date is not None
|
|
else None
|
|
),
|
|
# Carried for the abandon flow's reason-mapping seam; harmless to
|
|
# the other flows, which ignore it.
|
|
"outcome": new_deal.get("outcome"),
|
|
}
|
|
|
|
@staticmethod
|
|
def check_for_abri_job_logging(
|
|
new_deal: Dict[str, str],
|
|
new_project: Optional[ProjectData],
|
|
old_deal: HubspotDealData,
|
|
) -> bool:
|
|
if not HubspotDealDiffer._is_abri_condition_deal(new_deal, new_project):
|
|
return False
|
|
|
|
if old_deal.confirmed_survey_date is not None:
|
|
return False
|
|
|
|
return parse_hs_date(new_deal.get("confirmed_survey_date")) is not None
|
|
|
|
@staticmethod
|
|
def check_for_abri_job_amendment(
|
|
new_deal: Dict[str, str],
|
|
new_project: Optional[ProjectData],
|
|
old_deal: HubspotDealData,
|
|
) -> bool:
|
|
if not HubspotDealDiffer._is_abri_condition_deal(new_deal, new_project):
|
|
return False
|
|
|
|
# A first-time survey date logs a new job, not an amendment.
|
|
if old_deal.confirmed_survey_date is None:
|
|
return False
|
|
|
|
# TODO: a cleared survey date (was set, now empty) still fires this
|
|
# check, but the amendoptiappt route cannot express a cancellation.
|
|
# Cancellation semantics (the "delete" action / canceljob route) need
|
|
# a deliberate decision before this trigger is wired to a consumer.
|
|
new_survey_date = parse_hs_date(new_deal.get("confirmed_survey_date"))
|
|
if old_deal.confirmed_survey_date != new_survey_date:
|
|
return True
|
|
|
|
return old_deal.confirmed_survey_time != new_deal.get("confirmed_survey_time")
|
|
|
|
@staticmethod
|
|
def check_for_abri_tenant_data_fetch(
|
|
new_deal: Dict[str, str],
|
|
new_project: Optional[ProjectData],
|
|
old_deal: HubspotDealData,
|
|
) -> bool:
|
|
if not HubspotDealDiffer._is_abri_condition_deal(new_deal, new_project):
|
|
return False
|
|
|
|
if old_deal.expected_commencement_date is not None:
|
|
return False
|
|
|
|
return parse_hs_date(new_deal.get("expected_commencement_date")) is not None
|
|
|
|
@staticmethod
|
|
def check_for_abri_deal_abandonment(
|
|
new_deal: Dict[str, str],
|
|
new_project: Optional[ProjectData],
|
|
old_deal: HubspotDealData,
|
|
) -> bool:
|
|
if not HubspotDealDiffer._is_abri_condition_deal(new_deal, new_project):
|
|
return False
|
|
|
|
if HubspotDealDiffer._is_abandoned(
|
|
old_deal.number_of_attempts, old_deal.outcome
|
|
):
|
|
return False
|
|
|
|
return HubspotDealDiffer._is_abandoned(
|
|
new_deal.get("number_of_attempts"), new_deal.get("outcome")
|
|
)
|
|
|
|
@staticmethod
|
|
def check_for_magicplan_trigger(
|
|
new_deal: Dict[str, str], old_deal: HubspotDealData
|
|
) -> bool:
|
|
new_outcome = (new_deal.get("outcome") or "").lower()
|
|
old_outcome = (old_deal.outcome or "").lower()
|
|
return new_outcome == "surveyed" and old_outcome != "surveyed"
|
|
|
|
@staticmethod
|
|
def _is_abandoned(
|
|
number_of_attempts: Optional[str], outcome: Optional[str]
|
|
) -> bool:
|
|
attempts: Optional[int] = parse_hs_int(number_of_attempts)
|
|
if (
|
|
attempts is None
|
|
or attempts < HubspotDealDiffer.ABANDONMENT_ATTEMPTS_THRESHOLD
|
|
):
|
|
return False
|
|
|
|
return (outcome or "").lower() in HubspotDealDiffer.NEGATIVE_OUTCOMES
|
|
|
|
@staticmethod
|
|
def _is_abri_condition_deal(
|
|
new_deal: Dict[str, str], new_project: Optional[ProjectData]
|
|
) -> bool:
|
|
if new_project is not None:
|
|
return (
|
|
new_project["project_id"]
|
|
== HubspotDealDiffer.ABRI_CONDITION_ASSOCIATED_PROJECT_ID
|
|
)
|
|
|
|
# TEST_ABRI_PROJECT_CODE_OVERRIDE lets a non-production deploy fire on a
|
|
# sandbox deal's own project code; unset (the production default) falls
|
|
# back to the real stock-condition code.
|
|
expected_code = (
|
|
os.environ.get("TEST_ABRI_PROJECT_CODE_OVERRIDE")
|
|
or HubspotDealDiffer.ABRI_CONDITION_PROJECT_CODE
|
|
)
|
|
new_project_code = (new_deal.get("project_code") or "").lower()
|
|
return new_project_code == expected_code.lower()
|
|
|
|
@staticmethod
|
|
def _has_valid_pashub_link(new_pashub_link: str) -> bool:
|
|
return bool(new_pashub_link)
|
|
|
|
@staticmethod
|
|
def _new_or_updated_pashub_link(
|
|
new_pashub_link: str, old_deal: HubspotDealData
|
|
) -> bool:
|
|
if not old_deal.pashub_link:
|
|
return True
|
|
return old_deal.pashub_link != new_pashub_link
|
|
|
|
@staticmethod
|
|
def _coordination_completed(
|
|
new_deal: Dict[str, str], old_deal: HubspotDealData
|
|
) -> bool:
|
|
new_status: str = new_deal.get("coordination_status__stage_1_") or ""
|
|
return (
|
|
new_status != ""
|
|
and new_status.lower() in HubspotDealDiffer.COORDINATION_COMPLETE
|
|
and new_status != old_deal.coordination_status
|
|
)
|
|
|
|
@staticmethod
|
|
def _design_completed(new_deal: Dict[str, str], old_deal: HubspotDealData) -> bool:
|
|
new_status: str = new_deal.get("retrofit_design_status") or ""
|
|
return (
|
|
new_status != ""
|
|
and new_status.lower() == HubspotDealDiffer.RETROFIT_DESIGN_COMPLETE
|
|
and new_status != old_deal.design_status
|
|
)
|
|
|
|
@staticmethod
|
|
def _lodgement_completed(
|
|
new_deal: Dict[str, str], old_deal: HubspotDealData
|
|
) -> bool:
|
|
new_status: str = new_deal.get("lodgement_status") or ""
|
|
return (
|
|
new_status != ""
|
|
and new_status.lower() in HubspotDealDiffer.LODGEMENT_COMPLETE
|
|
and new_status != old_deal.lodgement_status
|
|
)
|