【超簡単】PythonでThreads APIを使って遊んでみた

Posted at 2024-11-11

はじめに

この記事は、ThreadsのAPIをちょっと触ってみるかーっていう記事です。。。
過去にTwitter(現X)のAPIに関する記事を書いたのですが、仕様変更、料金プラン変更によって、個人利用が難しくなってしまった気がしています

そこで、Threads!!
Threadsは、個人的にはXと大して違わないサービスだし、APIが見たところ無料です！
熱い！激アツ！

ってことで、ざっくり使用方法紹介していきます

リンク集

Threads
Threads API ドキュメント
 Github

API利用までの流れ

基本的には、ドキュメントのGet Started を見ていけば問題なくAPIの取得までできました〜
Before You Start と事前準備がいくつか必要なのでやっていきましょう！

1. Meta app の作成

Meta Appを作成しましょう！
facebookのログインが求められるので、新規作成しましょう！
(こういうときは大体同じですが、新規アカウントを作成することをおすすめします)

ログインできたら、右上の「利用を開始する」
いい感じに設定していけばおっけです

アプリを作成します

今回は、ThreadsAPIを使いたいので、Threads APIにアクセスを選択します

アプリ名は適当で大丈夫です。特にThreadsから見られるわけではないので、自分がわかりやすい名前で大丈夫だと思います
今回は、「naoya-manager」にしました

これでMeta Appの作成が終了です！
完了するとダッシュボードに飛ばされるかと思います

2. Public Server

こちらは、画像や動画をThreadsで投稿したい場合に、サーバーを自分で用意する必要があるっぽいです
つまり、画像/動画を投稿するには、

ローカルの画像/動画をサーバー(自分で用意)にアップロード
URLを取得
Threadsに投稿

これらのステップが必要ということです
一旦はテキストだけの投稿だと必要ないので、あとで詳しく解説します

3. Authorization

こちらは、データのアクセスには認証が必要であるようです
アプリがデータにアクセスする前に、認証ウィンドウ（Authorization Window)を通じてこれらの権限をアプリに付与する必要があるそうです
必要な権限（Permissions）一覧

threads_basic
- すべてのThreadsエンドポイントの利用に必須。
threads_content_publish
- Threadsの投稿エンドポイントを使用する場合に必要。
threads_manage_replies
- リプライエンドポイントへのPOSTリクエストを行うために必要。
threads_read_replies
- リプライエンドポイントへのGETリクエストを行うために必要。
threads_manage_insights
- インサイトエンドポイントへのGETリクエストを行うために必要。

なかなか大変そうですねw
こちらも実際にコードを書いていくところでやっていきましょう！

4. Threads User Access Tokens

アクセストークンを発行する必要があるということですね
トークンはいくつかあるので注意してください

Threads User Access Tokens
- Threadsのユーザにアクセスできる、Threadsに投稿などができるトークン
Short-Lived Access Tokens
- 有効期限1時間
- ユーザーを認証し、Threads APIへのリクエストを認可するためのトークン
Long-Lived Access Tokens
- 有効期限60日間
- ユーザーに頻繁な再認証を求めずに、長期間にわたりAPIアクセスを維持するためのトークン

つまり、
Threads User Access Tokensは無期限で使える、自分のThreadsアカウントを操作するためのもの
Short/Long-Lived Access Tokensは、webアプリや、モバイルアプリから認証し、任意のユーザがその本人のThreadsアカウントを操作するためのもの
※実際試していないので、間違ってたらすみません

このパートで説明しているのは、Threads User Access Tokensです
まずは、先ほどのダッシュボードからユースケースに移動します

「Threads APIにアクセス」の右側の「カスタマイズ」をクリックします

ここで権限を付与できるようですね
一旦そのままthreads_basicのみで進めます

「設定」の下の方に「ユーザートークン生成ツール」があります
ここでトークンを発行できます
「Threadsテスターを追加または削除」からトークンを発行します

「メンバーを追加」からThreadsテスターを追加します
ここでは、Threadsのアカウント名を入力して、そのユーザー専用のトークンを発行します

なので、まずはThreadsにログインしましょう

Threadsのユーザ名をコピーして追加します

追加するとこのような「承認待ち」になります

これはThreadsから承認する必要があります

「設定」→「アカウント」→「ウェブサイトのアクセス許可」
同意したら、ダッシュボードに戻り、リロードとかすると承認ができているはずです

では最後にトークンを発行します
もう一度、「ユースケース」→「カスタマイズ」→「設定」に移動します

「アクセストークンを生成」をクリックするとトークンが発行できます

5. その他

事前準備には、以下の3つもありましたが、今回は使いませんので、軽く紹介して終わります🙇‍♀️

Short-Lived Access Tokens
Long-Lived Access Tokens
Authorization Window

