import os, csv, io, zipfile, requests, time
from collections import defaultdict
QT = os.environ["QUALTRICS_TOKEN"]
DC = os.environ["QUALTRICS_DC"]
MV = os.environ["MAVERA_API_KEY"]
Q_BASE = f"https://{DC}.qualtrics.com/API/v3"
MB = "https://app.mavera.io/api/v1"
Q_H = {"X-API-TOKEN": QT, "Content-Type": "application/json"}
MV_H = {"Authorization": f"Bearer {MV}", "Content-Type": "application/json"}
SURVEY_ID = os.environ.get("ENPS_SURVEY_ID", "SV_xxxxx")
# 1. Export (same async pattern)
export = requests.post(f"{Q_BASE}/surveys/{SURVEY_ID}/export-responses",
headers=Q_H, json={"format": "csv"}).json()
pid = export["result"]["progressId"]
file_id = None
for _ in range(60):
time.sleep(5)
s = requests.get(f"{Q_BASE}/surveys/{SURVEY_ID}/export-responses/{pid}",
headers=Q_H).json()
if s["result"].get("percentComplete") == 100:
file_id = s["result"]["fileId"]; break
zip_data = requests.get(
f"{Q_BASE}/surveys/{SURVEY_ID}/export-responses/{file_id}/file",
headers=Q_H).content
with zipfile.ZipFile(io.BytesIO(zip_data)) as zf:
csv_name = [n for n in zf.namelist() if n.endswith(".csv")][0]
with zf.open(csv_name) as f:
rows = list(csv.DictReader(io.TextIOWrapper(f, encoding="utf-8-sig")))
responses = [r for r in rows if r.get("Finished") == "1"]
print(f"Employee responses: {len(responses)}")
# 2. Segment employees
ENPS_COL = os.environ.get("ENPS_COLUMN", "eNPS")
TENURE_COL = os.environ.get("TENURE_COLUMN", "Tenure_Months")
segments = {"engaged_advocate": [], "passive_performer": [],
"quiet_quitter": [], "new_hire": [], "at_risk_veteran": []}
for r in responses:
try:
enps = float(r.get(ENPS_COL, 0) or 0)
except ValueError:
continue
try:
tenure = float(r.get(TENURE_COL, 12) or 12)
except ValueError:
tenure = 12
if tenure < 6:
segments["new_hire"].append(r)
elif enps >= 9:
segments["engaged_advocate"].append(r)
elif enps >= 7:
segments["passive_performer"].append(r)
elif enps >= 4 and tenure > 24:
segments["at_risk_veteran"].append(r)
else:
segments["quiet_quitter"].append(r)
for name, members in segments.items():
print(f" {name}: {len(members)}")
# 3. Create internal personas
PERSONA_PROFILES = {
"engaged_advocate": {
"name": "QX Internal: Engaged Advocate",
"desc": "eNPS 9-10. Enthusiastic about company direction. Shares updates proactively. First to volunteer for initiatives.",
"mindset": "Invested, energized, wants to be recognized and involved",
},
"passive_performer": {
"name": "QX Internal: Passive Performer",
"desc": "eNPS 7-8. Doing their job well but not engaged beyond requirements. Reads comms but rarely responds.",
"mindset": "Content but disengaged from culture. Needs a reason to care more.",
},
"quiet_quitter": {
"name": "QX Internal: Quiet Quitter",
"desc": "eNPS 0-6. Doing minimum required. Skeptical of corporate comms. May be passively job searching.",
"mindset": "Disillusioned. Reads announcements looking for reasons to leave or stay.",
},
"new_hire": {
"name": "QX Internal: New Hire",
"desc": "Under 6 months tenure. Still forming opinions. Observing culture. Comparing to previous employer.",
"mindset": "Observant, optimistic but cautious. Every comm shapes first impression.",
},
"at_risk_veteran": {
"name": "QX Internal: At-Risk Veteran",
"desc": "2+ years tenure, eNPS 4-6. Was once engaged. Something shifted. Institutional knowledge at risk.",
"mindset": "Nostalgic for how things were. Needs to see positive change, not hear about it.",
},
}
persona_ids = {}
for seg_key, profile in PERSONA_PROFILES.items():
members = segments.get(seg_key, [])
if not members:
continue
depts = list({m.get("Department", "") for m in members if m.get("Department")})[:5]
p = requests.post(f"{MB}/personas", headers=MV_H, json={
"name": profile["name"],
"description": f"{profile['desc']} N={len(members)}. Departments: {', '.join(depts[:3])}.",
"psychographic": {"mindset": profile["mindset"], "engagement_tier": seg_key},
})
p.raise_for_status()
persona_ids[seg_key] = p.json()["id"]
time.sleep(0.3)
# 4. Test internal communication
DRAFT_COMMS = [
{"type": "Policy Change", "text": "Starting Q2, we're moving to a hybrid model: 3 days in-office, 2 remote. This balances collaboration and flexibility based on team feedback."},
{"type": "Benefit Update", "text": "We're expanding mental health coverage to include 12 therapy sessions/year (up from 6) and adding a $500 wellness stipend."},
{"type": "Reorg Announcement", "text": "To accelerate our enterprise strategy, we're combining the Sales and CS teams under a new Chief Revenue Officer starting April 1."},
]
for comm in DRAFT_COMMS:
fg = requests.post(f"{MB}/focus-groups", headers=MV_H, json={
"name": f"Internal Comms: {comm['type']}",
"persona_ids": list(persona_ids.values()),
"context": f"Your company just sent this internal announcement:\n\n\"{comm['text']}\"",
"questions": [
"What is your immediate reaction to this announcement?",
"What questions does this raise that aren't answered?",
"How does this affect your engagement or morale?",
"If you were writing this announcement, what would you change?",
],
"responses_per_persona": 2,
}).json()
for _ in range(20):
time.sleep(5)
result = requests.get(f"{MB}/focus-groups/{fg['id']}",
headers=MV_H).json()
if result.get("status") == "completed":
break
print(f"\n{'='*50}")
print(f"COMMS TEST: {comm['type']}")
print(f"{'='*50}")
for resp in result.get("responses", []):
print(f"\n[{resp.get('persona_name', '?')}] {resp.get('question', '')[:50]}...")
print(f" → {resp.get('answer', '')[:250]}")
