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"

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 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 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',
                'outcome', #outcome,
                'outcome_notes', #outcome notes
                'project_code',
                'major_condition_issue_description',
                'major_condition_issue_photos',
            ]
        )

        return deal.properties
    
    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.
        """
        import mimetypes
        import requests
        import os

        try:
            headers = {}
            if "hubspotusercontent" not in download_url:
                headers["Authorization"] = f"Bearer {self.access_token}"

            self.logger.info(f"Downloading HubSpot file: {download_url}")
            response = requests.get(download_url, headers=headers, stream=True, allow_redirects=True)
            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.RequestException as e:
            self.logger.error(f"Failed to download file from HubSpot: {e}")
            raise


    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