次のような場面で使う用のトークンだと思います

Authorization Windowを表示
- WebアプリでAuthorization Windowをユーザーに表示し、Threadsへのアクセス許可をリクエスト
短期間有効アクセストークンの発行
- ユーザーがアクセスを許可すると、短期間有効（1時間）のアクセストークンを取得
長期間有効アクセストークンの発行
- 短期間有効アクセストークンを使用して、長期間有効（60日）のアクセストークンを取得
Threadsの操作
- 長期間有効アクセストークンを使用して、Threads APIを通じてThreadsを操作

なので、個人的に自分のアカウントを運用等するのみだといらないということですね
サービスとして、Threads運用ツールであったり、分析ツールを作るなら必要かもしれません
※実際試していないので、間違ってたらすみません

APIでできること

長かったー、、、
ここから実際にトークンを使ってThreadsを操作していきます！

できること

ドキュメントを読んだ感じ以下のことができるようです

投稿の作成と管理
- 新しい投稿
- 既存の投稿を管理
メディアの取得
- 投稿に関連する画像や動画などのメディアを取得
プロフィール情報の取得
- ユーザーのプロフィール情報を取得
返信の管理
- 投稿への返信を取得、作成、削除
インサイトの取得
- 投稿のパフォーマンスやユーザーのエンゲージメントに関する分析データを取得
Webhookの設定
- 特定のイベント（例：新しい返信やフォロワーの追加）に対する通知を受け取るためのWebhookを設定

具体的使用例

ここからはPythonでThreadsAPIを使って色々遊んでみようと思います

1. 新規投稿(テキストのみ)

投稿までの流れは以下の通りです

user idの取得
threadの作成
threadの公開

この3回、APIを叩く必要があります
(user idは1度取得すれば変わらない)

コードだけ知りたい方はこちら

import requests
import os
from dotenv import load_dotenv

load_dotenv()

ACCESS_TOKEN = os.getenv("ACCESS_TOKEN")
API_BASE_URL = "https://graph.threads.net/v1.0"


def get_user_id():
    url = f"{API_BASE_URL}/me"
    headers = {"Authorization": f"Bearer {ACCESS_TOKEN}"}
    response = requests.get(url, headers=headers)
    if response.ok:
        return response.json().get("id")
    return None


def create_text_thread(user_id, text):
    url = f"{API_BASE_URL}/{user_id}/threads"
    data = {"text": text, "media_type": "TEXT"}

    headers = {
        "Authorization": f"Bearer {ACCESS_TOKEN}",
        "Content-Type": "application/json",
    }
    response = requests.post(url, json=data, headers=headers)
    if response.ok:
        return response.json().get("id")
    return None


def publish_thread(user_id, creation_id):
    url = f"{API_BASE_URL}/{user_id}/threads_publish"
    data = {"creation_id": creation_id}
    headers = {
        "Authorization": f"Bearer {ACCESS_TOKEN}",
        "Content-Type": "application/json",
    }
    response = requests.post(url, json=data, headers=headers)
    if response.ok:
        return response.json()
    return None


def post_text(text="Hello World!"):
    # ユーザーIDの取得
    user_id = get_user_id()
    print("ユーザーID:", user_id)
    if not user_id:
        return

    # スレッドの作成
    creation_id = create_text_thread(user_id, text)
    print("スレッドの作成に成功しました:", creation_id)
    if not creation_id:
        return

    # スレッドの公開
    publish_response = publish_thread(user_id, creation_id)
    if publish_response:
        print("スレッドの公開に成功しました:", publish_response)


if __name__ == "__main__":
    post_text("Sample Post Text")

まずは、Threads User Access Tokensを.envに入れておきましょう

.env

ACCESS_TOKEN='**************************'

.envの内容を読み込みます

pip install python-dotenv

run.py

import os
from dotenv import load_dotenv

load_dotenv()

ACCESS_TOKEN = os.getenv("ACCESS_TOKEN")

ここからは、必要な関数を作っていきます

ユーザIDを取得する関数

import requests


def get_user_id():
    url = f"{API_BASE_URL}/me"
    headers = {"Authorization": f"Bearer {ACCESS_TOKEN}"}
    response = requests.get(url, headers=headers)
    if response.ok:
        return response.json().get("id")
    return None

スレッドを作成する関数
(これは、スレッドを作っているだけなので、下書きのような状態です)

def create_text_thread(user_id, text):
    url = f"{API_BASE_URL}/{user_id}/threads"
    data = {"text": text, "media_type": "TEXT"}

    headers = {
        "Authorization": f"Bearer {ACCESS_TOKEN}",
        "Content-Type": "application/json",
    }
    response = requests.post(url, json=data, headers=headers)
    if response.ok:
        return response.json().get("id")
    return None

スレッドを公開する関数

