feat: Enhanced GA4 integration with engagement tracking
- Added scroll depth tracking (25%, 50%, 75%) - Track form interactions and outbound clicks - Added engagement rate, avg session duration, page views - Updated daily report with comprehensive engagement metrics - Created ga4-list-events.py to discover tracked events - All metrics now flow to morning brief
This commit is contained in:
@@ -1,10 +1,13 @@
|
||||
#!/usr/bin/env python3
|
||||
"""Google Analytics 4 - Direct JWT Authentication (No gcloud required)"""
|
||||
"""
|
||||
Google Analytics 4 - Direct JWT Authentication (No gcloud required)
|
||||
Enhanced to pull: sessions, users, engagement, scroll depth, form interactions
|
||||
"""
|
||||
import json
|
||||
import urllib.request
|
||||
from datetime import datetime, timedelta
|
||||
from pathlib import Path
|
||||
import subprocess
|
||||
from google.analytics.data import BetaAnalyticsDataClient
|
||||
from google.analytics.data_v1beta.types import RunReportRequest, DateRange, Metric, Dimension, Filter, FilterExpression
|
||||
|
||||
CONFIG_DIR = Path(__file__).parent.parent / "config"
|
||||
GA_CREDENTIALS = CONFIG_DIR / "ga-credentials.json"
|
||||
@@ -15,132 +18,144 @@ def load_credentials():
|
||||
with open(GA_CREDENTIALS) as f:
|
||||
return json.load(f)
|
||||
|
||||
def get_jwt_token(creds):
|
||||
"""Create and sign JWT for OAuth"""
|
||||
import base64
|
||||
import hashlib
|
||||
|
||||
# Check for PyJWT
|
||||
def get_engagement_data():
|
||||
"""Get comprehensive engagement metrics"""
|
||||
try:
|
||||
import jwt
|
||||
from cryptography.hazmat.primitives import serialization
|
||||
client = BetaAnalyticsDataClient.from_service_account_json(str(GA_CREDENTIALS))
|
||||
|
||||
now = datetime.utcnow()
|
||||
# Main traffic report
|
||||
request = RunReportRequest(
|
||||
property=f"properties/{GA_PROPERTY_ID}",
|
||||
date_ranges=[DateRange(start_date="1daysAgo", end_date="today")],
|
||||
metrics=[
|
||||
Metric(name="sessions"),
|
||||
Metric(name="activeUsers"),
|
||||
Metric(name="newUsers"),
|
||||
Metric(name="bounceRate"),
|
||||
Metric(name="averageSessionDuration"),
|
||||
Metric(name="engagementRate"),
|
||||
Metric(name="screenPageViews"),
|
||||
]
|
||||
)
|
||||
|
||||
claims = {
|
||||
"iss": creds['client_email'],
|
||||
"sub": creds['client_email'],
|
||||
"scope": "https://www.googleapis.com/auth/analytics.readonly",
|
||||
"aud": creds['token_uri'],
|
||||
"iat": now,
|
||||
"exp": now + timedelta(hours=1)
|
||||
response = client.run_report(request)
|
||||
|
||||
result = {
|
||||
"success": True,
|
||||
"timestamp": datetime.now().isoformat()
|
||||
}
|
||||
|
||||
private_key = creds['private_key']
|
||||
token = jwt.encode(claims, private_key, algorithm="RS256")
|
||||
return token
|
||||
except ImportError:
|
||||
return None
|
||||
|
||||
def get_access_token_with_jwt(creds):
|
||||
"""Get OAuth token using JWT"""
|
||||
jwt_token = get_jwt_token(creds)
|
||||
if not jwt_token:
|
||||
return None
|
||||
|
||||
body = {
|
||||
"grant_type": "urn:ietf:params:oauth:grant-type:jwt-bearer",
|
||||
"assertion": jwt_token
|
||||
}
|
||||
|
||||
req = urllib.request.Request(
|
||||
creds['token_uri'],
|
||||
data=json.dumps(body).encode(),
|
||||
headers={"Content-Type": "application/json"},
|
||||
method="POST"
|
||||
)
|
||||
|
||||
try:
|
||||
with urllib.request.urlopen(req, timeout=30) as r:
|
||||
data = json.loads(r.read().decode())
|
||||
return data.get('access_token')
|
||||
except Exception as e:
|
||||
print(f"Token error: {e}")
|
||||
return None
|
||||
|
||||
def get_access_token_with_curl(creds):
|
||||
"""Get token using curl"""
|
||||
try:
|
||||
result = subprocess.run(
|
||||
[
|
||||
"curl", "-s", "-X", "POST",
|
||||
creds['token_uri'],
|
||||
"-H", "Content-Type: application/x-www-form-urlencoded",
|
||||
"-d", f"grant_type=urn:ietf:params:oauth:grant-type:jwt-bearer",
|
||||
"--data-urlencode", f"assertion=<(echo 'JWT_PLACEHOLDER')"
|
||||
],
|
||||
capture_output=True,
|
||||
text=True,
|
||||
timeout=10
|
||||
)
|
||||
return None # Complex JWT signing needed
|
||||
except:
|
||||
return None
|
||||
|
||||
def query_ga4_direct():
|
||||
"""Query GA4 using Python requests if available"""
|
||||
try:
|
||||
creds = load_credentials()
|
||||
for row in response.rows:
|
||||
result['sessions'] = int(row.metric_values[0].value)
|
||||
result['activeUsers'] = int(row.metric_values[1].value)
|
||||
result['newUsers'] = int(row.metric_values[2].value)
|
||||
result['bounceRate'] = float(row.metric_values[3].value)
|
||||
result['averageSessionDuration'] = float(row.metric_values[4].value)
|
||||
result['engagementRate'] = float(row.metric_values[5].value)
|
||||
result['screenPageViews'] = int(row.metric_values[6].value)
|
||||
|
||||
# Method using google-analytics-data library
|
||||
try:
|
||||
from google.analytics.data import BetaAnalyticsDataClient
|
||||
from google.analytics.data_v1beta.types import RunReportRequest, DateRange, Metric, Dimension
|
||||
|
||||
client = BetaAnalyticsDataClient.from_service_account_json(str(GA_CREDENTIALS))
|
||||
|
||||
request = RunReportRequest(
|
||||
property=f"properties/{GA_PROPERTY_ID}",
|
||||
date_ranges=[DateRange(start_date="1daysAgo", end_date="today")],
|
||||
metrics=[
|
||||
Metric(name="sessions"),
|
||||
Metric(name="activeUsers"),
|
||||
Metric(name="newUsers")
|
||||
]
|
||||
# Get scroll depth events
|
||||
scroll_request = RunReportRequest(
|
||||
property=f"properties/{GA_PROPERTY_ID}",
|
||||
dimensions=[Dimension(name="eventName")],
|
||||
metrics=[Metric(name="eventCount")],
|
||||
date_ranges=[DateRange(start_date="1daysAgo", end_date="today")],
|
||||
dimension_filter=FilterExpression(
|
||||
filter=Filter(
|
||||
field_name="eventName",
|
||||
string_filter=Filter.StringFilter(
|
||||
match_type=Filter.StringFilter.MatchType.BEGINS_WITH,
|
||||
value="scroll"
|
||||
)
|
||||
)
|
||||
),
|
||||
limit=10
|
||||
)
|
||||
|
||||
scroll_response = client.run_report(scroll_request)
|
||||
scroll_data = {}
|
||||
for row in scroll_response.rows:
|
||||
event_name = row.dimension_values[0].value
|
||||
count = int(row.metric_values[0].value)
|
||||
scroll_data[event_name] = count
|
||||
|
||||
result['scroll_25'] = scroll_data.get('scroll_25', 0)
|
||||
result['scroll_50'] = scroll_data.get('scroll_50', 0)
|
||||
result['scroll_75'] = scroll_data.get('scroll_75', 0)
|
||||
result['scroll_total'] = sum(scroll_data.values())
|
||||
|
||||
# Get form interactions
|
||||
form_request = RunReportRequest(
|
||||
property=f"properties/{GA_PROPERTY_ID}",
|
||||
metrics=[Metric(name="eventCount")],
|
||||
date_ranges=[DateRange(start_date="1daysAgo", end_date="today")],
|
||||
dimension_filter=FilterExpression(
|
||||
filter=Filter(
|
||||
field_name="eventName",
|
||||
string_filter=Filter.StringFilter(
|
||||
match_type=Filter.StringFilter.MatchType.EXACT,
|
||||
value="form_start"
|
||||
)
|
||||
)
|
||||
)
|
||||
|
||||
response = client.run_report(request)
|
||||
|
||||
total_sessions = sum(int(r.metric_values[0].value) for r in response.rows)
|
||||
total_users = sum(int(r.metric_values[1].value) for r in response.rows)
|
||||
new_users = sum(int(r.metric_values[2].value) for r in response.rows)
|
||||
|
||||
return {
|
||||
"sessions": total_sessions,
|
||||
"activeUsers": total_users,
|
||||
"newUsers": new_users,
|
||||
"success": True
|
||||
}
|
||||
except ImportError:
|
||||
return {"error": "google-analytics-data library required", "install": "pip install google-analytics-data", "success": False}
|
||||
|
||||
)
|
||||
|
||||
form_response = client.run_report(form_request)
|
||||
form_count = 0
|
||||
for row in form_response.rows:
|
||||
form_count += int(row.metric_values[0].value)
|
||||
|
||||
result['form_interactions'] = form_count
|
||||
|
||||
# Get click events (outbound clicks)
|
||||
click_request = RunReportRequest(
|
||||
property=f"properties/{GA_PROPERTY_ID}",
|
||||
metrics=[Metric(name="eventCount")],
|
||||
date_ranges=[DateRange(start_date="1daysAgo", end_date="today")],
|
||||
dimension_filter=FilterExpression(
|
||||
filter=Filter(
|
||||
field_name="eventName",
|
||||
string_filter=Filter.StringFilter(
|
||||
match_type=Filter.StringFilter.MatchType.EXACT,
|
||||
value="click"
|
||||
)
|
||||
)
|
||||
)
|
||||
)
|
||||
|
||||
click_response = client.run_report(click_request)
|
||||
click_count = 0
|
||||
for row in click_response.rows:
|
||||
click_count += int(row.metric_values[0].value)
|
||||
|
||||
result['outbound_clicks'] = click_count
|
||||
|
||||
return result
|
||||
|
||||
except Exception as e:
|
||||
return {"error": str(e), "success": False}
|
||||
return {"success": False, "error": str(e)}
|
||||
|
||||
if __name__ == "__main__":
|
||||
print("🚀 Testing GA4 Direct Connection...")
|
||||
result = query_ga4_direct()
|
||||
print("🚀 Fetching GA4 Analytics Data...\n")
|
||||
|
||||
result = get_engagement_data()
|
||||
|
||||
if result.get('success'):
|
||||
print(f"""
|
||||
📊 GA4 Traffic Data (Last 24h):
|
||||
✅ Sessions: {result.get('sessions', 'N/A'):,}
|
||||
✅ Active Users: {result.get('activeUsers', 'N/A'):,}
|
||||
✅ New Users: {result.get('newUsers', 'N/A'):,}
|
||||
""")
|
||||
print("📊 Traffic (Last 24h):")
|
||||
print(f" ✅ Sessions: {result.get('sessions', 0):,}")
|
||||
print(f" ✅ Active Users: {result.get('activeUsers', 0):,}")
|
||||
print(f" ✅ New Users: {result.get('newUsers', 0):,}")
|
||||
print(f" ✅ Engagement Rate: {result.get('engagementRate', 0):.1%}")
|
||||
print(f" ✅ Bounce Rate: {result.get('bounceRate', 0):.1%}")
|
||||
print(f" ✅ Avg Session: {result.get('averageSessionDuration', 0):.1f}s")
|
||||
print(f" ✅ Page Views: {result.get('screenPageViews', 0):,}")
|
||||
|
||||
print("\n🎯 Engagement Events:")
|
||||
print(f" • Scroll Depth 25%: {result.get('scroll_25', 0)}")
|
||||
print(f" • Scroll Depth 50%: {result.get('scroll_50', 0)}")
|
||||
print(f" • Scroll Depth 75%: {result.get('scroll_75', 0)}")
|
||||
print(f" • Form Interactions: {result.get('form_interactions', 0)}")
|
||||
print(f" • Outbound Clicks: {result.get('outbound_clicks', 0)}")
|
||||
|
||||
else:
|
||||
print(f"❌ Error: {result.get('error')}")
|
||||
print(f"📦 Install: {result.get('install', 'N/A')}")
|
||||
print("")
|
||||
print("Quick fix:")
|
||||
print(" pip install google-analytics-data")
|
||||
Reference in New Issue
Block a user