feat: fetch profile photo, cover, post stats from Facebook Graph API during sync
Some checks are pending
NordaBiz Tests / Unit & Integration Tests (push) Waiting to run
NordaBiz Tests / E2E Tests (Playwright) (push) Blocked by required conditions
NordaBiz Tests / Smoke Tests (Production) (push) Blocked by required conditions
NordaBiz Tests / Send Failure Notification (push) Blocked by required conditions

- Add picture,cover fields to get_page_info() request
- New get_page_posts_stats() fetches post count (30d/365d) and last post date
- Set has_profile_photo, has_cover_photo, posts_count_30d/365d, last_post_date, posting_frequency_score
- Include profile photo in completeness score (5 fields × 20% = 100%)

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
This commit is contained in:
Maciej Pienczyn 2026-03-12 12:34:23 +01:00
parent ce0a6863d2
commit f8dacd264f

View File

@ -47,9 +47,40 @@ class FacebookGraphService:
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'
'fields': 'id,name,fan_count,category,link,about,website,phone,single_line_address,followers_count,picture,cover'
})
def get_page_posts_stats(self, page_id: str) -> Dict:
"""Get post counts and last post date from page feed."""
result = {'posts_30d': 0, 'posts_365d': 0, 'last_post_date': None}
now = datetime.now()
since_365d = int((now - timedelta(days=365)).timestamp())
data = self._get(f'{page_id}/feed', {
'fields': 'created_time',
'since': since_365d,
'limit': 100,
})
if not data:
return result
posts = data.get('data', [])
cutoff_30d = now - timedelta(days=30)
for post in posts:
ct = post.get('created_time', '')
if ct:
try:
post_date = datetime.fromisoformat(ct.replace('+0000', '+00:00').replace('Z', '+00:00'))
post_date_naive = post_date.replace(tzinfo=None)
result['posts_365d'] += 1
if post_date_naive >= cutoff_30d:
result['posts_30d'] += 1
if result['last_post_date'] is None or post_date_naive > result['last_post_date']:
result['last_post_date'] = post_date_naive
except (ValueError, TypeError):
result['posts_365d'] += 1
return result
def get_page_insights(self, page_id: str, days: int = 28) -> Dict:
"""Get page insights (impressions, engaged users, reactions).
@ -416,6 +447,9 @@ def sync_facebook_to_social_media(db, company_id: int) -> dict:
# 4. Fetch page insights (best-effort, may be empty)
insights = fb.get_page_insights(page_id, 28)
# 4b. Fetch post stats (best-effort)
post_stats = fb.get_page_posts_stats(page_id)
# 5. Calculate metrics
followers = page_info.get('followers_count') or page_info.get('fan_count') or 0
engaged_users = insights.get('page_engaged_users', 0)
@ -423,16 +457,18 @@ def sync_facebook_to_social_media(db, company_id: int) -> dict:
if followers > 0 and engaged_users > 0:
engagement_rate = round((engaged_users / followers) * 100, 2)
# Profile completeness: 25 points each for about, website, phone, address
# Profile completeness: 20 points each for 5 key fields
completeness = 0
if page_info.get('about'):
completeness += 25
completeness += 20
if page_info.get('website'):
completeness += 25
completeness += 20
if page_info.get('phone'):
completeness += 25
completeness += 20
if page_info.get('single_line_address'):
completeness += 25
completeness += 20
if page_info.get('picture', {}).get('data', {}).get('url'):
completeness += 20
# URL: prefer API vanity link, then existing URL, then numeric fallback
# Don't replace a vanity URL with a numeric one (e.g. facebook.com/inpipl → facebook.com/123456)
@ -475,6 +511,15 @@ def sync_facebook_to_social_media(db, company_id: int) -> dict:
if engagement_rate is not None:
csm.engagement_rate = engagement_rate
csm.profile_completeness_score = completeness
csm.has_profile_photo = bool(page_info.get('picture', {}).get('data', {}).get('url'))
csm.has_cover_photo = bool(page_info.get('cover', {}).get('source'))
csm.posts_count_30d = post_stats.get('posts_30d', 0)
csm.posts_count_365d = post_stats.get('posts_365d', 0)
if post_stats.get('last_post_date'):
csm.last_post_date = post_stats['last_post_date']
# Posting frequency score: 0-10 based on posts per month
p30 = post_stats.get('posts_30d', 0)
csm.posting_frequency_score = min(10, p30) if p30 > 0 else 0
csm.verified_at = now
csm.last_checked_at = now