| import os |
| import json |
| import warnings |
| from functools import lru_cache |
| from datetime import date, timedelta |
| from urllib.parse import urlparse |
|
|
| import pandas as pd |
| import gradio as gr |
| import plotly.graph_objects as go |
|
|
| from google.oauth2 import service_account |
| from google.analytics.data_v1beta import BetaAnalyticsDataClient |
| from google.analytics.data_v1beta.types import ( |
| RunReportRequest, |
| FilterExpression, |
| FilterExpressionList, |
| Filter |
| ) |
|
|
| from openai import OpenAI |
|
|
| |
| |
| |
| warnings.filterwarnings("ignore", category=FutureWarning) |
| warnings.filterwarnings("ignore", message=".*Invalid file descriptor.*") |
|
|
| |
| |
| |
| GA4_PROPERTY_ID = os.environ["GA4_PROPERTY_ID"] |
| SERVICE_ACCOUNT_JSON = json.loads(os.environ["SERVICE_ACCOUNT_JSON"]) |
| OPENAI_API_KEY = os.environ["OPENAI_API_KEY"] |
| SITE_DOMAIN = "rishabhsoft.com" |
|
|
| llm = OpenAI(api_key=OPENAI_API_KEY) |
|
|
| |
| |
| |
| with open("service_account.json", "w") as f: |
| json.dump(SERVICE_ACCOUNT_JSON, f) |
|
|
| creds = service_account.Credentials.from_service_account_file("service_account.json") |
| ga = BetaAnalyticsDataClient(credentials=creds) |
|
|
| |
| |
| |
| def clean_path(v): |
| if not v: |
| return "Unattributed (Page Unknown)" |
| p = urlparse(v).path or v |
| return "/ (Home)" if p == "/" else p |
|
|
| def classify_source(src): |
| if not src or src == "(direct)": |
| return "External" |
| if src.startswith("/"): |
| return "Internal" |
| if src.startswith("http"): |
| return "Internal" if SITE_DOMAIN in urlparse(src).netloc else "External" |
| return "External" |
|
|
| def prev_period(start, end): |
| s = pd.to_datetime(start) |
| e = pd.to_datetime(end) |
| delta = (e - s).days + 1 |
| prev_end = s - timedelta(days=1) |
| prev_start = prev_end - timedelta(days=delta - 1) |
| return str(prev_start.date()), str(prev_end.date()) |
|
|
| |
| |
| |
| @lru_cache(maxsize=3) |
| def fetch_top_pages(start, end): |
| req = RunReportRequest( |
| property=f"properties/{GA4_PROPERTY_ID}", |
| dimensions=[{"name": "pagePath"}], |
| metrics=[{"name": "screenPageViews"}], |
| date_ranges=[{"start_date": start, "end_date": end}], |
| order_bys=[{"metric": {"metric_name": "screenPageViews"}, "desc": True}], |
| limit=300 |
| ) |
| res = ga.run_report(req) |
| return sorted({r.dimension_values[0].value for r in res.rows}) |
|
|
| |
| |
| |
| def fetch_journey(start, end, landing_page): |
| req = RunReportRequest( |
| property=f"properties/{GA4_PROPERTY_ID}", |
| dimensions=[{"name": "pageReferrer"}, {"name": "pagePath"}], |
| metrics=[{"name": "screenPageViews"}], |
| date_ranges=[{"start_date": start, "end_date": end}], |
| dimension_filter=FilterExpression( |
| filter=Filter( |
| field_name="landingPage", |
| string_filter={"match_type": 4, "value": landing_page} |
| ) |
| ), |
| limit=3000 |
| ) |
|
|
| rows = [] |
| for r in ga.run_report(req).rows: |
| rows.append({ |
| "From": r.dimension_values[0].value or "(direct)", |
| "To": r.dimension_values[1].value, |
| "Views": int(r.metric_values[0].value) |
| }) |
|
|
| return pd.DataFrame(rows) |
|
|
| |
| |
| |
| def fetch_action_attribution(start, end, landing_page): |
| """ |
| Returns dataframe with enforced schema: |
| Final Action Page | Previous Page / Source | Country | Source Type | Action | Count |
| """ |
|
|
| ACTION_EVENTS = { |
| "Form Submit": [ |
| "generate_lead", |
| "form_submit", |
| "submit", |
| "contact_form_submit" |
| ], |
| "Email Click": [ |
| "email_click", |
| "mailto_click", |
| "click_email" |
| ] |
| } |
|
|
| rows = [] |
|
|
| for action, events in ACTION_EVENTS.items(): |
| req = RunReportRequest( |
| property=f"properties/{GA4_PROPERTY_ID}", |
| dimensions=[ |
| {"name": "pagePath"}, |
| {"name": "pageReferrer"}, |
| {"name": "country"}, |
| {"name": "eventName"} |
| ], |
| metrics=[{"name": "eventCount"}], |
| date_ranges=[{"start_date": start, "end_date": end}], |
| dimension_filter=FilterExpression( |
| and_group=FilterExpressionList(expressions=[ |
| FilterExpression( |
| filter=Filter( |
| field_name="eventName", |
| in_list_filter={"values": events} |
| ) |
| ), |
| FilterExpression( |
| filter=Filter( |
| field_name="landingPage", |
| string_filter={"match_type": 4, "value": landing_page} |
| ) |
| ) |
| ]) |
| ), |
| limit=3000 |
| ) |
|
|
| res = ga.run_report(req) |
|
|
| for r in res.rows: |
| final_page = clean_path(r.dimension_values[0].value) |
| prev_page = clean_path(r.dimension_values[1].value) |
| country = r.dimension_values[2].value or "Unknown" |
| count = int(r.metric_values[0].value) |
|
|
| if final_page == "Unattributed (Page Unknown)": |
| rows.append({ |
| "Final Action Page": "Unattributed (Page Unknown)", |
| "Previous Page / Source": "Unknown", |
| "Country": country, |
| "Source Type": "Unknown", |
| "Action": action, |
| "Count": count |
| }) |
| else: |
| rows.append({ |
| "Final Action Page": final_page, |
| "Previous Page / Source": prev_page, |
| "Country": country, |
| "Source Type": classify_source(prev_page), |
| "Action": action, |
| "Count": count |
| }) |
|
|
| EXPECTED_COLUMNS = [ |
| "Final Action Page", |
| "Previous Page / Source", |
| "Country", |
| "Source Type", |
| "Action", |
| "Count" |
| ] |
|
|
| df = pd.DataFrame(rows) |
|
|
| if df.empty: |
| return pd.DataFrame(columns=EXPECTED_COLUMNS) |
|
|
| return df |
|
|
| |
| |
| |
| @lru_cache(maxsize=100) |
| def classify_intent(start_page, pages_tuple, converted): |
| prompt = f""" |
| Classify user intent into ONE: |
| Conversion, Evaluation, Research, Navigation, Low Intent |
| Journey: |
| Start Page: {start_page} |
| Pages Visited: {", ".join(pages_tuple[:5])} |
| Converted: {"Yes" if converted else "No"} |
| Return: |
| Intent: <Intent> |
| Reason: <Short explanation> |
| """ |
| resp = llm.chat.completions.create( |
| model="gpt-4o-mini", |
| messages=[{"role": "user", "content": prompt}], |
| temperature=0 |
| ) |
|
|
| intent, reason = "Unknown", "" |
| for line in resp.choices[0].message.content.splitlines(): |
| if line.lower().startswith("intent"): |
| intent = line.split(":", 1)[1].strip() |
| if line.lower().startswith("reason"): |
| reason = line.split(":", 1)[1].strip() |
|
|
| return intent, reason |
|
|
| def intent_confidence(form_count, page_depth): |
| if form_count > 0: |
| return "High" |
| if page_depth >= 3: |
| return "Medium" |
| return "Low" |
|
|
| |
| |
| |
| def run_journey(start, end, landing_page): |
| df = fetch_journey(start, end, landing_page) |
|
|
| if df.empty: |
| return None, pd.DataFrame(), "### No data for this selection." |
|
|
| action_df = fetch_action_attribution(start, end, landing_page) |
|
|
| if action_df.empty: |
| form_count = 0 |
| email_count = 0 |
| else: |
| form_count = int( |
| action_df.loc[action_df["Action"] == "Form Submit", "Count"].sum() |
| ) |
| email_count = int( |
| action_df.loc[action_df["Action"] == "Email Click", "Count"].sum() |
| ) |
|
|
| total_users = int(df["Views"].sum()) |
| conv_rate = round((form_count / total_users) * 100, 2) if total_users else 0.0 |
|
|
| top_pages = df.sort_values("Views", ascending=False)["To"].unique().tolist() |
|
|
| intent, reason = classify_intent( |
| clean_path(landing_page), |
| tuple(clean_path(p) for p in top_pages), |
| form_count > 0 |
| ) |
| confidence = intent_confidence(form_count, len(top_pages)) |
|
|
| labels = ["Start", "Exit", "Form Submit", "Email Click"] |
| sources, targets, values = [], [], [] |
|
|
| if total_users: |
| sources.append(0) |
| targets.append(1) |
| values.append(total_users - form_count - email_count) |
|
|
| if form_count: |
| sources.append(0) |
| targets.append(2) |
| values.append(form_count) |
|
|
| if email_count: |
| sources.append(0) |
| targets.append(3) |
| values.append(email_count) |
|
|
| fig = go.Figure(go.Sankey( |
| node=dict(label=labels, pad=24, thickness=18), |
| link=dict(source=sources, target=targets, value=values) |
| )) |
|
|
| fig.update_layout( |
| title=f"Journey → Final Actions ({clean_path(landing_page)})", |
| height=600 |
| ) |
|
|
| summary = f""" |
| ### 🧭 Journey Summary |
| **Start Page:** `{clean_path(landing_page)}` |
| **Date Range:** {start} → {end} |
| **Conversions** |
| - Form Submits: **{form_count}** |
| - Email Clicks: **{email_count}** |
| - Conversion Rate: **{conv_rate:.2f}%** |
| **Intent** |
| - Primary Intent: **{intent}** |
| - Confidence: **{confidence}** |
| - Reason: {reason} |
| ℹ️ *Unattributed actions are clearly labeled when GA4 does not provide page context.* |
| """ |
|
|
| return fig, action_df, summary |
|
|
| |
| |
| |
| with gr.Blocks() as app: |
| gr.Markdown("## 🚀 Page-Aware Customer Journey Attribution (GA4 + AI)") |
|
|
| with gr.Row(): |
| start_date = gr.Textbox("2025-12-01", label="Start Date") |
| end_date = gr.Textbox(str(date.today()), label="End Date") |
|
|
| page_path = gr.Dropdown(label="Starting Page", choices=[], allow_custom_value=True) |
|
|
| with gr.Row(): |
| load_btn = gr.Button("⚡ Load Pages") |
| run_btn = gr.Button("Analyze Journey", variant="primary") |
|
|
| journey_chart = gr.Plot() |
| action_table = gr.Dataframe(label="Final Action Attribution (Page-Scoped)") |
| summary_md = gr.Markdown() |
|
|
| load_btn.click( |
| lambda s, e: gr.update(choices=fetch_top_pages(s, e)), |
| inputs=[start_date, end_date], |
| outputs=page_path |
| ) |
|
|
| run_btn.click( |
| run_journey, |
| inputs=[start_date, end_date, page_path], |
| outputs=[journey_chart, action_table, summary_md] |
| ) |
|
|
| if __name__ == "__main__": |
| app.launch() |
|
|
