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