generated from westfarn/web_django_template
251 lines
8.5 KiB
Python
251 lines
8.5 KiB
Python
"""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,
|
|
)
|