""" Facebook + Instagram Graph API Client ====================================== Uses OAuth 2.0 page tokens to access Facebook Page and Instagram Business data. API docs: https://developers.facebook.com/docs/graph-api/ """ import logging from datetime import datetime, timedelta from typing import Dict, List, Optional import requests logger = logging.getLogger(__name__) class FacebookGraphService: """Facebook + Instagram Graph API client.""" BASE_URL = "https://graph.facebook.com/v21.0" def __init__(self, access_token: str): self.access_token = access_token self.session = requests.Session() self.session.timeout = 15 def _get(self, endpoint: str, params: dict = None) -> Optional[Dict]: """Make authenticated GET request.""" params = params or {} params['access_token'] = self.access_token try: resp = self.session.get(f"{self.BASE_URL}/{endpoint}", params=params) resp.raise_for_status() return resp.json() except Exception as e: logger.error(f"Facebook API {endpoint} failed: {e}") return None def get_managed_pages(self) -> List[Dict]: """Get Facebook pages managed by the authenticated user.""" data = self._get('me/accounts', {'fields': 'id,name,category,access_token,fan_count'}) return data.get('data', []) if data else [] def get_page_info(self, page_id: str) -> Optional[Dict]: """Get detailed page information.""" return self._get(page_id, { 'fields': 'id,name,fan_count,category,link,about,website,phone,single_line_address,followers_count' }) def get_page_insights(self, page_id: str, days: int = 28) -> Dict: """Get page insights (impressions, engaged users, reactions). Note: Requires page access token, not user access token. The page token should be stored during OAuth connection. """ since = datetime.now() - timedelta(days=days) until = datetime.now() metrics = 'page_impressions,page_engaged_users,page_fans,page_views_total' data = self._get(f'{page_id}/insights', { 'metric': metrics, 'period': 'day', 'since': int(since.timestamp()), 'until': int(until.timestamp()), }) if not data: return {} result = {} for metric in data.get('data', []): name = metric.get('name', '') values = metric.get('values', []) if values: total = sum(v.get('value', 0) for v in values if isinstance(v.get('value'), (int, float))) result[name] = total return result def get_instagram_account(self, page_id: str) -> Optional[str]: """Get linked Instagram Business account ID from a Facebook Page.""" data = self._get(page_id, {'fields': 'instagram_business_account'}) if data and 'instagram_business_account' in data: return data['instagram_business_account'].get('id') return None def get_ig_media_insights(self, ig_account_id: str, days: int = 28) -> Dict: """Get Instagram account insights. Returns follower_count, media_count, and recent media engagement. """ result = {} # Basic account info account_data = self._get(ig_account_id, { 'fields': 'followers_count,media_count,username,biography' }) if account_data: result['followers_count'] = account_data.get('followers_count', 0) result['media_count'] = account_data.get('media_count', 0) result['username'] = account_data.get('username', '') # Account insights (reach, impressions) since = datetime.now() - timedelta(days=days) until = datetime.now() insights_data = self._get(f'{ig_account_id}/insights', { 'metric': 'impressions,reach,follower_count', 'period': 'day', 'since': int(since.timestamp()), 'until': int(until.timestamp()), }) if insights_data: for metric in insights_data.get('data', []): name = metric.get('name', '') values = metric.get('values', []) if values: total = sum(v.get('value', 0) for v in values if isinstance(v.get('value'), (int, float))) result[f'ig_{name}_total'] = total return result