Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
30 changes: 28 additions & 2 deletions providers/facebook.py
Original file line number Diff line number Diff line change
Expand Up @@ -491,8 +491,16 @@ def get_post_metrics(self, access_token: str, post_id: str) -> PostMetrics:

def _resolve_post_fields(self, access_token: str, post_id: str) -> tuple[dict, list[str]]:
fields = self._get_post_fields(access_token, post_id)
candidate_ids = [fields.get("post_id")]
video_fields: dict = {}
if not fields and "_" not in str(post_id):
video_fields = self._get_video_post_metadata(access_token, post_id)

page_id = self.credentials.get("page_id")
candidate_ids = []
for candidate_id in (fields.get("post_id"), video_fields.get("post_id")):
if page_id and candidate_id and "_" not in str(candidate_id):
candidate_ids.append(f"{page_id}_{candidate_id}")
candidate_ids.append(candidate_id)
if page_id and "_" not in str(post_id):
candidate_ids.append(f"{page_id}_{post_id}")
candidate_ids.append(post_id)
Expand All @@ -505,10 +513,28 @@ def _resolve_post_fields(self, access_token: str, post_id: str) -> tuple[dict, l
continue
feed_fields = self._get_post_fields(access_token, candidate_id)
if feed_fields:
best_fields = {**fields, **feed_fields}
best_fields = {**fields, **video_fields, **feed_fields}
break
return best_fields, deduped_candidate_ids

def _get_video_post_metadata(self, access_token: str, video_id: str) -> dict:
"""Resolve a bare Facebook video object ID to its feed post ID.

Legacy rows can store the video node ID when the post-publish metadata
lookup was not ready. Full post fields such as ``message`` are invalid
on that video node, so ask only for video-supported metadata first.
"""
try:
return self._request(
"GET",
f"{BASE_URL}/{video_id}",
access_token=access_token,
params={"fields": "post_id,permalink_url"},
).json()
except Exception as exc:
logger.debug("Facebook video %s post_id unavailable during analytics: %s", video_id, exc)
return {}

def _get_post_insights(self, access_token: str, post_ids: list[str]) -> tuple[dict, dict[str, str], str]:
metric = ",".join(FACEBOOK_POST_INSIGHTS)
errors_by_post_id: dict[str, str] = {}
Expand Down
67 changes: 67 additions & 0 deletions tests/providers/test_facebook.py
Original file line number Diff line number Diff line change
Expand Up @@ -481,6 +481,73 @@ def test_get_post_metrics_tries_page_scoped_feed_id_for_numeric_object_id():
)


def test_get_post_metrics_resolves_bare_video_id_to_exact_feed_post_id():
provider = FacebookProvider({"client_id": "id", "client_secret": "secret", "page_id": "page-1"})
provider._request = MagicMock(
side_effect=[
APIError("(#100) Tried accessing nonexisting field (message)", platform="Facebook"),
_resp(
{
"id": "video-1",
"post_id": "post-2",
"permalink_url": "https://www.facebook.com/page-1/videos/video-1/",
}
),
_resp(
{
"id": "page-1_post-2",
"shares": {"count": 8},
"comments": {"summary": {"total_count": 6}},
"reactions": {"summary": {"total_count": 3}},
}
),
_resp(
{
"data": [
{"name": "post_media_view", "values": [{"value": 90}]},
{"name": "post_total_media_view_unique", "values": [{"value": 70}]},
{"name": "post_clicks", "values": [{"value": 5}]},
{"name": "post_reactions_by_type_total", "values": [{"value": {"like": 3}}]},
]
}
),
]
)

metrics = provider.get_post_metrics("page-token", "video-1")

assert metrics.video_views == 90
assert metrics.reach == 70
assert metrics.comments == 6
assert metrics.shares == 8
assert metrics.extra["insight_post_id"] == "page-1_post-2"
assert metrics.extra["attempted_insight_post_ids"] == ["page-1_post-2", "post-2", "page-1_video-1", "video-1"]
provider._request.assert_any_call(
"GET",
"https://graph.facebook.com/v25.0/video-1",
access_token="page-token",
params={"fields": FACEBOOK_POST_FIELDS_PARAM},
)
provider._request.assert_any_call(
"GET",
"https://graph.facebook.com/v25.0/video-1",
access_token="page-token",
params={"fields": "post_id,permalink_url"},
)
provider._request.assert_any_call(
"GET",
"https://graph.facebook.com/v25.0/page-1_post-2",
access_token="page-token",
params={"fields": FACEBOOK_POST_FIELDS_PARAM},
)
provider._request.assert_any_call(
"GET",
"https://graph.facebook.com/v25.0/page-1_post-2/insights",
access_token="page-token",
params={"metric": FACEBOOK_POST_INSIGHTS_PARAM},
)


def test_get_post_metrics_tries_next_candidate_when_feed_id_has_no_insights_edge():
provider = FacebookProvider({"client_id": "id", "client_secret": "secret", "page_id": "page-1"})
provider._request = MagicMock(
Expand Down
Loading