"""Meta Graph API connector (Facebook Page + Instagram via Facebook Login).""" from __future__ import annotations import json import logging import time import requests from social.crypto import decrypt_tokens from social.media import split_media logger = logging.getLogger(__name__) class MetaConnector: platform = "meta" GRAPH = "https://graph.facebook.com/v21.0" def publish(self, post, target) -> str: tokens = json.loads(decrypt_tokens(target.account.encrypted_tokens) or "{}") access_token = tokens.get("access_token") if not access_token: raise RuntimeError("Meta/Instagram account missing access_token") media = post.media if isinstance(post.media, list) else [] images, videos = split_media(media) if target.platform == "instagram": ig_user_id = target.account.external_id or tokens.get("ig_user_id") if not ig_user_id: raise RuntimeError("Instagram account missing ig_user_id") return self._publish_instagram( ig_user_id=ig_user_id, access_token=access_token, caption=post.body or "", images=images, videos=videos, ) page_id = target.account.external_id or tokens.get("page_id") if not page_id: raise RuntimeError("Meta account missing page_id") return self._publish_facebook( page_id=page_id, access_token=access_token, message=post.body or "", images=images, videos=videos, ) def _publish_facebook( self, *, page_id: str, access_token: str, message: str, images: list[dict], videos: list[dict], ) -> str: if videos: video = videos[0] response = requests.post( f"{self.GRAPH}/{page_id}/videos", data={ "file_url": video["url"], "description": message, "access_token": access_token, }, timeout=120, ) self._raise_graph(response, "Facebook video publish") return str(response.json().get("id") or "") if len(images) == 1: response = requests.post( f"{self.GRAPH}/{page_id}/photos", data={ "url": images[0]["url"], "caption": message, "access_token": access_token, }, timeout=60, ) self._raise_graph(response, "Facebook photo publish") return str(response.json().get("id") or response.json().get("post_id") or "") if len(images) > 1: attached = [] for image in images: resp = requests.post( f"{self.GRAPH}/{page_id}/photos", data={ "url": image["url"], "published": "false", "access_token": access_token, }, timeout=60, ) self._raise_graph(resp, "Facebook multi-photo upload") photo_id = resp.json().get("id") if photo_id: attached.append({"media_fbid": photo_id}) data = { "message": message, "access_token": access_token, } for idx, item in enumerate(attached): data[f"attached_media[{idx}]"] = json.dumps(item) response = requests.post( f"{self.GRAPH}/{page_id}/feed", data=data, timeout=60, ) self._raise_graph(response, "Facebook multi-photo feed") return str(response.json().get("id") or "") response = requests.post( f"{self.GRAPH}/{page_id}/feed", data={"message": message, "access_token": access_token}, timeout=30, ) self._raise_graph(response, "Facebook text feed") return str(response.json().get("id") or "") def _publish_instagram( self, *, ig_user_id: str, access_token: str, caption: str, images: list[dict], videos: list[dict], ) -> str: if not images and not videos: raise RuntimeError( "Instagram requires an image or video attachment (caption-only not allowed)." ) if videos: creation_id = self._ig_create_container( ig_user_id, access_token, { "media_type": "VIDEO", "video_url": videos[0]["url"], "caption": caption, }, ) self._ig_wait_container(creation_id, access_token) return self._ig_publish(ig_user_id, access_token, creation_id) if len(images) == 1: creation_id = self._ig_create_container( ig_user_id, access_token, {"image_url": images[0]["url"], "caption": caption}, ) return self._ig_publish(ig_user_id, access_token, creation_id) # Carousel: children first, then parent container. children = [] for image in images[:10]: child_id = self._ig_create_container( ig_user_id, access_token, {"image_url": image["url"], "is_carousel_item": "true"}, ) children.append(child_id) creation_id = self._ig_create_container( ig_user_id, access_token, { "media_type": "CAROUSEL", "children": ",".join(children), "caption": caption, }, ) return self._ig_publish(ig_user_id, access_token, creation_id) def _ig_create_container( self, ig_user_id: str, access_token: str, fields: dict ) -> str: data = {**fields, "access_token": access_token} response = requests.post( f"{self.GRAPH}/{ig_user_id}/media", data=data, timeout=60, ) self._raise_graph(response, "Instagram media container") creation_id = response.json().get("id") if not creation_id: raise RuntimeError("Instagram media container response missing id") return str(creation_id) def _ig_wait_container( self, creation_id: str, access_token: str, *, attempts: int = 20 ) -> None: """Poll video container until FINISHED (or fail).""" for _ in range(attempts): response = requests.get( f"{self.GRAPH}/{creation_id}", params={ "fields": "status_code,status", "access_token": access_token, }, timeout=30, ) self._raise_graph(response, "Instagram container status") status = (response.json().get("status_code") or "").upper() if status == "FINISHED": return if status in {"ERROR", "EXPIRED"}: raise RuntimeError( f"Instagram video processing failed ({status}): " f"{response.json().get('status') or ''}" ) time.sleep(3) raise RuntimeError("Instagram video still processing — try again shortly.") def _ig_publish( self, ig_user_id: str, access_token: str, creation_id: str ) -> str: response = requests.post( f"{self.GRAPH}/{ig_user_id}/media_publish", data={"creation_id": creation_id, "access_token": access_token}, timeout=60, ) self._raise_graph(response, "Instagram media_publish") return str(response.json().get("id") or "") @staticmethod def _raise_graph(response: requests.Response, label: str) -> None: if response.status_code < 400: return detail = response.text[:500] try: err = response.json().get("error") or {} detail = err.get("message") or detail except Exception: # noqa: BLE001 pass logger.warning("%s failed: %s %s", label, response.status_code, detail) raise RuntimeError(f"{label} failed ({response.status_code}): {detail}") def refresh_token(self, account) -> None: logger.info( "Meta/Instagram token refresh not yet implemented for account %s", account.pk, )