|
|
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): |
|
|
|
|
|
|
|
|
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): |
|
|
|
|
|
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)} |
|
|
|
