nordabiz/gbp_management_service.py

"""
Google Business Profile Management API Client
==============================================

Uses OAuth 2.0 access tokens to access owner-specific GBP data
not available through the public Places API (e.g., owner responses to reviews, posts).

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

import logging
from typing import List, Dict, Optional

import requests

logger = logging.getLogger(__name__)


class GBPManagementService:
    """Google Business Profile Management API client."""

    ACCOUNT_MGMT_URL = "https://mybusinessaccountmanagement.googleapis.com/v1"
    BUSINESS_INFO_URL = "https://mybusinessbusinessinformation.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 = 15

    def list_accounts(self) -> List[Dict]:
        """List all GBP accounts for the authenticated user."""
        try:
            resp = self.session.get(f"{self.ACCOUNT_MGMT_URL}/accounts")
            resp.raise_for_status()
            return resp.json().get('accounts', [])
        except Exception as e:
            logger.error(f"GBP list_accounts failed: {e}")
            return []

    def list_locations(self, account_name: str) -> List[Dict]:
        """List locations for a GBP account.

        Args:
            account_name: Full account resource name (e.g., 'accounts/123456')
        """
        try:
            resp = self.session.get(
                f"{self.BUSINESS_INFO_URL}/{account_name}/locations",
                params={'readMask': 'name,title,storefrontAddress,metadata'}
            )
            resp.raise_for_status()
            return resp.json().get('locations', [])
        except Exception as e:
            logger.error(f"GBP list_locations failed for {account_name}: {e}")
            return []

    def get_reviews(self, location_name: str, max_results: int = 50) -> List[Dict]:
        """Get reviews INCLUDING owner responses (reviewReply).

        This is the key advantage over Places API — reviewReply contains
        the business owner's response to each review.

        Args:
            location_name: Full location resource name (e.g., 'accounts/123/locations/456')
            max_results: Max reviews to return
        """
        try:
            reviews = []
            page_token = None
            while len(reviews) < max_results:
                params = {'pageSize': min(50, max_results - len(reviews))}
                if page_token:
                    params['pageToken'] = page_token
                resp = self.session.get(
                    f"{self.ACCOUNT_MGMT_URL}/{location_name}/reviews",
                    params=params
                )
                resp.raise_for_status()
                data = resp.json()
                reviews.extend(data.get('reviews', []))
                page_token = data.get('nextPageToken')
                if not page_token:
                    break
            return reviews
        except Exception as e:
            logger.error(f"GBP get_reviews failed for {location_name}: {e}")
            return []

    def get_local_posts(self, location_name: str) -> List[Dict]:
        """Get Google Posts for a location."""
        try:
            resp = self.session.get(
                f"{self.ACCOUNT_MGMT_URL}/{location_name}/localPosts",
                params={'pageSize': 20}
            )
            resp.raise_for_status()
            return resp.json().get('localPosts', [])
        except Exception as e:
            logger.error(f"GBP get_local_posts failed for {location_name}: {e}")
            return []

    def match_location_by_place_id(self, account_name: str, place_id: str) -> Optional[Dict]:
        """Find a location that matches a Google Place ID.

        Iterates locations and checks metadata.placeId.
        """
        locations = self.list_locations(account_name)
        for loc in locations:
            metadata = loc.get('metadata', {})
            if metadata.get('placeId') == place_id:
                return loc
        return None