Hello! I am trying to work with the API and am having some issues. I admit I am not super technical with APIs in general. I lead our Customer Success Team and just want to export our conversation data so I can organize locally with sheets and AI analysis.
I am trying to make a python script to just have something I can easily run that will generate a CSV. I am open to using other methods but I need something that does not involve a Beta integration or anything like that because I want to have control over where the data goes.
My code is:
import requests
import time
import csv
from datetime import datetime, timedelta
# HubSpot API credentials
access_token = ‘[not shown for security]’
# Define the time range for fetching messages (last 30 days)
end_time = datetime.now()
start_time = end_time - timedelta(days=30)
# Convert to ISO 8601 format
start_time_iso = start_time.isoformat() + ‘Z’
end_time_iso = end_time.isoformat() + ‘Z’
# Define the endpoint for retrieving inbox messages
# Updated to use the correct endpoint for inbox messages
url = ‘https://api.hubapi.com/conversations/v3/conversations/search’
# Define the headers
headers = {
‘Content-Type’: ‘application/json’,
‘Authorization’: f’Bearer {access_token}’
}
# Define the request payload
payload = {
“filters”: [
{
“propertyName”: “createDate”,
“operator”: “BETWEEN”,
“value”: {
“from”: start_time_iso,
“to”: end_time_iso
}
}
],
“sorts”: [
{
“propertyName”: “createDate”,
“direction”: “DESCENDING”
}
],
“limit”: 100,
“after”: 0
}
# Initialize list to store messages
messages = []
has_more = True
while has_more:
try:
response = requests.post(url, headers=headers, json=payload)
response.raise_for_status() # Raises an HTTPError for bad responses
data = response.json()
messages.extend(data[‘results’])
# Check if there’s more data to fetch
if data[‘paging’].get(‘next’, {}).get(‘after’):
payload[‘after’] = data[‘paging’][‘next’][‘after’]
else:
has_more = False
except requests.exceptions.RequestException as e:
print(f"An error occurred: {e}“)
print(f"Response status code: {response.status_code}”)
print(f"Response content: {response.content.decode(‘utf-8’)}“)
print(f"Request URL: {response.request.url}”)
print(f"Request headers: {response.request.headers}“)
print(f"Request body: {response.request.body}”)
break
# Export the retrieved messages to a CSV file
if messages:
with open(‘inbox_messages.csv’, ‘w’, newline=‘’, encoding=‘utf-8’) as file:
writer = csv.writer(file)
writer.writerow([‘id’, ‘type’, ‘text’, ‘timestamp’, ‘sender_type’, ‘sender’]) # Adjust headers as needed
for message in messages:
writer.writerow([
message[‘id’],
message.get(‘type’, ‘’),
message.get(‘text’, ‘’),
message.get(‘createdAt’, ‘’),
message.get(‘senderType’, ‘’),
message.get(‘sender’, {}).get(‘name’, ‘’)
])
print(f"Exported {len(messages)} inbox messages to inbox_messages.csv")
else:
print(“No messages were retrieved. Check your API credentials and filters.”)
print(“Script execution completed.”)
Can anyone assist? I am trying to avoid having our engineers getting involved to help because I feel like exporting all convos shouldnt be that hard? What am I missing and is there a way better way to do this for someone like me? (Basically just using AI to code for me, which has worked well for other APIs)