nordabiz/gbp_performance_service.py

"""
Google Business Profile Performance API Client
===============================================

Uses OAuth 2.0 to fetch visibility and interaction metrics from GBP.
Provides daily metrics time series and monthly search keyword impressions.

API docs: https://developers.google.com/my-business/reference/performance/rest
"""

import logging
from datetime import datetime, timedelta
from typing import Dict, List, Optional

import requests

logger = logging.getLogger(__name__)


# Metrics to request from getDailyMetricsTimeSeries
DAILY_METRICS = [
    'BUSINESS_IMPRESSIONS_DESKTOP_MAPS',
    'BUSINESS_IMPRESSIONS_MOBILE_MAPS',
    'BUSINESS_IMPRESSIONS_DESKTOP_SEARCH',
    'BUSINESS_IMPRESSIONS_MOBILE_SEARCH',
    'CALL_CLICKS',
    'WEBSITE_CLICKS',
    'BUSINESS_DIRECTION_REQUESTS',
    'BUSINESS_CONVERSATIONS',
    'BUSINESS_BOOKINGS',
]


class GBPPerformanceService:
    """Google Business Profile Performance API client."""

    BASE_URL = "https://businessprofileperformance.googleapis.com/v1"

    def __init__(self, access_token: str):
        self.session = requests.Session()
        self.session.headers.update({
            'Authorization': f'Bearer {access_token}',
            'Content-Type': 'application/json',
        })
        self.session.timeout = 20

    def get_daily_metrics(self, location_name: str, days: int = 30) -> Dict:
        """Get aggregated daily metrics for a location.

        Args:
            location_name: Full location resource name (e.g., 'locations/123456')
            days: Number of days to aggregate (default 30)

        Returns:
            Dict with aggregated metrics:
            {
                'maps_impressions': int,
                'search_impressions': int,
                'call_clicks': int,
                'website_clicks': int,
                'direction_requests': int,
                'conversations': int,
                'period_days': int,
            }
        """
        result = {
            'maps_impressions': 0,
            'search_impressions': 0,
            'call_clicks': 0,
            'website_clicks': 0,
            'direction_requests': 0,
            'conversations': 0,
            'period_days': days,
        }

        end_date = datetime.now() - timedelta(days=1)  # Yesterday (today may be incomplete)
        start_date = end_date - timedelta(days=days)

        # Normalize location_name format
        if not location_name.startswith('locations/'):
            location_name = f'locations/{location_name}'

        for metric in DAILY_METRICS:
            try:
                resp = self.session.get(
                    f"{self.BASE_URL}/{location_name}:getDailyMetricsTimeSeries",
                    params={
                        'dailyMetric': metric,
                        'dailyRange.startDate.year': start_date.year,
                        'dailyRange.startDate.month': start_date.month,
                        'dailyRange.startDate.day': start_date.day,
                        'dailyRange.endDate.year': end_date.year,
                        'dailyRange.endDate.month': end_date.month,
                        'dailyRange.endDate.day': end_date.day,
                    }
                )

                if resp.status_code != 200:
                    logger.debug(f"Performance API metric {metric} returned {resp.status_code}")
                    continue

                data = resp.json()
                time_series = data.get('timeSeries', {})
                daily_values = time_series.get('datedValues', [])

                total = sum(
                    dv.get('value', 0) for dv in daily_values
                    if dv.get('value') is not None
                )

                # Map metric to result key
                if 'MAPS' in metric:
                    result['maps_impressions'] += total
                elif 'SEARCH' in metric:
                    result['search_impressions'] += total
                elif metric == 'CALL_CLICKS':
                    result['call_clicks'] = total
                elif metric == 'WEBSITE_CLICKS':
                    result['website_clicks'] = total
                elif metric == 'BUSINESS_DIRECTION_REQUESTS':
                    result['direction_requests'] = total
                elif metric == 'BUSINESS_CONVERSATIONS':
                    result['conversations'] = total

            except Exception as e:
                logger.debug(f"Performance API metric {metric} failed: {e}")
                continue

        logger.info(
            f"GBP Performance for {location_name}: "
            f"maps={result['maps_impressions']}, search={result['search_impressions']}, "
            f"calls={result['call_clicks']}, web={result['website_clicks']}"
        )

        return result

    def get_search_keywords(self, location_name: str) -> List[Dict]:
        """Get monthly search keyword impressions.

        Args:
            location_name: Full location resource name

        Returns:
            List of dicts: [{'keyword': str, 'impressions': int}, ...]
        """
        # Normalize location_name format
        if not location_name.startswith('locations/'):
            location_name = f'locations/{location_name}'

        try:
            resp = self.session.get(
                f"{self.BASE_URL}/{location_name}/searchkeywords/impressions/monthly",
                params={'pageSize': 20}
            )

            if resp.status_code != 200:
                logger.debug(f"Search keywords API returned {resp.status_code}")
                return []

            data = resp.json()
            keywords = []

            for item in data.get('searchKeywordsCounts', []):
                keyword = item.get('searchKeyword', '')
                # Get the most recent month's data
                monthly = item.get('insightsValue', {})
                impressions = monthly.get('value', 0)

                if keyword:
                    keywords.append({
                        'keyword': keyword,
                        'impressions': impressions,
                    })

            # Sort by impressions descending
            keywords.sort(key=lambda x: x['impressions'], reverse=True)

            logger.info(f"GBP search keywords for {location_name}: {len(keywords)} keywords")
            return keywords[:20]

        except Exception as e:
            logger.warning(f"GBP search keywords failed for {location_name}: {e}")
            return []

    def get_all_performance_data(self, location_name: str, days: int = 30) -> Dict:
        """Get all performance data (metrics + keywords) in one call.

        Returns:
            Dict with all performance data combined.
        """
        metrics = self.get_daily_metrics(location_name, days)
        keywords = self.get_search_keywords(location_name)
        metrics['search_keywords'] = keywords
        return metrics