File size: 4,475 Bytes
e5ab217 |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 |
from google.analytics.data_v1beta import BetaAnalyticsDataClient
from google.analytics.data_v1beta.types import RunReportRequest, DateRange, Metric, Dimension
from google.oauth2 import service_account
from googleapiclient.discovery import build
import pandas as pd
from datetime import datetime, timedelta
import requests
import logging
import os
logger = logging.getLogger(__name__)
class PerformanceAnalyticsAgent:
def __init__(self, ga4_property_id, ga4_credentials_path, gsc_credentials_path, site_url):
self.property_id = ga4_property_id
self.site_url = site_url
self.ga4_client = None
self.gsc = None
if os.path.exists(ga4_credentials_path):
try:
creds = service_account.Credentials.from_service_account_file(ga4_credentials_path)
self.ga4_client = BetaAnalyticsDataClient(credentials=creds)
except Exception as e: logger.error(f"GA4 init failed: {e}")
if os.path.exists(gsc_credentials_path):
try:
creds = service_account.Credentials.from_service_account_file(gsc_credentials_path)
self.gsc = build('searchconsole', 'v1', credentials=creds)
except Exception as e: logger.error(f"GSC init failed: {e}")
def get_underperforming_pages(self, days=30):
if not self.gsc: return []
try:
response = self.gsc.searchanalytics().query(
siteUrl=self.site_url,
body={
'startDate': (datetime.now() - timedelta(days=days)).strftime('%Y-%m-%d'),
'endDate': datetime.now().strftime('%Y-%m-%d'),
'dimensions': ['page'],
'rowLimit': 100
}
).execute()
underperforming = []
for row in response.get('rows', []):
ctr = row['ctr']
if row['impressions'] > 100 and ctr < 0.02:
underperforming.append({
'url': row['keys'][0],
'impressions': row['impressions'],
'ctr': ctr
})
return underperforming
except Exception as e:
logger.error(f"GSC query failed: {e}")
return []
def detect_algorithm_update(self):
# Simplified volatility check
# In production this would compare day-over-day ranking distributions
return {
'volatility_detected': False,
'avg_position_change': 0.0
}
def generate_weekly_report(self):
report = {
'summary': {'total_pageviews': 0},
'top_pages': [],
'underperforming_pages': []
}
if self.ga4_client:
try:
request = RunReportRequest(
property=f"properties/{self.property_id}",
date_ranges=[DateRange(start_date="7daysAgo", end_date="today")],
dimensions=[Dimension(name="pagePath")],
metrics=[Metric(name="screenPageViews")]
)
response = self.ga4_client.run_report(request)
total_pv = 0
for row in response.rows:
pv = int(row.metric_values[0].value)
total_pv += pv
report['top_pages'].append({
'path': row.dimension_values[0].value,
'views': pv
})
report['summary']['total_pageviews'] = total_pv
except Exception as e:
logger.error(f"GA4 report failed: {e}")
return report
def monitor_core_web_vitals(self, url):
# Uses public PageSpeed Insights API
try:
psi_url = "https://www.googleapis.com/pagespeedonline/v5/runPagespeed"
params = {'url': url, 'strategy': 'mobile', 'category': ['performance', 'seo']}
resp = requests.get(psi_url, params=params)
data = resp.json()
lighthouse = data.get('lighthouseResult', {})
return {
'url': url,
'performance_score': lighthouse.get('categories', {}).get('performance', {}).get('score')
}
except Exception as e:
logger.error(f"CWV check failed: {e}")
return {'error': str(e)}
|