89 lines
3.1 KiB
Python
89 lines
3.1 KiB
Python
import logging
|
|
from typing import Optional
|
|
|
|
import httpx
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
META_GRAPH_VERSION = "v18.0"
|
|
META_GRAPH_BASE = f"https://graph.facebook.com/{META_GRAPH_VERSION}"
|
|
|
|
|
|
class MetaAdapter:
|
|
"""Adapter for Meta Platform API (Instagram Graph API).
|
|
|
|
Requires:
|
|
- access_token: long-lived Page or Instagram access token
|
|
- instagram_account_id: Instagram Business Account ID
|
|
"""
|
|
|
|
def __init__(self, access_token: str, instagram_account_id: str):
|
|
self.access_token = access_token
|
|
self.instagram_account_id = instagram_account_id
|
|
|
|
async def post_to_feed(self, image_url: str, caption: str) -> Optional[str]:
|
|
"""Upload image and publish to Instagram feed.
|
|
|
|
Returns the post ID on success, raises on failure.
|
|
"""
|
|
async with httpx.AsyncClient(timeout=30.0) as client:
|
|
# Step 1: create media container
|
|
resp = await client.post(
|
|
f"{META_GRAPH_BASE}/{self.instagram_account_id}/media",
|
|
data={
|
|
"image_url": image_url,
|
|
"caption": caption,
|
|
"access_token": self.access_token,
|
|
},
|
|
)
|
|
resp.raise_for_status()
|
|
creation_id = resp.json().get("id")
|
|
if not creation_id:
|
|
raise ValueError(f"No creation_id from Meta API: {resp.text}")
|
|
|
|
# Step 2: publish
|
|
resp2 = await client.post(
|
|
f"{META_GRAPH_BASE}/{self.instagram_account_id}/media_publish",
|
|
data={
|
|
"creation_id": creation_id,
|
|
"access_token": self.access_token,
|
|
},
|
|
)
|
|
resp2.raise_for_status()
|
|
post_id = resp2.json().get("id")
|
|
logger.info(f"Published to Instagram feed: {post_id}")
|
|
return post_id
|
|
|
|
async def post_to_story(self, image_url: str) -> Optional[str]:
|
|
"""Upload image and publish to Instagram story.
|
|
|
|
Returns the story ID on success, raises on failure.
|
|
"""
|
|
async with httpx.AsyncClient(timeout=30.0) as client:
|
|
# Step 1: create story container
|
|
resp = await client.post(
|
|
f"{META_GRAPH_BASE}/{self.instagram_account_id}/media",
|
|
data={
|
|
"image_url": image_url,
|
|
"media_type": "STORIES",
|
|
"access_token": self.access_token,
|
|
},
|
|
)
|
|
resp.raise_for_status()
|
|
creation_id = resp.json().get("id")
|
|
if not creation_id:
|
|
raise ValueError(f"No creation_id from Meta API: {resp.text}")
|
|
|
|
# Step 2: publish
|
|
resp2 = await client.post(
|
|
f"{META_GRAPH_BASE}/{self.instagram_account_id}/media_publish",
|
|
data={
|
|
"creation_id": creation_id,
|
|
"access_token": self.access_token,
|
|
},
|
|
)
|
|
resp2.raise_for_status()
|
|
story_id = resp2.json().get("id")
|
|
logger.info(f"Published to Instagram story: {story_id}")
|
|
return story_id
|