import os, requests, time
from google.analytics.data_v1beta import BetaAnalyticsDataClient
from google.analytics.data_v1beta.types import (
RunFunnelReportRequest, Funnel, FunnelStep, FunnelFilterExpression,
FunnelFieldFilter, StringFilter, DateRange,
)
PROPERTY_ID = os.environ["GA4_PROPERTY_ID"]
MV = os.environ["MAVERA_API_KEY"]
MB = "https://app.mavera.io/api/v1"
MH = {"Authorization": f"Bearer {MV}", "Content-Type": "application/json"}
client = BetaAnalyticsDataClient()
funnel_report = client.run_funnel_report(RunFunnelReportRequest(
property=f"properties/{PROPERTY_ID}",
date_ranges=[DateRange(start_date="30daysAgo", end_date="today")],
funnel=Funnel(steps=[
FunnelStep(name="Homepage visit", filter_expression=FunnelFilterExpression(
funnel_field_filter=FunnelFieldFilter(
field_name="pagePath", string_filter=StringFilter(value="/", match_type=StringFilter.MatchType.EXACT)))),
FunnelStep(name="Product page", filter_expression=FunnelFilterExpression(
funnel_field_filter=FunnelFieldFilter(
field_name="pagePath", string_filter=StringFilter(value="/product", match_type=StringFilter.MatchType.BEGINS_WITH)))),
FunnelStep(name="Pricing page", filter_expression=FunnelFilterExpression(
funnel_field_filter=FunnelFieldFilter(
field_name="pagePath", string_filter=StringFilter(value="/pricing", match_type=StringFilter.MatchType.EXACT)))),
FunnelStep(name="Signup start", filter_expression=FunnelFilterExpression(
funnel_field_filter=FunnelFieldFilter(
field_name="pagePath", string_filter=StringFilter(value="/signup", match_type=StringFilter.MatchType.BEGINS_WITH)))),
FunnelStep(name="Signup complete", filter_expression=FunnelFilterExpression(
funnel_field_filter=FunnelFieldFilter(
field_name="eventName", string_filter=StringFilter(value="sign_up", match_type=StringFilter.MatchType.EXACT)))),
]),
))
steps = []
for row in funnel_report.funnel_table.rows:
step_name = row.dimension_values[0].value
users = int(row.metric_values[0].value)
steps.append({"step": step_name, "users": users})
drop_offs = []
for i in range(1, len(steps)):
prev = steps[i - 1]["users"]
curr = steps[i]["users"]
rate = (prev - curr) / max(prev, 1)
drop_offs.append({
"from": steps[i - 1]["step"], "to": steps[i]["step"],
"users_lost": prev - curr, "drop_rate": rate,
})
drop_offs.sort(key=lambda d: d["drop_rate"], reverse=True)
funnel_summary = "\n".join(f"- {s['step']}: {s['users']} users" for s in steps)
dropoff_summary = "\n".join(
f"- {d['from']} → {d['to']}: {d['users_lost']} users lost ({d['drop_rate']:.0%} drop-off)"
for d in drop_offs
)
PERSONA_IDS = os.environ.get("FUNNEL_PERSONA_IDS", "").split(",")
if not PERSONA_IDS[0]:
for name, desc in [
("First-Time Visitor", "New to the product. Browsing casually, easily distracted, skeptical of claims."),
("Comparison Shopper", "Evaluating 3-4 tools. Focused on pricing, features, and proof points."),
("Technical Decision-Maker", "Engineer or IT lead. Needs API docs, security info, and integration details."),
]:
p = requests.post(f"{MB}/personas", headers=MH, json={"name": name, "description": desc}).json()
PERSONA_IDS.append(p["id"])
time.sleep(0.3)
worst = drop_offs[0] if drop_offs else {"from": "Unknown", "to": "Unknown", "drop_rate": 0}
fg = requests.post(f"{MB}/focus-groups", headers=MH, json={
"name": "GA4 Funnel Drop-Off Investigation",
"persona_ids": [pid for pid in PERSONA_IDS if pid],
"questions": [
f"You're browsing our website and reach the '{worst['from']}' step. What would make you leave before reaching '{worst['to']}'? Be specific.",
f"Rank these reasons for abandoning a signup flow: (1) Too many form fields, (2) No pricing transparency, (3) Required credit card, (4) Unclear value proposition, (5) Slow page load. Explain your #1.",
"What's the minimum information you need to see before signing up for a new software tool?",
"If you saw a pricing page with three tiers, what would make you click away instead of choosing one?",
],
"context": f"""This is a funnel analysis for a B2B marketing platform. Here's the conversion funnel:
{funnel_summary}
Drop-off analysis:
{dropoff_summary}
The biggest drop-off is {worst['from']} → {worst['to']} at {worst['drop_rate']:.0%}.""",
"responses_per_persona": 2,
}).json()
for _ in range(24):
time.sleep(5)
data = requests.get(f"{MB}/focus-groups/{fg['id']}", headers=MH).json()
if data.get("status") == "completed":
break
print(f"Focus Group: {data.get('id')} — {data.get('status')}\n")
for resp in data.get("responses", []):
print(f"[{resp.get('persona_id','?')}] {resp.get('question','')[:70]}")
print(f" → {resp.get('answer','')[:300]}\n")
