olsch01
5319bcd30b
- Created chatwoot-agent-bot/ with Node.js webhook server - Bot detects intent (greeting, billing, technical, features, account) - Auto-responds from FAQ knowledge base or escalates to human - FAQ-KB.md: Living knowledge base that grows with customer questions - CHATWOOT-SETUP.md: Complete deployment and configuration guide - Supports Telegram notifications on escalation - Bot runs on port 3001, ready for Chatwoot webhook integration
129 lines
3.9 KiB
Python
129 lines
3.9 KiB
Python
#!/usr/bin/env python3
|
|
"""Google Analytics 4 Integration for SEO Agent"""
|
|
import json
|
|
import os
|
|
import subprocess
|
|
from datetime import datetime, timedelta
|
|
from pathlib import Path
|
|
|
|
CONFIG_DIR = Path(__file__).parent.parent / "config"
|
|
GA_CREDENTIALS = CONFIG_DIR / "ga-credentials.json"
|
|
GA_PROPERTY_ID = "526394825" # Your GA4 Property ID
|
|
|
|
def load_credentials():
|
|
"""Load service account credentials"""
|
|
with open(GA_CREDENTIALS) as f:
|
|
return json.load(f)
|
|
|
|
def get_access_token():
|
|
"""Get OAuth access token using gcloud or direct call"""
|
|
creds = load_credentials()
|
|
|
|
# Method 1: Try gcloud auth activate-service-account
|
|
try:
|
|
result = subprocess.run([
|
|
"gcloud", "auth", "activate-service-account",
|
|
creds['client_email'],
|
|
"--key-file", str(GA_CREDENTIALS),
|
|
"--project", creds['project_id']
|
|
], capture_output=True, text=True, timeout=30)
|
|
|
|
if result.returncode == 0:
|
|
# Get token
|
|
token_result = subprocess.run(
|
|
["gcloud", "auth", "print-access-token"],
|
|
capture_output=True, text=True, timeout=10
|
|
)
|
|
if token_result.returncode == 0:
|
|
return token_result.stdout.strip()
|
|
except:
|
|
pass
|
|
|
|
return None
|
|
|
|
def query_ga4_run_report(access_token, start_date, end_date):
|
|
"""Query GA4 for site traffic"""
|
|
import urllib.request
|
|
|
|
url = f"https://analyticsdata.googleapis.com/v1beta/properties/{GA_PROPERTY_ID}:runReport"
|
|
|
|
# Request body
|
|
body = {
|
|
"dateRanges": [{"startDate": start_date, "endDate": end_date}],
|
|
"metrics": [
|
|
{"name": "sessions"},
|
|
{"name": "activeUsers"},
|
|
{"name": "newUsers"},
|
|
{"name": "bounceRate"},
|
|
{"name": "averageSessionDuration"}
|
|
],
|
|
"dimensions": [{"name": "date"}]
|
|
}
|
|
|
|
req = urllib.request.Request(
|
|
url,
|
|
data=json.dumps(body).encode(),
|
|
headers={
|
|
"Authorization": f"Bearer {access_token}",
|
|
"Content-Type": "application/json"
|
|
},
|
|
method="POST"
|
|
)
|
|
|
|
try:
|
|
with urllib.request.urlopen(req, timeout=30) as r:
|
|
return json.loads(r.read().decode())
|
|
except Exception as e:
|
|
return {"error": str(e)}
|
|
|
|
def get_yesterday_traffic():
|
|
"""Get yesterday's traffic summary"""
|
|
yesterday = (datetime.now() - timedelta(days=1)).strftime('%Y-%m-%d')
|
|
|
|
token = get_access_token()
|
|
if not token:
|
|
return None
|
|
|
|
return query_ga4_run_report(token, yesterday, yesterday)
|
|
|
|
def get_weekly_summary():
|
|
"""Get 7-day traffic summary"""
|
|
end = (datetime.now() - timedelta(days=1)).strftime('%Y-%m-%d')
|
|
start = (datetime.now() - timedelta(days=7)).strftime('%Y-%m-%d')
|
|
|
|
token = get_access_token()
|
|
if not token:
|
|
return None
|
|
|
|
return query_ga4_run_report(token, start, end)
|
|
|
|
def format_traffic_report(data):
|
|
"""Format GA4 data for reporting"""
|
|
if not data or 'error' in data:
|
|
return f"❌ GA4 Error: {data.get('error', 'Unknown error')}"
|
|
|
|
rows = data.get('rows', [])
|
|
if not rows:
|
|
return "📊 No data for period"
|
|
|
|
# Sum metrics
|
|
total_sessions = sum(int(r['metricValues'][0]['value']) for r in rows)
|
|
total_users = sum(int(r['metricValues'][1]['value']) for r in rows)
|
|
new_users = sum(int(r['metricValues'][2]['value']) for r in rows)
|
|
|
|
return f"""📊 Traffic Report
|
|
• Sessions: {total_sessions:,}
|
|
• Active Users: {total_users:,}
|
|
• New Users: {new_users:,}
|
|
• Period: {len(rows)} days"""
|
|
|
|
if __name__ == "__main__":
|
|
print("🔍 Testing GA4 Connection...")
|
|
print(f"Property ID: {GA_PROPERTY_ID}")
|
|
|
|
# Test yesterday
|
|
data = get_yesterday_traffic()
|
|
if data:
|
|
print(format_traffic_report(data))
|
|
else:
|
|
print("❌ Could not fetch data (check gcloud installation)") |