python-facebook-api

To receive real-time updates for live video comments and reactions, you must use the ServerSentEventAPI class. This implementation follows the SSE web standard to establish a continuous data stream from Facebook.

To use this feature, you must:

1. Subclass ServerSentEventAPI.
2. Override the on_data method to define how incoming data is processed.
3. Instantiate your class with a valid access_token.
4. Call live_comments() with the target live_video_id and the desired fields.

import json
from pyfacebook import ServerSentEventAPI

class MyEvent(ServerSentEventAPI):
    def on_data(self, data):
        # data is received as bytes
        raw_data: str = data.decode()

        # SSE data typically starts with 'data: ', so we slice from index 5
        data_json = json.loads(raw_data[5:])
        print(f"Comment Data: {data_json}")

# Initialize and connect
event_api = MyEvent(access_token="Your access token")
event_api.live_comments(
    live_video_id="ID for the live video",
    fields="from{id,name},message"
)

To publish content, you must first create a container and then publish it once the container status is FINISHED.

1. Create Container: Use create_photo(...) or create_video(...). This returns an IgProContainer object.
2. Check Status: Use get_container_info(container_id=...) to ensure the status is FINISHED.
3. Publish: Use publish_container(creation_id=...). On success, it returns the new media ID.

Note: Instagram accounts are limited to 25 API-published posts within a 24-hour period. Check your remaining quota with get_publish_limit().

# 1. Create a photo container
container = api.create_photo(
    image_url="https://www.example.com/images/gugges.jpg",
    caption="publish test",
    location_id="7640348500",
    user_tags='[{"username": "somebody", "x": 0.5, "y": 0.8}]'
)

# 2. Check if container is ready
status = api.get_container_info(container_id=container.id)
if status.status_code == 'FINISHED':
    # 3. Publish
    result = api.publish_container(creation_id=container.id)
    print(f"Published! Media ID: {result['id']}")

# Check daily limit
print(api.get_publish_limit())

To enable users to create and publish content on Threads via your app, you must implement an OAuth flow using ThreadsGraphAPI.

1. Initialize ThreadsGraphAPI: Provide your app_id, app_secret, set oauth_flow=True, define your redirect_uri, and specify the required scope (e.g., threads_basic, threads_content_publish).
2. Get Authorization URL: Call api.get_authorization_url() to generate the URL where the user must authorize your app.
3. Exchange Code for Token: After the user authorizes, they will be redirected to your redirect_uri with a code parameter in the URL. Pass this full redirected URL to api.exchange_user_access_token(response="...") to retrieve the access_token and user_id.

from pyfacebook import ThreadsGraphAPI

api = ThreadsGraphAPI(
    app_id="Your app id",
    app_secret="Your app secret",
    oauth_flow=True,
    redirect_uri="Your callback domain",
    scope=[
        "threads_basic", 
        "threads_content_publish", 
        "threads_read_replies", 
        "threads_manage_replies",
        "threads_manage_insights"
    ]
)

# 1. Get the URL for the user to visit
auth_url = api.get_authorization_url()
print(f"Authorize here: {auth_url}")

# 2. After redirection, exchange the response URL for a token
# Example response URL: https://example.com/callback?code=AQBZzYhLZB&state=PyFacebook#_
token_data = api.exchange_user_access_token(response="Your response url")
print(token_data)
# Output: {'access_token': 'access_token', 'user_id': 12342412}