import os, requests, time
from collections import defaultdict
SEG_TOKEN = os.environ["SEGMENT_TOKEN"]
MV = os.environ["MAVERA_API_KEY"]
MB = "https://app.mavera.io/api/v1"
MH = {"Authorization": f"Bearer {MV}", "Content-Type": "application/json"}
SH = {"Authorization": f"Bearer {SEG_TOKEN}", "Content-Type": "application/json"}
SPACE_ID = os.environ.get("SEGMENT_SPACE_ID", "spa_xxxxx")
r = requests.get(
f"https://api.segmentapis.com/spaces/{SPACE_ID}/collections/users/profiles",
headers=SH,
params={"limit": 100, "include": "traits,external_ids"},
)
profiles = []
if r.status_code == 200:
profiles = r.json().get("data", {}).get("profiles", [])
if not profiles:
profiles = [
{
"traits": {"email": "j.doe@example.com", "ltv": 890, "plan": "pro", "last_web_visit": "2026-03-15", "last_mobile_open": "2026-03-16", "last_email_click": "2026-03-12", "web_sessions_30d": 14, "mobile_sessions_30d": 22, "emails_opened_30d": 8},
"external_ids": [{"type": "user_id", "id": "usr_001"}, {"type": "ios_id", "id": "ios_abc"}, {"type": "email", "id": "j.doe@example.com"}],
},
{
"traits": {"email": "s.chen@example.com", "ltv": 340, "plan": "starter", "last_web_visit": "2026-03-10", "last_mobile_open": "2026-02-28", "last_email_click": "2026-03-14", "web_sessions_30d": 4, "mobile_sessions_30d": 1, "emails_opened_30d": 12},
"external_ids": [{"type": "user_id", "id": "usr_002"}, {"type": "email", "id": "s.chen@example.com"}],
},
{
"traits": {"email": "m.patel@example.com", "ltv": 1450, "plan": "enterprise", "last_web_visit": "2026-03-17", "last_mobile_open": "2026-03-17", "last_email_click": "2026-03-16", "web_sessions_30d": 28, "mobile_sessions_30d": 35, "emails_opened_30d": 15},
"external_ids": [{"type": "user_id", "id": "usr_003"}, {"type": "ios_id", "id": "ios_xyz"}, {"type": "android_id", "id": "and_789"}, {"type": "email", "id": "m.patel@example.com"}],
},
] * 10
channel_patterns = {"web_dominant": [], "mobile_dominant": [], "email_dominant": [], "omnichannel": []}
for p in profiles:
traits = p.get("traits", {})
ext_ids = p.get("external_ids", [])
id_types = [e.get("type", "") for e in ext_ids]
web = int(traits.get("web_sessions_30d", 0) or 0)
mobile = int(traits.get("mobile_sessions_30d", 0) or 0)
email = int(traits.get("emails_opened_30d", 0) or 0)
total = web + mobile + email
if total == 0:
continue
profile_data = {
"ltv": float(traits.get("ltv", 0) or 0),
"plan": traits.get("plan", "unknown"),
"web": web, "mobile": mobile, "email": email,
"id_count": len(id_types),
"channels": id_types,
}
web_pct = web / total
mobile_pct = mobile / total
email_pct = email / total
if web_pct > 0.5:
channel_patterns["web_dominant"].append(profile_data)
elif mobile_pct > 0.5:
channel_patterns["mobile_dominant"].append(profile_data)
elif email_pct > 0.5:
channel_patterns["email_dominant"].append(profile_data)
else:
channel_patterns["omnichannel"].append(profile_data)
def summarize_pattern(users, label):
if not users:
return ""
avg_ltv = sum(u["ltv"] for u in users) / len(users)
avg_web = sum(u["web"] for u in users) / len(users)
avg_mobile = sum(u["mobile"] for u in users) / len(users)
avg_email = sum(u["email"] for u in users) / len(users)
avg_ids = sum(u["id_count"] for u in users) / len(users)
plans = defaultdict(int)
for u in users:
plans[u["plan"]] += 1
top_plans = sorted(plans, key=plans.get, reverse=True)[:3]
return (
f"**{label}** (N={len(users)})\n"
f" Avg LTV: ${avg_ltv:.0f} | Web sessions: {avg_web:.0f} | Mobile: {avg_mobile:.0f} | Email opens: {avg_email:.0f}\n"
f" Avg identity count: {avg_ids:.1f} | Plans: {', '.join(top_plans)}"
)
pattern_block = "\n\n".join(
summarize_pattern(users, label.replace("_", " ").title())
for label, users in channel_patterns.items()
if users
)
mave = requests.post(
f"{MB}/mave/chat",
headers=MH,
json={"message": f"""Create a unified persona framework that accounts for cross-channel behavior. Users interact with our product via website, mobile app, and email — and the same user behaves differently on each channel.
CROSS-CHANNEL PATTERNS (from Segment identity graph, 30 days):
{pattern_block}
For each channel pattern, provide:
1. A persona name and 2-sentence description capturing their cross-channel identity
2. Why they prefer their dominant channel — what need does it serve?
3. How to reach them on their weaker channels — what message/format would work?
4. Cross-channel journey map — how do they typically move between channels?
5. Content format recommendations per channel
6. Risk of losing them — which channel loss would cause churn?
Then provide an overall cross-channel strategy:
- Where to invest in consistency vs. channel-specific adaptation
- Identity resolution opportunities (matching anonymous to known)
- Moments when users switch channels and what triggers the switch"""},
).json()
print("--- Cross-Channel Unified Persona Framework ---")
print(mave.get("content", "")[:3000])
