From 4065d5c1e555469cd996c02d92f904b2ffd10740 Mon Sep 17 00:00:00 2001 From: Heshan Wanigasooriya Date: Wed, 24 Jun 2026 09:42:54 -0400 Subject: [PATCH] Fix Facebook video analytics ID resolution --- providers/facebook.py | 63 +++++++++++++++++++++++++------- tests/providers/test_facebook.py | 40 +++++++++++++++++++- 2 files changed, 89 insertions(+), 14 deletions(-) diff --git a/providers/facebook.py b/providers/facebook.py index ad47d05e..c09fa0f1 100644 --- a/providers/facebook.py +++ b/providers/facebook.py @@ -294,9 +294,10 @@ def _publish_text_or_link(self, access_token: str, page_id: str, content: Publis json=payload, ) data = resp.json() + post_id = self._page_scoped_post_id(page_id, data["id"]) return PublishResult( - platform_post_id=data["id"], - url=f"https://www.facebook.com/{data['id']}", + platform_post_id=post_id, + url=f"https://www.facebook.com/{post_id}", extra=data, ) @@ -314,7 +315,7 @@ def _publish_photo(self, access_token: str, page_id: str, content: PublishConten json=payload, ) data = resp.json() - post_id = data.get("post_id", data["id"]) + post_id = self._page_scoped_post_id(page_id, data["post_id"]) if data.get("post_id") else data["id"] return PublishResult(platform_post_id=post_id, url=f"https://www.facebook.com/{post_id}", extra=data) def _publish_multi_photo(self, access_token: str, page_id: str, content: PublishContent) -> PublishResult: @@ -363,7 +364,7 @@ def _publish_multi_photo(self, access_token: str, page_id: str, content: Publish access_token=access_token, json=payload, ).json() - post_id = data.get("id") + post_id = self._page_scoped_post_id(page_id, data.get("id")) if not post_id: raise PublishError( "Failed to publish Facebook multi-photo post", @@ -433,7 +434,7 @@ def _publish_video(self, access_token: str, page_id: str, content: PublishConten # body making .json() raise). Catch broadly; fall back to video_id. logger.debug("Facebook video %s post_id unavailable: %s", video_id, exc) - post_id = video_fields.get("post_id") or video_id + post_id = self._page_scoped_post_id(page_id, video_fields["post_id"]) if video_fields.get("post_id") else video_id url = video_fields.get("permalink_url") or f"https://www.facebook.com/{post_id}" return PublishResult( platform_post_id=post_id, @@ -490,19 +491,29 @@ 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) + page_id = self.credentials.get("page_id") + primary_post_id = self._page_scoped_post_id(page_id, post_id) + fields = self._get_post_fields(access_token, primary_post_id) + if primary_post_id != str(post_id) and not fields: + logger.warning( + "Facebook post fields unavailable for page-scoped id %s; falling back to stored id %s", + primary_post_id, + post_id, + ) + fields = self._get_post_fields(access_token, 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}") + scoped_candidate_id = self._page_scoped_post_id(page_id, candidate_id) + if scoped_candidate_id: + candidate_ids.append(scoped_candidate_id) candidate_ids.append(candidate_id) - if page_id and "_" not in str(post_id): - candidate_ids.append(f"{page_id}_{post_id}") + if primary_post_id: + candidate_ids.append(primary_post_id) candidate_ids.append(post_id) deduped_candidate_ids = list(dict.fromkeys(str(candidate_id) for candidate_id in candidate_ids if candidate_id)) @@ -515,6 +526,12 @@ def _resolve_post_fields(self, access_token: str, post_id: str) -> tuple[dict, l if feed_fields: best_fields = {**fields, **video_fields, **feed_fields} break + if not best_fields: + logger.warning( + "Facebook post fields unavailable for stored id %s after trying candidates: %s", + post_id, + ", ".join(deduped_candidate_ids), + ) return best_fields, deduped_candidate_ids def _get_video_post_metadata(self, access_token: str, video_id: str) -> dict: @@ -532,9 +549,18 @@ def _get_video_post_metadata(self, access_token: str, video_id: str) -> dict: 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) + logger.warning("Facebook video %s post_id unavailable during analytics: %s", video_id, exc) return {} + @staticmethod + def _page_scoped_post_id(page_id: str | None, post_id: str | None) -> str: + if not post_id: + return "" + post_id = str(post_id) + if page_id and "_" not in post_id: + return f"{page_id}_{post_id}" + return post_id + 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] = {} @@ -547,11 +573,15 @@ def _get_post_insights(self, access_token: str, post_ids: list[str]) -> tuple[di access_token=access_token, params={"metric": metric}, ) + return parse_insights_response(resp.json()), {}, post_id except APIError as exc: errors_by_post_id[post_id] = str(exc) logger.warning("Skipping unsupported Facebook post insights at %s: %s", endpoint, exc) continue - return parse_insights_response(resp.json()), {}, post_id + except Exception as exc: + errors_by_post_id[post_id] = str(exc) + logger.warning("Facebook post insights failed at %s: %s", endpoint, exc) + continue error_text = "; ".join(f"{post_id}: {error}" for post_id, error in errors_by_post_id.items()) return {}, {key: error_text for key in FACEBOOK_POST_INSIGHTS}, post_ids[0] if post_ids else "" @@ -562,6 +592,7 @@ def _get_post_fields(self, access_token: str, post_id: str) -> dict: # invalid-field error. Retry without ``post_id`` so the engagement fields # (comments/shares/reactions summaries) still load for normal posts. field_sets = (FACEBOOK_POST_FIELDS, [f for f in FACEBOOK_POST_FIELDS if f != "post_id"]) + last_error: Exception | None = None for fields in field_sets: try: fields_resp = self._request( @@ -572,12 +603,18 @@ def _get_post_fields(self, access_token: str, post_id: str) -> dict: ) return fields_resp.json() except APIError as exc: + last_error = exc logger.debug("Facebook post %s fields unavailable: %s", post_id, exc) if "post_id" not in str(exc): # Only the invalid-`post_id`-field error is worth retrying # without it; other errors (auth, 5xx, not-found) fail # identically, so don't issue a second doomed request. break + except Exception as exc: + last_error = exc + break + if last_error: + logger.warning("Facebook post %s fields unavailable: %s", post_id, last_error) return {} @staticmethod diff --git a/tests/providers/test_facebook.py b/tests/providers/test_facebook.py index 043c706a..4692472c 100644 --- a/tests/providers/test_facebook.py +++ b/tests/providers/test_facebook.py @@ -137,6 +137,43 @@ def test_publish_single_photo_uses_photos_edge_without_staging(): assert "attached_media" not in sent +def test_publish_text_scopes_bare_feed_post_id_to_page(): + provider = FacebookProvider({"client_id": "id", "client_secret": "secret"}) + provider._request = MagicMock(return_value=_resp({"id": "post-1"})) + + result = provider.publish_post( + "page-token", + PublishContent(text="Caption", post_type=PostType.TEXT, extra={"page_id": "page-1"}), + ) + + assert result.platform_post_id == "page-1_post-1" + assert result.url == "https://www.facebook.com/page-1_post-1" + + +def test_publish_video_scopes_bare_feed_post_id_to_page(): + provider = FacebookProvider({"client_id": "id", "client_secret": "secret"}) + provider._request = MagicMock( + side_effect=[ + _resp({"id": "video-1"}), + _resp({"post_id": "post-1", "permalink_url": "https://www.facebook.com/page-1/videos/video-1/"}), + ] + ) + + result = provider.publish_post( + "page-token", + PublishContent( + text="Video caption", + media_urls=["https://cdn.example.com/clip.mp4"], + post_type=PostType.VIDEO, + extra={"page_id": "page-1"}, + ), + ) + + assert result.platform_post_id == "page-1_post-1" + assert result.url == "https://www.facebook.com/page-1/videos/video-1/" + assert result.extra["video_id"] == "video-1" + + def test_is_video_url_ignores_query_string(): """Presigned URLs carry query strings; the check must look at the path only.""" assert FacebookProvider._is_video_url("https://cdn.example.com/clip.mp4?X-Amz-Sig=abc&x=1") is True @@ -485,6 +522,7 @@ 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("(#10) Object does not exist", platform="Facebook"), APIError("(#100) Tried accessing nonexisting field (message)", platform="Facebook"), _resp( { @@ -524,7 +562,7 @@ def test_get_post_metrics_resolves_bare_video_id_to_exact_feed_post_id(): 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", + "https://graph.facebook.com/v25.0/page-1_video-1", access_token="page-token", params={"fields": FACEBOOK_POST_FIELDS_PARAM}, )