junte/Model
1
0
Fork 0
mirror of https://github.com/Hestia-Homes/Model.git synced 2026-06-08 11:17:27 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions

diff checker for pashub trigger 🟪

Browse source
This commit is contained in:
Daniel Roth 2026-04-08 15:59:29 +00:00
parent 9da0cabb0f
commit 2d0bc67731
1 changed files with 72 additions and 61 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

133
backend/hubspot_trigger_orchestrator/hubspot_deal_differ.py
View file

@ -4,67 +4,12 @@ from backend.app.db.models.organisation import HubspotDealData
class HubspotDealDiffer:
@staticmethod
def check_for_pashub_trigger(
new_deal: Dict[str, str], old_deal: HubspotDealData
) -> bool:
"""
Case 1: PasHub Link is updated
Case 2: Coordination is completed (and PasHub Link is populated)
Case 3: Design is completed (and PasHub Link is populated)
Case 4: Lodgement is completed (and PasHub Link is populated)
"""
COORDINATION_COMPLETE: List[str] = [
"v1 ioe/mtp complete",
"v2 ioe/mtp complete",
]
RETROFIT_DESIGN_COMPLETE = "uploaded"
LODGEMENT_COMPLETE: List[str] = ["lodgement complete", "measures lodged"]
new_pashub_link: str = new_deal.get("pashub_link", "")
# Case 1
if not new_pashub_link:
return False
if not old_deal.pashub_link:
return True
if old_deal.pashub_link != new_pashub_link:
return True
# Case 2
new_coordination_status: str = new_deal.get("coordination_status", "")
if (
new_coordination_status
and new_coordination_status in COORDINATION_COMPLETE
and new_coordination_status != old_deal.coordination_status
):
return True
# Case 3
new_design_status: str = new_deal.get("design_status", "")
if (
new_design_status
and new_design_status == RETROFIT_DESIGN_COMPLETE
and new_design_status != old_deal.design_status
):
return True
# Case 4
new_lodgement_status: str = new_deal.get("lodgement_status", "")
if (
new_lodgement_status
and new_lodgement_status in LODGEMENT_COMPLETE
and new_lodgement_status != old_deal.lodgement_status
):
return True
return False
COORDINATION_COMPLETE: List[str] = [
"v1 ioe/mtp complete",
"v2 ioe/mtp complete",
]
RETROFIT_DESIGN_COMPLETE = "uploaded"
LODGEMENT_COMPLETE: List[str] = ["lodgement complete", "measures lodged"]
@staticmethod
def check_for_db_update_trigger(
@ -74,3 +19,69 @@ class HubspotDealDiffer:
old_deal: HubspotDealData,
) -> bool:
raise NotImplementedError
@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 _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", "")
return (
new_status != ""
and new_status 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("design_status", "")
return (
new_status != ""
and new_status == 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", "")
return (
new_status != ""
and new_status in HubspotDealDiffer.LODGEMENT_COMPLETE
and new_status != old_deal.lodgement_status
)