survey-extraction/etl/hubSpotClient/hubspotClient.py

import hubspot
from enum import Enum
from etl.utils.logger import Logger
import logging
from hubspot.crm.associations import ApiException
import os
import requests

from hubspot.crm.objects import SimplePublicObjectInput
from hubspot.crm.associations.v4 import AssociationSpec
from hubspot.crm.associations import ApiException


class Companies(Enum):
    ABRI = "237615001799"
    SOUTHERN_HOUSING_GROUP = "109343619305"
    LIVEWEST = "86205872354"
    SURESERVE = "301745289413"
    HOMEGROUP = "94946071794"
    APPLE = "184769046716"
    THE_GUINESS_PARTNERSHIP = "86970043613"


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 Pipeline(Enum):
    OPERATIONS_SOCIAL_HOUSING = "1167582403"


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()
        self.all_deals = None

    def get_all_deals(self):
        self.all_deals = self.client.crm.deals.get_all()
        return self.all_deals

    def get_deal_ids_by_pipeline(self, pipeline_id):
        """
        Get all deal IDs associated with a given pipeline.
        """
        if self.all_deals is None:
            self.get_all_deals()

        # Filter deals where properties['pipeline'] matches the given pipeline_id
        filtered_deals = [
            deal
            for deal in self.all_deals
            if deal.properties["pipeline"] == str(pipeline_id)
        ]

        # Extract and return only the deal IDs
        deal_ids = [deal.id for deal in filtered_deals]

        return deal_ids

    def get_deals_from_company(self, company_id: str) -> list[str]:
        associations_api = self.client.crm.associations.v4.basic_api

        deal_ids = []
        after = None

        while True:
            response = associations_api.get_page(
                object_type="companies",
                object_id=company_id,
                to_object_type="deals",
                limit=100,
                after=after,
            )

            deal_ids.extend(assoc.to_object_id for assoc in response.results)

            if not response.paging or not response.paging.next:
                break

            after = response.paging.next.after

        return deal_ids

    def from_deal_get_associated_listing(self, deal_id: str):
        """
        Get the associated listing information for a given deal.
        Returns a dictionary of listing properties, or None if not found.
        """
        associations_api = self.client.crm.associations.v4.basic_api
        listings_api = (
            self.client.crm.objects.basic_api
        )  # works for custom objects like "listing"

        # Fetch associated listing(s)
        response = associations_api.get_page(
            object_type="deals",
            object_id=deal_id,
            to_object_type="0-420",  # <-- use your exact custom object name slug here
            limit=1,
        )

        if not response.results:
            self.logger.info(f"No listing association found for deal {deal_id}")
            return None

        listing_id = response.results[0].to_object_id
        self.logger.info(f"Associated listing ID for deal {deal_id}: {listing_id}")

        # Fetch listing details (the "listing information")
        listing = listings_api.get_by_id(
            object_type="0-420",  # again, must match your HubSpot object name
            object_id=listing_id,
            properties=[
                "national_uprn",
                "domna_property_id",
                "owner_property_id",
            ],
        )

        listing_info = listing.properties
        self.logger.info(f"Listing info for deal {deal_id}: {listing_info}")
        return listing_info

    def from_deal_get_info(self, deal_id):
        deal = self.client.crm.deals.basic_api.get_by_id(
            deal_id,
            properties=[
                "dealname",
                "dealstage",
                "pipeline",
                "outcome",  # outcome,
                "outcome_notes",  # outcome notes
                "project_code",
                "major_condition_issue_description",
                "major_condition_issue_photos",
                "coordination_status__stage_1_",  # Coordiantion Status (Stage 1),
                "retrofit_design_status",  # Retrofit Design Status
            ],
        )

        return deal.properties
    
    def from_deal_get_associated_company_id(self, deal_id: str):
            """
            Get the associated company ID from a given deal ID.
            Returns the associated company ID, or None if not found.
            """
            try:
                associations_api = self.client.crm.associations.v4.basic_api

                # Fetch associations for this specific deal only
                response = associations_api.get_page(
                    object_type="deals",
                    object_id=deal_id,
                    to_object_type="companies",
                    limit=1  # Expect only one associated company
                )

                if not response.results:
                    self.logger.info(f"No company association found for deal {deal_id}")
                    return None

                company_id = response.results[0].to_object_id
                self.logger.info(f"Associated company ID for deal {deal_id}: {company_id}")
                return company_id

            except ApiException as e:
                self.logger.error(f"Error fetching associated company for deal {deal_id}: {e}")
                return None

    def get_deal_info_for_db(self, deal_id):
        deal = self.from_deal_get_info(deal_id)
        company = self.from_deal_get_associated_company_id(deal_id)
        listing = self.from_deal_get_associated_listing(deal_id)

        return deal, company, listing

    def get_company_information(self, company_id):
        company = self.client.crm.companies.basic_api.get_by_id(
            company_id,
            properties=[
                "name",
            ],
        )
        company_info = company.properties
        return company_info

    def get_all_pipelines(self):
        """
        Retrieve all pipelines for deals, returning a list of dicts with pipeline names and IDs.
        """
        try:
            pipelines_api = self.client.crm.pipelines.pipelines_api
            response = pipelines_api.get_all(object_type="deals")

            pipelines = [
                {"name": pipeline.label, "id": pipeline.id}
                for pipeline in response.results
            ]

            self.logger.info(f"Retrieved {len(pipelines)} pipelines.")
            return pipelines

        except Exception as e:
            self.logger.error(f"Error retrieving pipelines: {e}")
            return []

    def get_deal_stages(self, pipeline_id=None):
        """
        Retrieve all deal stages for a given pipeline.
        If no pipeline_id is provided, retrieves all stages for all pipelines.
        Returns a list of dicts with pipeline name, stage name, and stage ID.
        """
        try:
            pipelines_api = self.client.crm.pipelines.pipelines_api
            response = pipelines_api.get_all(object_type="deals")

            all_stages = []

            for pipeline in response.results:
                # Skip other pipelines if a specific one is requested
                if pipeline_id and pipeline.id != str(pipeline_id):
                    continue

                stages = [
                    {
                        "pipeline_name": pipeline.label,
                        "pipeline_id": pipeline.id,
                        "stage_name": stage.label,
                        "stage_id": stage.id,
                    }
                    for stage in pipeline.stages
                ]

                all_stages.extend(stages)

            if not all_stages:
                self.logger.info(
                    f"No deal stages found for pipeline {pipeline_id if pipeline_id else 'ALL'}"
                )
            else:
                self.logger.info(f"Retrieved {len(all_stages)} deal stages.")

            return all_stages

        except Exception as e:
            self.logger.error(f"Error retrieving deal stages: {e}")
            return []

    def download_file_from_url(self, download_url: str, save_path: str = None) -> str:
        """
        Download a file from a HubSpot file URL (public or private), keeping its original file type.
        Includes retry logic for transient failures and normalization of URL-encoded special characters.
        """
        import mimetypes
        import requests
        import os
        import time
        import re
        from urllib.parse import urlparse, urlunparse, unquote, quote

        # Strip signature and expiration from CDN URLs to get a fresh one with response-content-disposition
        # This is how HubSpot UI generates longer-lived URLs
        if "cdnp1.hubspotusercontent" in download_url:
            from urllib.parse import urlparse, urlunparse
            parsed = urlparse(download_url)
            # Keep only the path, strip all query params to force CDN to generate fresh signature
            download_url = f"{parsed.scheme}://{parsed.netloc}{parsed.path}?response-content-disposition=attachment"
            self.logger.info(
                f"Requesting fresh CDN signature with response-content-disposition: {download_url}"
            )

        # Normalize URL-encoded special characters in the path (not query string)
        # to avoid signature validation issues with special characters like %2c, %20
        parsed = urlparse(download_url)
        clean_path = quote(unquote(parsed.path), safe="/@:")
        download_url = urlunparse(parsed._replace(path=clean_path))

        max_attempts = 3
        retry_delays = [1, 2, 4]  # exponential backoff in seconds
        last_exception = None

        for attempt in range(max_attempts):
            try:
                headers = {}
                # Add auth token for API endpoints (not direct CDN URLs)
                if "hubspot.com/form-integrations" in download_url or "api-eu1.hubspot.com" in download_url:
                    headers["Authorization"] = f"Bearer {self.access_token}"

                self.logger.info(
                    f"Downloading HubSpot file (attempt {attempt + 1}/{max_attempts}): {download_url}"
                )
                response = requests.get(
                    download_url,
                    headers=headers,
                    stream=True,
                    allow_redirects=True,
                    timeout=(10, 30),
                )
                response.raise_for_status()

                # Try to infer filename from Content-Disposition header
                content_disposition = response.headers.get("content-disposition")
                if content_disposition and "filename=" in content_disposition:
                    filename = content_disposition.split("filename=")[1].strip('"')
                else:
                    # fallback: extract from URL or content-type
                    filename = (
                        os.path.basename(download_url.split("?")[0]) or "hubspot_download"
                    )
                    if "." not in filename:
                        content_type = response.headers.get("content-type")
                        ext = (
                            mimetypes.guess_extension(content_type.split(";")[0])
                            if content_type
                            else None
                        )
                        if ext:
                            filename += ext

                # Make sure save_path is valid
                if save_path is None:
                    save_path = os.path.abspath(filename)
                elif os.path.isdir(save_path):
                    save_path = os.path.join(save_path, filename)
                else:
                    # if user passes a file path directly, leave it
                    save_path = os.path.abspath(save_path)

                with open(save_path, "wb") as f:
                    for chunk in response.iter_content(chunk_size=8192):
                        f.write(chunk)

                self.logger.info(f"File downloaded successfully → {save_path}")
                return save_path

            except requests.exceptions.HTTPError as e:
                # Don't retry on 404 — file genuinely doesn't exist
                if e.response is not None and e.response.status_code == 404:
                    self.logger.error(f"Failed to download file from HubSpot: {e}")
                    raise
                last_exception = e
                if attempt < max_attempts - 1:
                    delay = retry_delays[attempt]
                    self.logger.warning(
                        f"HTTP error (attempt {attempt + 1}/{max_attempts}): {e} — retrying in {delay}s"
                    )
                    time.sleep(delay)
            except (
                requests.exceptions.ConnectionError,
                requests.exceptions.Timeout,
            ) as e:
                last_exception = e
                if attempt < max_attempts - 1:
                    delay = retry_delays[attempt]
                    self.logger.warning(
                        f"Connection/timeout error (attempt {attempt + 1}/{max_attempts}): {e} — retrying in {delay}s"
                    )
                    time.sleep(delay)
            except requests.exceptions.RequestException as e:
                # Other request errors (e.g., invalid URL) — don't retry
                self.logger.error(f"Failed to download file from HubSpot: {e}")
                raise

        # If we got here, all retries failed
        self.logger.error(
            f"Failed to download file after {max_attempts} attempts: {last_exception}"
        )
        raise last_exception

    def create_line_item_from_product(self, product_id: str, quantity: int = 1):
        # Fetch product mapping
        product = self.client.crm.products.basic_api.get_by_id(
            product_id, properties=["name", "price", "hs_price"]
        )

        name = product.properties.get("name")
        price = (
            product.properties.get("price") or product.properties.get("hs_price") or "0"
        )

        # Build line item payload
        line_item_input = SimplePublicObjectInput(
            properties={
                "hs_product_id": product_id,
                "name": name,
                "quantity": str(quantity),
                "price": price,
                "amount": str(float(price) * quantity),
                "invoiced": "Outstanding",
            }
        )

        # Create line item
        line_item = self.client.crm.line_items.basic_api.create(line_item_input)
        return line_item.id

    def associate_line_item_to_deal(self, line_item_id: str, deal_id: str):
        self.logger.info(f"Associating line item {line_item_id} → deal {deal_id}")

        association_api = self.client.crm.associations.v4.basic_api

        association_api.create(
            "0-3",  # to object type
            deal_id,  # to object id
            "line_items",  # from object type
            line_item_id,  # from object id
            [
                AssociationSpec(
                    association_category="HUBSPOT_DEFINED",
                    association_type_id=19,  # line_item → deal
                )
            ],
        )

    def add_product_line_item_to_deal(
        self, deal_id: str, product_id: str, quantity: int = 1
    ):
        # Step 1: Create the line item from product mapping
        line_item_id = self.create_line_item_from_product(product_id, quantity)

        # Step 2: Associate the created line item to the deal
        self.associate_line_item_to_deal(line_item_id, deal_id)

        return line_item_id

    def delete_line_item(self, line_item_id: str):
        """
        Delete (archive) a line item in HubSpot by its ID.
        """
        try:
            self.logger.info(f"Deleting line item {line_item_id}...")

            self.client.crm.line_items.basic_api.archive(line_item_id)

            self.logger.info(f"Line item {line_item_id} deleted successfully.")
            return True

        except ApiException as e:
            self.logger.error(f"Failed to delete line item {line_item_id}: {e}")
            return False