def publish_thread(user_id, creation_id):
    url = f"{API_BASE_URL}/{user_id}/threads_publish"
    data = {"creation_id": creation_id}
    headers = {
        "Authorization": f"Bearer {ACCESS_TOKEN}",
        "Content-Type": "application/json",
    }
    response = requests.post(url, json=data, headers=headers)
    if response.ok:
        return response.json()
    return None

最後にこれらの関数を使って一連の作業をするだけです

def post_text(text="Hello World!"):
    # ユーザーIDの取得
    user_id = get_user_id()
    print("ユーザーID:", user_id)
    if not user_id:
        return

    # スレッドの作成
    creation_id = create_text_thread(user_id, text)
    print("スレッドの作成に成功しました:", creation_id)
    if not creation_id:
        return

    # スレッドの公開
    publish_response = publish_thread(user_id, creation_id)
    if publish_response:
        print("スレッドの公開に成功しました:", publish_response)

if __name__ == "__main__":
    post_text("Sample Post Text")

これでpost_text関数を実行すればThredsにテキストが投稿されます！

2. 新規投稿(画像/動画付き)

画像はURLしか設定できません
つまり、ローカルの画像 or 動画を一度publicにアップロードする必要があります

2-1 すでにパブリックなURLがある場合

一旦URLが公開されている前提で投稿する方法を見てみましょう
今回は、いらすとやの画像をお借りします

https://blogger.googleusercontent.com/img/b/R29vZ2xl/AVvXsEiThh51O_5PBczGCVOAZqWk0NniNOu2Fxun8BlELAmHwR8Rltl1Gnqb_u0dkHvf34yGijTLvwnjWDAe6f-LtgOXAiX3sj__yCp5rsa2KTeaR0uaGye3zKUaTCUd8PiHDAObRfDSW8JT9qc/s800/hirameki_man.png

user idの取得、スレッドの公開に使った、
get_user_id関数とpublish_thread関数は先ほどと同じため割愛します

スレッドの作成

def create_text_img_thread(user_id, text, image_url):
    url = f"{API_BASE_URL}/{user_id}/threads"
    data = {
        "text": text,
        "image_url": image_url,
        "media_type": "IMAGE",
    }
    headers = {
        "Authorization": f"Bearer {ACCESS_TOKEN}",
        "Content-Type": "application/json",
    }
    response = requests.post(url, json=data, headers=headers)
    if response.ok:
        return response.json().get("id")
    return None

さっきとほとんど同じですね

def post_text_with_image(text="Hello World!", image_url=None):
    # ユーザーIDの取得
    user_id = get_user_id()
    print("ユーザーID:", user_id)
    if not user_id:
        return

    # スレッドの作成
    creation_id = create_text_img_thread(user_id, text, image_url)
    print("スレッドの作成に成功しました:", creation_id)
    if not creation_id:
        return

    # スレッドの公開
    publish_response = publish_thread(user_id, creation_id)
    if publish_response:
        print("スレッドの公開に成功しました:", publish_response)


if __name__ == "__main__":
    image_url = "https://blogger.googleusercontent.com/img/b/R29vZ2xl/AVvXsEiThh51O_5PBczGCVOAZqWk0NniNOu2Fxun8BlELAmHwR8Rltl1Gnqb_u0dkHvf34yGijTLvwnjWDAe6f-LtgOXAiX3sj__yCp5rsa2KTeaR0uaGye3zKUaTCUd8PiHDAObRfDSW8JT9qc/s800/hirameki_man.png"
    post_text_with_image("Sample Post with Image URL", image_url)

これで終わりです

2-2 ローカルの画像を投稿したい場合

ローカルにしかない画像/動画を投稿したい場合は、面倒ですが、一度パブリックにアップロードするほかありません
いくつか選択肢があるかなと思うので、自分の使い慣れているものや好きなもので問題ありません

Amazon S3
Imgur
Dropbox API
Google Cloud Storage
Firebase Storage
Microsoft Azure Blob Storage
ImageKit

こんないっぱいあるんですね
今回は個人的にCloudinaryが良さげだと思ったので、それでいきます

料金は基本的には無料でFreePlanを使えば良いかと思います
https://cloudinary.com/pricing

GET STARTEDからログインするとすんごく丁寧に使い方書いてくれているのでそのままその通りにやっていけばいけます

「③Upload, Optimize and Transform」の右にある「View API Keys」からAPI Key諸々をコピーしておきます

.env

CLOUDINARY_CLOUD_NAME='***************'
CLOUDINARY_API_KEY='*****************'
CLOUDINARY_API_SECRET='******************'

zsh

pip install cloudinary
pip install ulid-py

import cloudinary
import cloudinary.uploader
import ulid

