web_django_template/site/social/connectors/meta.py

"""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,
        )