Model/etl/hubspot/hubspot_deal_differ.py

from typing import Dict, List, Optional

from backend.app.db.models.hubspot_deal_data import HubspotDealData
from etl.hubspot.utils import parse_hs_date


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"]

    @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",
            "dealstage": "dealstage",
            "dealname": "dealname",
            "project_code": "project_code",
            "outcome_notes": "outcome_notes",
            "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",
            "block_reference": "block_reference",
            "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": "coordinator",
            "proposed_measures_dropdown": "proposed_measures",
            "approved_package": "approved_package",
            "designer": "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",
        }

        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"),
        ]

        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

        # --- 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 _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") 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("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
        )