import os, requests, time
X = os.environ["X_BEARER_TOKEN"]; MV = os.environ["MAVERA_API_KEY"]
X_BASE = "https://api.x.com/2"; MV_BASE = "https://app.mavera.io/api/v1"
X_H = {"Authorization": f"Bearer {X}"}
MV_H = {"Authorization": f"Bearer {MV}", "Content-Type": "application/json"}
BRAND = "yourbrand"
queries = [f"@{BRAND} -is:retweet", f"url:x.com/{BRAND} is:quote"]
# 1. Collect interactions
interactions = []
for q in queries:
nt = None
for _ in range(3):
params = {"query": q, "max_results": 100, "tweet.fields": "public_metrics,author_id",
"expansions": "author_id", "user.fields": "username,description,public_metrics"}
if nt: params["next_token"] = nt
r = requests.get(f"{X_BASE}/tweets/search/recent", headers=X_H, params=params)
if r.status_code == 429:
time.sleep(int(r.headers.get("x-rate-limit-reset", time.time()+60)) - int(time.time()))
r = requests.get(f"{X_BASE}/tweets/search/recent", headers=X_H, params=params)
r.raise_for_status(); data = r.json()
users = {u["id"]: u for u in data.get("includes",{}).get("users",[])}
for t in data.get("data",[]):
a = users.get(t.get("author_id"),{})
interactions.append({"text": t["text"], "type": "quote" if "is:quote" in q else "reply",
"username": a.get("username",""), "bio": a.get("description",""),
"followers": a.get("public_metrics",{}).get("followers_count",0)})
nt = data.get("meta",{}).get("next_token")
if not nt: break
time.sleep(1)
# 2. Pattern analysis
block = "\n\n".join(f"@{i['username']} ({i['followers']:,} fol) [{i['type']}]\nBio: {i['bio'][:120]}\n{i['text'][:200]}"
for i in sorted(interactions, key=lambda x: -x["followers"])[:40])
analysis = requests.post(f"{MV_BASE}/mave/chat", headers=MV_H, json={
"message": f"Analyze reply/quote patterns around @{BRAND}. {len(interactions)} interactions.\n\n{block}\n\n"
"Identify 4-6 persona archetypes. For each: name, interaction pattern, topics, language, estimated %, value to brand."
}).json()
archetypes = analysis.get("content","")
print(archetypes[:1200])
# 3. Create personas
blocks = archetypes.split("##")[1:] if "##" in archetypes else [archetypes]
for b in blocks[:6]:
lines = b.strip().split("\n")
name = lines[0].strip().strip("#").strip() if lines else "X Persona"
p = requests.post(f"{MV_BASE}/personas", headers=MV_H, json={
"name": f"X Audience: {name}", "description": "\n".join(lines[1:])[:500],
"psychographic": {"source": "x_reply_mining", "brand": BRAND},
}).json()
print(f" Created: {name} — {p['id']}")
time.sleep(0.3)