Hey @karstenkoehler,
I have not found a custom code solution to this problem. It is a little code extensive, but it seems to be working pretty well.
I am using the HubSpot API to retrieve the “Call Notes” extracting the AirCall tags within and then updating the “Call Type” based on the tags. There are a couple of “issues” with this, which I will describe in more detail below:
1. The code does not have a trigger so I am just running it once pr. 24 hours. For me, this is fine since I am using the type for reporting and not any immediate trigger actions.
2. The retrieval of calls with the HubSpot API does not allow calls done in a specific timeframe and will always start its query from the oldest records. This means that as you get more calls in your CRM, the script will take longer and longer to run. However, there is the ability to provide an offset to the query by providing the ID of the call you want to start with; this solves the problem. However, this introduces the issue of needing to store this offset value and updating it every time the script is run. The code provided below contains some pseudo code for this, which needs to be updated to fit the environment you are running it in. I am personally running this in Zapier, where I am using “Storage by Zapier” to keep track of it.
Code:
import requests
import json
import re
from datetime import datetime, timedelta
# HubSpot API Access Token
ACCESS_TOKEN = "YOUR ACCESS TOKEN HERE"
# Headers with authorization token
HEADERS = {
'Authorization': f'Bearer {ACCESS_TOKEN}',
'Content-Type': 'application/json'
}
# HubSpot API Endpoints
CALLS_URL = 'https://api.hubapi.com/crm/v3/objects/calls'
UPDATE_CALL_URL = 'https://api.hubapi.com/crm/v3/objects/calls/{call_id}'
CALL_ID_OFFSET = 70202354556 # Offset
# Define your tag-to-call-type mapping
TAG_TO_TYPE_MAPPING = {
'prospect': 'Prospect',
'onboarding': 'Onboarding',
'checkin': 'Check In',
'support call': 'Tech Support',
'testimonial': 'Testimonial',
'late checkin': 'Late Check In',
}
def save_after_token(after_token):
"""
Save the after token in Zapier Storage for the next execution.
"""
data = {"after": after_token}
response = requests.post(CALL_ID_OFFSET, json=data)
if response.status_code != 200:
print(f"Error saving after token: {response.status_code}, {response.text}")
def load_after_token():
"""
Load the after token from Zapier Storage.
"""
response = requests.get(CALL_ID_OFFSET)
if response.status_code == 200:
data = response.json()
return data.get("after", None)
return None
def fetch_calls():
"""
Fetch calls from HubSpot, including call notes (hs_call_body), resuming from the last saved 'after' token if available.
Returns:
list: A list of call records.
"""
params = {
'limit': 100, # Max records per request
'properties': ['hs_call_disposition', 'hs_timestamp', 'hs_call_title', 'hs_call_body'],
'associations': 'contacts'
}
after = 70202354556 #load_after_token()
all_calls = []
while True:
if after:
params['after'] = after
response = requests.get(CALLS_URL, headers=HEADERS, params=params)
if response.status_code == 200:
data = response.json()
results = data.get('results', [])
after = data.get('paging', {}).get('next', {}).get('after')
for call in results:
all_calls.append({
'id': call['id'],
'timestamp': call['properties'].get('hs_timestamp'),
'subject': call['properties'].get('hs_call_title', 'No title provided'),
'notes': call['properties'].get('hs_call_body', ''), # Retrieve HTML call notes
'contact_ids': call.get('associations', {}).get('contacts', [])
})
if after:
save_after_token(after)
if not after:
break
else:
print(f"Error fetching calls: {response.status_code}, {response.text}")
break
return all_calls
def determine_call_type(notes):
"""
Determine the call type based on extracted tags.
Args:
tags (list): The list of tags found in the call notes.
Returns:
str: The corresponding call type, or None if no match.
"""
for tag in TAG_TO_TYPE_MAPPING.keys():
if tag in notes:
return tag
return None
def update_call_type(call_id, new_type):
"""
Update the call type in HubSpot.
Args:
call_id (str): The HubSpot call ID.
new_type (str): The new call type.
Returns:
bool: True if the update was successful, False otherwise.
"""
url = UPDATE_CALL_URL.format(call_id=call_id)
data = {
"properties": {
"hs_activity_type": new_type
}
}
response = requests.patch(url, headers=HEADERS, json=data)
if response.status_code == 200:
return True
else:
print(f"Failed to update call {call_id}: {response.status_code}, {response.text}")
return False
def process_calls():
"""
Main function to fetch calls, analyze tags, and update call types.
"""
calls = fetch_calls()
print(f"Fetched {len(calls)} calls from the last run.")
for call in calls:
tag = determine_call_type(call["notes"])
if tag:
print(f"Updating Call ID {call['id']} to Type: {tag}")
update_call_type(call['id'], tag)
else:
print(f"No matching call type for Call ID {call['id']}. Skipping.")
# Run in once pr. day wherever you host your code (I use zapier, which can also store my offset value)
if __name__ == '__main__':
process_calls()
I hope this helps, and if you have any questions, feel free to reach out.
Best,
Christopher Ravn Boe Jensen
Head of RevOps at OptoCeutics
And now my team doesn’t have to leave Aircall to log call types or outcomes and I can just make sure to train them if they’re not logging one or the other correctly.