HOALedgerIQ_Website/agents/marketing-seo/scripts/ga4-direct.py

#!/usr/bin/env python3
"""Google Analytics 4 - Direct JWT Authentication (No gcloud required)"""
import json
import urllib.request
from datetime import datetime, timedelta
from pathlib import Path
import subprocess

CONFIG_DIR = Path(__file__).parent.parent / "config"
GA_CREDENTIALS = CONFIG_DIR / "ga-credentials.json"
GA_PROPERTY_ID = "526394825"

def load_credentials():
    """Load service account 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
    try:
        import jwt
        from cryptography.hazmat.primitives import serialization
        
        now = datetime.utcnow()
        
        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)
        }
        
        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()
        
        # 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")
                ]
            )
            
            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}
            
    except Exception as e:
        return {"error": str(e), "success": False}

if __name__ == "__main__":
    print("🚀 Testing GA4 Direct Connection...")
    result = query_ga4_direct()
    
    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'):,}
""")
    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")