import os, requests
from collections import defaultdict
DOMAIN = os.environ["PIPEDRIVE_DOMAIN"]
PD_TOKEN = os.environ["PIPEDRIVE_API_TOKEN"]
PD_BASE = f"https://{DOMAIN}.pipedrive.com"
MAVERA_KEY = os.environ["MAVERA_API_KEY"]
def pd_get(path, params=None):
params = params or {}
params["api_token"] = PD_TOKEN
r = requests.get(f"{PD_BASE}{path}", params=params)
r.raise_for_status()
return r.json()
# 1. Pull activities grouped by person
person_acts = defaultdict(lambda: defaultdict(int))
for act_type in ["call", "email", "meeting"]:
for act in (pd_get("/api/v2/activities", {"type": act_type, "limit": 200, "done": 1}).get("data", []) or []):
if act.get("person_id"):
person_acts[act["person_id"]][act_type] += 1
# 2. Classify into persona archetypes
def classify(counts):
c, e, m = counts.get("call", 0), counts.get("email", 0), counts.get("meeting", 0)
if m >= 3 and c >= 5: return "Relationship Buyer (exec, high-touch)"
if e >= 10 and c <= 2: return "Self-Serve Researcher (dev/IC, async-first)"
if c >= 5 and e >= 5: return "Engaged Evaluator (mid-level, multi-channel)"
return "Low-Touch Prospect (early stage)"
groups = defaultdict(list)
for pid, counts in person_acts.items():
groups[classify(counts)].append({"person_id": pid, **counts})
# 3. Build and run Focus Group
personas = []
for archetype, members in groups.items():
personas.append({
"name": archetype.split("(")[0].strip(),
"description": (
f"{archetype}. {len(members)} contacts. "
f"Avg calls: {sum(m.get('call',0) for m in members)//max(len(members),1)}, "
f"Avg emails: {sum(m.get('email',0) for m in members)//max(len(members),1)}."
),
})
fg_resp = requests.post(
"https://app.mavera.io/api/v1/focus-groups",
headers={"Authorization": f"Bearer {MAVERA_KEY}"},
json={
"title": "Pipedrive Activity-Based Persona Scoring",
"personas": personas,
"questions": [
"How many touchpoints (calls, emails, meetings) before considering a demo?",
"What outreach feels helpful vs. intrusive at each stage?",
"At what point does follow-up become too aggressive?",
],
},
)
fg_resp.raise_for_status()
fg = fg_resp.json()
for resp in fg.get("responses", []):
print(f"\n[{resp['persona_name']}]")
for a in resp.get("answers", []):
print(f" Q: {a['question']}\n A: {a['answer'][:300]}")
