import os, requests, time
from collections import defaultdict
HS = os.environ["HUBSPOT_ACCESS_TOKEN"]
MV = os.environ["MAVERA_API_KEY"]
MB = "https://app.mavera.io/api/v1"
MH = {"Authorization": f"Bearer {MV}", "Content-Type": "application/json"}
ENG_PROPS = ["hs_email_open_count", "hs_email_click_count",
"hs_analytics_num_page_views", "num_conversion_events",
"jobtitle", "industry", "lifecyclestage"]
# 1. Pull engaged contacts
contacts, after = [], None
while len(contacts) < 300:
payload = {
"filterGroups": [{"filters": [{"propertyName": "hs_email_open_count", "operator": "GT", "value": "0"}]}],
"properties": ENG_PROPS,
"sorts": [{"propertyName": "hs_email_open_count", "direction": "DESCENDING"}],
"limit": 100,
}
if after: payload["after"] = after
r = requests.post("https://api.hubapi.com/crm/v3/objects/contacts/search",
headers={"Authorization": f"Bearer {HS}"}, json=payload)
if r.status_code == 429: time.sleep(1); continue
r.raise_for_status()
data = r.json()
contacts.extend(data.get("results", []))
after = data.get("paging", {}).get("next", {}).get("after")
if not after: break
# 2. Score and tier
def score(c):
p = c.get("properties", {})
return int(p.get("hs_email_open_count") or 0) + int(p.get("hs_email_click_count") or 0)*3 + int(p.get("hs_analytics_num_page_views") or 0)*2 + int(p.get("num_conversion_events") or 0)*10
scored = sorted([(c, score(c)) for c in contacts], key=lambda x: -x[1])
third = len(scored) // 3
tiers = {"High": scored[:third], "Medium": scored[third:2*third], "Low": scored[2*third:]}
def profile(tier):
titles, inds = defaultdict(int), defaultdict(int)
for c, _ in tier:
p = c.get("properties", {})
if p.get("jobtitle"): titles[p["jobtitle"]] += 1
if p.get("industry"): inds[p["industry"]] += 1
return {"n": len(tier), "avg": sum(s for _,s in tier)/max(len(tier),1),
"titles": sorted(titles, key=titles.get, reverse=True)[:5],
"industries": sorted(inds, key=inds.get, reverse=True)[:3]}
# 3. Get existing personas
existing = requests.get(f"{MB}/personas", headers=MH).json()
hs_personas = {p["name"]: p for p in (existing if isinstance(existing, list) else []) if "HubSpot" in p.get("name","")}
# 4. Update or create
for tier_name, tier_data in tiers.items():
prof = profile(tier_data)
name = f"HubSpot {tier_name} Engagement"
desc = f"{tier_name} engagement. Avg: {prof['avg']:.1f}. N={prof['n']}. Titles: {', '.join(prof['titles'][:3])}."
payload = {"name": name, "description": desc, "demographic": {"job_titles": prof["titles"], "industries": prof["industries"]},
"psychographic": {"engagement_level": tier_name.lower(), "avg_score": prof["avg"]}}
if name in hs_personas:
r = requests.patch(f"{MB}/personas/{hs_personas[name]['id']}", headers=MH, json=payload)
r.raise_for_status()
print(f"Updated: {name} ({hs_personas[name]['id']})")
else:
r = requests.post(f"{MB}/personas", headers=MH, json=payload)
r.raise_for_status()
print(f"Created: {name} ({r.json()['id']})")
time.sleep(0.3)
