Maciej Pienczyn
4a033f0d81
Admin panel module for publishing posts on NORDA chamber Facebook page. Includes AI content generation (Gemini), post workflow (draft/approved/ scheduled/published), Facebook Graph API publishing, and engagement tracking. New: migration 070, SocialPost/SocialMediaConfig models, publisher service, admin routes with AJAX, 3 templates (list/form/settings). Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
276 lines
10 KiB
Python
276 lines
10 KiB
Python
"""
|
|
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 json
|
|
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
|
|
|
|
# ============================================================
|
|
# PUBLISHING METHODS (Social Publisher)
|
|
# ============================================================
|
|
|
|
def _post(self, endpoint: str, data: dict = None, files: dict = None) -> Optional[Dict]:
|
|
"""Make authenticated POST request."""
|
|
data = data or {}
|
|
params = {'access_token': self.access_token}
|
|
try:
|
|
if files:
|
|
resp = self.session.post(f"{self.BASE_URL}/{endpoint}", params=params, data=data, files=files)
|
|
else:
|
|
resp = self.session.post(f"{self.BASE_URL}/{endpoint}", params=params, data=data)
|
|
resp.raise_for_status()
|
|
return resp.json()
|
|
except requests.exceptions.HTTPError as e:
|
|
error_data = None
|
|
try:
|
|
error_data = e.response.json()
|
|
except Exception:
|
|
pass
|
|
logger.error(f"Facebook API POST {endpoint} failed: {e}, response: {error_data}")
|
|
return None
|
|
except Exception as e:
|
|
logger.error(f"Facebook API POST {endpoint} failed: {e}")
|
|
return None
|
|
|
|
def _delete(self, endpoint: str) -> Optional[Dict]:
|
|
"""Make authenticated DELETE request."""
|
|
params = {'access_token': self.access_token}
|
|
try:
|
|
resp = self.session.delete(f"{self.BASE_URL}/{endpoint}", params=params)
|
|
resp.raise_for_status()
|
|
return resp.json()
|
|
except Exception as e:
|
|
logger.error(f"Facebook API DELETE {endpoint} failed: {e}")
|
|
return None
|
|
|
|
def upload_photo_unpublished(self, page_id: str, image_path: str) -> Optional[str]:
|
|
"""Upload photo as unpublished to get photo_id for feed attachment.
|
|
|
|
Two-step process:
|
|
1. Upload photo with published=false -> get photo_id
|
|
2. Use photo_id in create_post() attached_media
|
|
|
|
Args:
|
|
page_id: Facebook Page ID
|
|
image_path: Local path to image file
|
|
|
|
Returns:
|
|
Photo ID (media_fbid) or None on failure
|
|
"""
|
|
try:
|
|
with open(image_path, 'rb') as f:
|
|
files = {'source': (image_path.split('/')[-1], f, 'image/png')}
|
|
result = self._post(f'{page_id}/photos', data={'published': 'false'}, files=files)
|
|
if result and 'id' in result:
|
|
logger.info(f"Uploaded unpublished photo: {result['id']}")
|
|
return result['id']
|
|
logger.error(f"Photo upload failed: {result}")
|
|
return None
|
|
except FileNotFoundError:
|
|
logger.error(f"Image file not found: {image_path}")
|
|
return None
|
|
|
|
def create_post(self, page_id: str, message: str, image_path: str = None,
|
|
published: bool = True, scheduled_time: int = None) -> Optional[Dict]:
|
|
"""Create a post on Facebook Page feed.
|
|
|
|
For posts with images, uses two-step process:
|
|
1. Upload photo as unpublished -> photo_id
|
|
2. Create feed post with attached_media[0]
|
|
|
|
Args:
|
|
page_id: Facebook Page ID
|
|
message: Post text content
|
|
image_path: Optional local path to image
|
|
published: If False, creates draft (visible only to page admins)
|
|
scheduled_time: Unix timestamp for scheduled publishing (min 10 min ahead)
|
|
|
|
Returns:
|
|
API response dict with 'id' key, or None on failure
|
|
"""
|
|
data = {'message': message}
|
|
|
|
# Handle image upload
|
|
if image_path:
|
|
photo_id = self.upload_photo_unpublished(page_id, image_path)
|
|
if photo_id:
|
|
data['attached_media[0]'] = json.dumps({'media_fbid': photo_id})
|
|
else:
|
|
logger.warning("Image upload failed, posting without image")
|
|
|
|
# Handle scheduling
|
|
if scheduled_time:
|
|
data['published'] = 'false'
|
|
data['scheduled_publish_time'] = str(scheduled_time)
|
|
elif not published:
|
|
data['published'] = 'false'
|
|
|
|
result = self._post(f'{page_id}/feed', data=data)
|
|
if result and 'id' in result:
|
|
logger.info(f"Created post: {result['id']} (published={published})")
|
|
return result
|
|
|
|
def publish_draft(self, post_id: str) -> Optional[Dict]:
|
|
"""Publish an unpublished (draft) post.
|
|
|
|
Args:
|
|
post_id: Facebook post ID (format: PAGE_ID_POST_ID)
|
|
|
|
Returns:
|
|
API response or None on failure
|
|
"""
|
|
return self._post(post_id, data={'is_published': 'true'})
|
|
|
|
def get_post_engagement(self, post_id: str) -> Optional[Dict]:
|
|
"""Get engagement metrics for a published post.
|
|
|
|
Args:
|
|
post_id: Facebook post ID
|
|
|
|
Returns:
|
|
Dict with likes, comments, shares, reactions_total or None
|
|
"""
|
|
fields = 'id,created_time,message,likes.summary(true),comments.summary(true),shares,reactions.summary(true).limit(0)'
|
|
result = self._get(post_id, {'fields': fields})
|
|
if not result:
|
|
return None
|
|
|
|
return {
|
|
'post_id': result.get('id'),
|
|
'created_time': result.get('created_time'),
|
|
'likes': result.get('likes', {}).get('summary', {}).get('total_count', 0),
|
|
'comments': result.get('comments', {}).get('summary', {}).get('total_count', 0),
|
|
'shares': result.get('shares', {}).get('count', 0) if result.get('shares') else 0,
|
|
'reactions_total': result.get('reactions', {}).get('summary', {}).get('total_count', 0),
|
|
}
|
|
|
|
def delete_post(self, post_id: str) -> bool:
|
|
"""Delete a post (published or unpublished).
|
|
|
|
Args:
|
|
post_id: Facebook post ID
|
|
|
|
Returns:
|
|
True if deleted successfully
|
|
"""
|
|
result = self._delete(post_id)
|
|
return bool(result and result.get('success'))
|