CLOUDINARY_CLOUD_NAME = os.getenv("CLOUDINARY_CLOUD_NAME")
CLOUDINARY_API_KEY = os.getenv("CLOUDINARY_API_KEY")
CLOUDINARY_API_SECRET = os.getenv("CLOUDINARY_API_SECRET")


def upload_media(image_path):
    cloudinary.config(
        cloud_name=CLOUDINARY_CLOUD_NAME,
        api_key=CLOUDINARY_API_KEY,
        api_secret=CLOUDINARY_API_SECRET,
        secure=True,
    )

    upload_result = cloudinary.uploader.upload(
        image_path, public_id=f"threads/{ulid.new()}"
    )
    return upload_result["secure_url"]

url = upload_media('sample_img.jpg') # 自分のローカルの画像のpath
print(url)

実行すると

https://res.cloudinary.com/dfpowgarc/image/upload/v1731330734/threads/01JCDNMWPSXHNFCX3DRH4H9PSF.jpg

cloudinaryの方はMedia Explorerという箇所からアップロードした画像を確認できます

threadsというディレクトリにULIDを使用して保存しました

あとは、upload_media関数から受け取ったURLをThreadsへ投稿するだけなので、2-1とほぼ同じです

完成コード

import requests
import os
from dotenv import load_dotenv
import cloudinary
import cloudinary.uploader
import ulid

load_dotenv()

ACCESS_TOKEN = os.getenv("ACCESS_TOKEN")
API_BASE_URL = "https://graph.threads.net/v1.0"
CLOUDINARY_CLOUD_NAME = os.getenv("CLOUDINARY_CLOUD_NAME")
CLOUDINARY_API_KEY = os.getenv("CLOUDINARY_API_KEY")
CLOUDINARY_API_SECRET = os.getenv("CLOUDINARY_API_SECRET")


def get_user_id():
    url = f"{API_BASE_URL}/me"
    headers = {"Authorization": f"Bearer {ACCESS_TOKEN}"}
    response = requests.get(url, headers=headers)
    if response.ok:
        return response.json().get("id")
    return None


def upload_media(image_path):
    cloudinary.config(
        cloud_name=CLOUDINARY_CLOUD_NAME,
        api_key=CLOUDINARY_API_KEY,
        api_secret=CLOUDINARY_API_SECRET,
        secure=True,
    )

    upload_result = cloudinary.uploader.upload(
        image_path, public_id=f"threads/{ulid.new()}"
    )
    return upload_result["secure_url"]


def create_text_img_thread(user_id, text, image_url):
    url = f"{API_BASE_URL}/{user_id}/threads"
    data = {
        "text": text,
        "image_url": image_url,
        "media_type": "IMAGE",
    }
    headers = {
        "Authorization": f"Bearer {ACCESS_TOKEN}",
        "Content-Type": "application/json",
    }
    response = requests.post(url, json=data, headers=headers)
    if response.ok:
        return response.json().get("id")
    return None


def publish_thread(user_id, creation_id):
    url = f"{API_BASE_URL}/{user_id}/threads_publish"
    data = {"creation_id": creation_id}
    headers = {
        "Authorization": f"Bearer {ACCESS_TOKEN}",
        "Content-Type": "application/json",
    }
    response = requests.post(url, json=data, headers=headers)
    if response.ok:
        return response.json()
    return None


def post_text_with_image(text="Hello World!", image_path=None):
    # ユーザーIDの取得
    user_id = get_user_id()
    print("ユーザーID:", user_id)
    if not user_id:
        return

    # 画像のアップロード
    image_url = upload_media(image_path)

    # スレッドの作成
    creation_id = create_text_img_thread(user_id, text, image_url)
    print("スレッドの作成に成功しました:", creation_id)
    if not creation_id:
        return

    # スレッドの公開
    publish_response = publish_thread(user_id, creation_id)
    if publish_response:
        print("スレッドの公開に成功しました:", publish_response)


if __name__ == "__main__":
    image_path = "media/sample.PNG"
    post_text_with_image("Sample Post with Image", image_path)

動画投稿の場合は、スレッドの作成とスレッドの公開の間に30秒ほど待機を入れないとエラー吐きます

ちなみに、
cloudinaryに画像をアップロード
→ Threadsに画像を投稿
→ cloudinaryの画像を削除

のように画像を削除してもThreadsの投稿が消えることはありませんでした。。。なんでだろう
キャッシュされてるのが生きていただけなのか。。。？？
知っている方ご教授ください

終わり

疲れたのでこの記事はここまでにします
次の記事で続き書きます〜〜

今回のコード(改): https://github.com/naoya25/threads-bot

3. 投稿の取得(テキスト、画像、動画)

4. プロフィールの取得

5. インサイトの取得

You get articles that match your needs
You can efficiently read back useful information
You can use dark theme

What you can do with signing up