import os, requests, time
from collections import defaultdict
AMP_KEY = os.environ["AMPLITUDE_API_KEY"]
AMP_SECRET = os.environ["AMPLITUDE_SECRET_KEY"]
MV = os.environ["MAVERA_API_KEY"]
MB = "https://app.mavera.io/api/v1"
MH = {"Authorization": f"Bearer {MV}", "Content-Type": "application/json"}
amp_auth = (AMP_KEY, AMP_SECRET)
r = requests.get(
"https://amplitude.com/api/2/composition",
auth=amp_auth,
params={"start": "20260215", "end": "20260317"},
)
if r.status_code == 429:
retry = int(r.headers.get("Retry-After", 60))
time.sleep(retry)
r = requests.get(
"https://amplitude.com/api/2/composition",
auth=amp_auth,
params={"start": "20260215", "end": "20260317"},
)
r.raise_for_status()
composition = r.json()
series = composition.get("data", {}).get("series", [])
xvals = composition.get("data", {}).get("xValues", [])
segments = defaultdict(lambda: {"total": 0, "countries": defaultdict(int),
"platforms": defaultdict(int), "devices": defaultdict(int)})
for i, date_label in enumerate(xvals):
for seg in series:
name = seg.get("name", "all")
val = seg.get("value", [0] * len(xvals))
count = val[i] if i < len(val) else 0
segments[name]["total"] += count
existing = requests.get(f"{MB}/personas", headers=MH).json()
existing_names = {p.get("name", ""): p for p in (existing if isinstance(existing, list) else [])}
PERSONA_SEGMENTS = {
"Mobile Power Users": {
"filter": lambda s: "mobile" in s.lower() or "ios" in s.lower() or "android" in s.lower(),
"description": "Users primarily accessing via mobile devices. Shorter sessions, gesture-driven navigation, on-the-go usage patterns.",
},
"Desktop Evaluators": {
"filter": lambda s: "desktop" in s.lower() or "windows" in s.lower() or "mac" in s.lower(),
"description": "Desktop-first users. Longer sessions, deeper feature exploration, likely evaluating the product for team adoption.",
},
"International Users": {
"filter": lambda s: any(c in s.lower() for c in ["uk", "de", "fr", "in", "br", "jp", "au"]),
"description": "Non-US users who may need localized content, timezone-aware engagement, and region-specific messaging.",
},
}
created = []
for persona_name, config in PERSONA_SEGMENTS.items():
matching = {k: v for k, v in segments.items() if config["filter"](k)}
if not matching:
continue
total = sum(v["total"] for v in matching.values())
segment_names = list(matching.keys())[:5]
full_name = f"Amplitude: {persona_name}"
payload = {
"name": full_name,
"description": (
f"{config['description']} "
f"Amplitude composition data (30d). Total: {total:,} users. "
f"Segments: {', '.join(segment_names)}."
),
"demographic": {
"source": "amplitude_composition",
"segments": segment_names,
"total_users": total,
},
}
if full_name in existing_names:
pid = existing_names[full_name]["id"]
requests.patch(f"{MB}/personas/{pid}", headers=MH, json=payload).raise_for_status()
created.append({"name": full_name, "id": pid, "action": "updated", "users": total})
print(f" Updated: {full_name} ({pid}) — {total:,} users")
else:
p = requests.post(f"{MB}/personas", headers=MH, json=payload).json()
created.append({"name": full_name, "id": p["id"], "action": "created", "users": total})
print(f" Created: {full_name} ({p['id']}) — {total:,} users")
time.sleep(0.3)
overall_total = sum(s["total"] for s in segments.values())
print(f"\nCalibrated {len(created)} personas from {overall_total:,} total composition data points")
