[Instagram Graph API]Instagramの投稿をハッシュタグベースで取得・保存する

Instagram Graph APIを使用して，特定のハッシュタグを使ったInstagramの投稿を検索・取得・保存する方法の備忘録です．
（記入時対象のInstagramのAPIのversionは19.0です）
次のようなことについて扱っています．

長期トークンの発行
ハッシュタグ文字列を用いたハッシュタグの取得
ハッシュタグIDを用いた投稿（メディア）の取得（ページネーション対応）
アプリレベルのレート制限・ハッシュタグのクエリ制限

前提：

Instagramビジネスアカウント（のような，ハッシュタグでの検索権限があるアカウント）取得，アプリの作成（開発モード - Mode: In developmentでよい）は済んでいるものとします．
コードについてはPython3.6以上．

サンプルとしてPythonのコードをおいておきます（あまりお行儀がよくないのですが）．なお，割と情報をprintするようにしています．

API確認と長期トークンの発行

はじめに，投稿を取得するための土台固めをします．
具体的には，そもそもAPIが使える状態かどうか（曖昧な言い方ですが）を確認し，長期トークンの発行・取得を行います．

ここではブラウザを使った方法を記載します．

APIが使える状態かどうかの確認

にアクセスして，Facebookにログインしてください．
右側の「アクセストークン」と書かれた場所の下に「Metaアプリ」とあるのを確認し，自分の作成したアプリ名が選択されている状態にしてください．
「ユーザーまたはページ」（User or Page）は「ユーザートークン」（User Token）でOKだと思われます．

その下側を見て，今後の操作で必要なアクセス許可（Permissions）があることを確認してください（instagram_basicとかbusiness_managementとか）．
足りなかったら増やしてください．

上部のURL欄では，初期値だと，「GET」が選択されており，
"https://graph.facebook.com/" + version + "me?fields=id,name"
といったURLが出ているのではないかと思われます．
この状態で「送信」（Submit）をクリックして，中央部に以下のようなものが出てきたら，OKです．

{
  "id": "（数字が複数）",
  "name": "（FaceBookでの名前）"
}

ではアクセストークンのコピーです．
右側のアクセストークン欄で，「Generate Access Token」の上側にある文字列をコピーしてください．（右のアイコンをクリックすればコピーできるはず）

長期アクセストークンの取得

上のアクセストークンは数時間で期限が切れてしまうので，60日程度OKらしい長期アクセストークンを取得します．

（ブラウザを使って取得する場合は）アクセストークンデバッガーのページ

にアクセスし，中央部で「アクセストークン」が選択された状態で，
その下の空欄に，先ほどコピーしたアクセストークンを貼り付けて「デバッグ」してください．
（※APIバージョンはアプリと合わせた方がいいかも）

その後，画面の下の方に出てくる「アクセストークンを延長」をクリックし，長期アクセストークンを取得してください．

なお，長期アクセストークンの詳細はこちらにあります．

ハッシュタグを使った投稿の収集

ハッシュタグを使った投稿の収集では，「ハッシュタグID」を経由する．

ことに注意．

「#〇〇」というハッシュタグを使った投稿を取得する場合には，
(1) 「〇〇」に割り当てられたハッシュタグIDを取得して，
(2) そのハッシュタグIDをもとに投稿を取得する，
という流れになります．

以下に記載するサンプルコードも，

ハッシュタグIDの取得
ハッシュタグIDを使った投稿の取得

という2つのパートを分けて記載しています．

さらに今回は，データのリクエストを行うためのURLを組み立てる処理を別の関数として切り分けています．

リクエストとデータ類の準備

import csv
import datetime
import json
import time

import requests

# Business AccountのIDの入力．
# 注意：（me?fields=id,name）で出てきたものではない．忘れた場合は，
# https://accountscenter.instagram.com/profiles/
# でビジネスアカウントを選択すればURL欄に出てくる．
IG_USER_ID = "YOUR ID（文字列）に置き換え" 
LONG_ACCESS_TOKEN = "YOUR LONG ACCESS TOKEN（文字列）に置き換え" 
# 一応短期トークンを使いたいなら↓のLONG_ACCESS_TOKENを書き換えて．

APP_INFO = {
    "INSTAGRAM_APP_NAME" : "（備忘録的にアプリ名を入れておく）",
    "API_VERSION": "v19.0"
    }


ACCESS_TOKEN_TEST = LONG_ACCESS_TOKEN

URL_GRAPH_API_ROOT = "https://graph.facebook.com/" + APP_INFO["API_VERSION"] + "/"
BASEURL_GET_HASHTAG_ID_BY_NAME = URL_GRAPH_API_ROOT + "ig_hashtag_search?"


# 投稿に関する欲しいfieldを入力しておく．
WANTED_FIELDS_LIST_BASE = ["id", "timestamp", "permalink", "media_product_type", "media_type", "comments_count", "caption"]

URLの組み立て

# URLの組み立て．

def make_url_get_hashtag_id_by_name(
        hashtag_name, user_id=IG_USER_ID, access_token=ACCESS_TOKEN_TEST):
    
    url = BASEURL_GET_HASHTAG_ID_BY_NAME + "user_id=" + user_id 
    url = url + "&access_token=" + access_token + "&q=" + hashtag_name
    return url
 

def make_url_get_posts_by_hashtag_id(
        hashtag_id, recent_or_top="recent", after=None, fields_list=WANTED_FIELDS_LIST_BASE, 
        user_id=IG_USER_ID, access_token=ACCESS_TOKEN_TEST):

    fields_str = str(",".join(fields_list))
    print(fields_str)

    request_url = URL_GRAPH_API_ROOT + hashtag_id
    if recent_or_top == "top":
        request_url = request_url + "/top_media?"
    elif recent_or_top == "recent":
        request_url = request_url + "/recent_media?"
    else:
        # どちらでもなかったらとりあえずrecentで取る．
        request_url = request_url + "/recent_media?"
    request_url = request_url + f"user_id={user_id}&access_token={access_token}&fields={fields_str}"
    
    if after is not None:
        request_url = request_url + f"&after={str(after)}"

    result = request_url
    
    return result

ハッシュタグIDの取得（「IGハッシュタグ検索」）

参考：

IGハッシュタグ検索 - https://developers.facebook.com/docs/instagram-api/reference/ig-hashtag-search?locale=ja_JP

# ハッシュタグIDの取得．
def get_hashtag_id_by_name(hashtag_name, user_id=IG_USER_ID, access_token=ACCESS_TOKEN_TEST):
    
    url = make_url_get_hashtag_id_by_name(hashtag_name=hashtag_name,
        user_id=user_id, access_token=access_token)
    print(url)
    response = requests.get(url)
    res_text = json.loads(response.text)
    print(res_text)
    
    if "error" in res_text.keys():
        print("response error")
        return None
    
    result = json.loads(response.text)["data"][0]["id"]
    result = str(result)
    print(result)

    return result

投稿の取得

参考：

IGハッシュタグ - https://developers.facebook.com/docs/instagram-api/reference/ig-hashtag?locale=ja_JP
- IGハッシュタグ付きの最近のメディア - https://developers.facebook.com/docs/instagram-api/reference/ig-hashtag/recent-media?locale=ja_JP
- IGハッシュタグ付きのトップメディア - https://developers.facebook.com/docs/instagram-api/reference/ig-hashtag/top-media?locale=ja_JP」

ページネーションについて：
「応答は、ページあたりの結果が最大50件のlimitでページネーションされます。」（https://developers.facebook.com/docs/instagram-api/reference/ig-hashtag/recent-media?locale=ja_JP）
とあります．
1回のクエリでは結果は最大50件までしか取得できないようなので，その「ページネーション」に対応します．
具体的には，afterカーソルの値があれば取得し，それを用いてリクエスト対象のURLを構築し，また問い合わせします．
（※このafterを使ったURLの仕様は，ハッシュタグIDに対する投稿取得の場合と，特定ユーザの投稿取得の場合でちょっと違う）

# 投稿の取得（ページネーション対応）．
def get_posts_by_hashtag_id_with_paging(
        hashtag_id, max_paging=7, recent_or_top="recent", 
        fields_list=WANTED_FIELDS_LIST_BASE, user_id=IG_USER_ID, 
        access_token=ACCESS_TOKEN_TEST):

    notice = "ページネーション対応でpost取得。"
    print(notice)

    now_paging = 0
    cursors_after = None
    will_continue = True
    
    posts_list = []

    while will_continue:
    
        url = make_url_get_posts_by_hashtag_id(
            hashtag_id=hashtag_id, recent_or_top=recent_or_top, 
            after=cursors_after, fields_list=fields_list, 
            user_id=user_id, access_token=access_token
            )
        print(url)
        response = requests.get(url)
        print(response)
        res_text = json.loads(response.text)

        if "error" in res_text.keys():
            print("Error.")
            print(response.headers)
            print(res_text)
            if cursors_after is not None:
                print("cursors_after: " + str(cursors_after))
            print(f"{hashtag_id} : Please check the error message. " + 
                  "If (#4) ('Application request limit reached'), please try again after a while.")
            break

        cursors_after = None

        if "paging" in res_text.keys():
            if "cursors" in res_text["paging"].keys():
                cursors = res_text["paging"]["cursors"]
                if "after" in cursors.keys():
                    cursors_after = cursors["after"]
        if cursors_after is None:
            will_continue = False
        if now_paging >= max_paging:
            will_continue = False
        
        try:
            if "data" in res_text.keys():
                posts_list.extend(res_text["data"])
        except Exception as e:
            print(e)
            print(f"{hashtag_id} : The data is not good.")
            break
        
        now_paging = now_paging + 1

        time.sleep(2)
    
    return posts_list

リクエストの余裕有無の確認用

（レート制限については後述）

def get_api_response_header(access_token):

    url = URL_GRAPH_API_ROOT +  "me?fields=id,name&access_token=" + access_token
    response = requests.get(url)
    print(response.headers)

    return str(response.headers)


def is_limit_reached(access_token):

    result = get_api_response_header(access_token=access_token)
    if "(#4) Application request limit reached" in result:
        return True
    else:
        return False

JSONファイル保存

今回はJSONファイルを保存する方法を載せておきます．
目視確認したい場合向けにjson.dump()でのensure_asciiがFalseでの保存をデフォルトにします．必要ならTrueのファイルも保存．

def make_json_from_posts_list_ymd(label: str, posts_list: list, will_save_ascii: bool = False):

    today = datetime.date.today()
    filename = label + '_' + today.strftime('%Y%m%d_%H') + '.json'
    make_json_from_list(filename=filename, ls=posts_list, will_save_ascii=will_save_ascii)

    return True


def make_json_from_list(filename: str, ls: list, data_dir: str = "./", will_save_ascii: bool = False):

    notice = "指定filenameで、指定listをjsonとして保存。"
    print(notice)

    if not filename.endswith('.json'):
        filename = filename + ".json"
    
    file_path = os.path.join(data_dir, filename)

    with open(file_path, 'w') as f:
        json.dump(ls, f, indent=2, ensure_ascii=False)
    
    if will_save_ascii:
        print("ensure_asciiも保存")
        filename_ascii = "ascii_" + filename
        file_path_ascii = os.path.join(data_dir, filename_ascii)
        with open(file_path_ascii, 'w') as f:
            json.dump(ls, f, indent=2, ensure_ascii=True)

    return ls

色々呼び出す

def get_posts_by_hashtag_name_with_paging(
        hashtag_name, max_paging=7, recent_or_top="recent", 
        fields_list=WANTED_FIELDS_LIST_BASE, user_id=IG_USER_ID, access_token=ACCESS_TOKEN_TEST):
    
    hashtag_id = get_hashtag_id_by_name(
        hashtag_name=hashtag_name, user_id=user_id, access_token=access_token)
    
    posts_list = get_posts_by_hashtag_id_with_paging(
        hashtag_id=hashtag_id, max_paging=max_paging, 
        recent_or_top=recent_or_top, fields_list=fields_list, 
        user_id=user_id, access_token=access_token)

    return posts_list


def make_json_by_hashtag_name_with_paging(
        hashtag_name, max_paging=7, recent_or_top="recent", 
        fields_list=WANTED_FIELDS_LIST_BASE, will_save_ascii=True, 
        user_id=IG_USER_ID, access_token=ACCESS_TOKEN_TEST):

    posts_list = get_posts_by_hashtag_name_with_paging(
        hashtag_name=hashtag_name, max_paging=max_paging, 
        recent_or_top=recent_or_top, fields_list=fields_list, 
        user_id=user_id, access_token=access_token)
    
    label = "IG_tag_" + hashtag_name + "_pmax_" + str(max_paging) + "_" + recent_or_top

    make_json_from_posts_list_ymd(label=label, posts_list=posts_list, will_save_ascii=will_save_ascii)

    return posts_list

実行用

使い方等

hashtag_names_for_recent と hashtag_names_for_top に，検索したいハッシュタグの#以降をリスト形式で入れる．
- [情報]絵文字ハッシュタグには未対応らしい（2024/03現在） https://developers.facebook.com/docs/instagram-api/guides/hashtag-search?locale=ja_JP 「ハッシュタグ内の絵文字のクエリはサポートされていません」
関数 make_json_by_hashtag_name_with_paging を呼び出すところの max_paging の指定を変えると，最大で何ページ分読み込むかを調整できる．

def main():

    access_token = LONG_ACCESS_TOKEN

    hashtag_names_for_recent = ['これ', 'それ']
    hashtag_names_for_top = ['あれ', 'どれ']

    for hashtag_name in hashtag_names_for_recent:
        now_info = make_json_by_hashtag_name_with_paging(hashtag_name=hashtag_name, 
                                        max_paging=8, recent_or_top="recent", 
                                        will_save_ascii=False,
                                        user_id=IG_USER_ID, access_token=access_token)
        if now_info is None:
            with open("error.csv", mode="a", newline="") as error_file:
                error_writer = csv.writer(error_file)
                error_writer.writerow([datetime.datetime.now(), hashtag_name, "繰り返しでエラー(recent)"])
            if is_limit_reached(access_token=access_token):
                break
            else:
                pass
        time.sleep(5)

    for hashtag_name in hashtag_names_for_top:
        now_info = make_json_by_hashtag_name_with_paging(hashtag_name=hashtag_name, 
                                        max_paging=8, recent_or_top="top", 
                                        will_save_ascii=False,
                                        user_id=IG_USER_ID, access_token=access_token)
        if now_info is None:
            with open("error.csv", mode="a", newline="") as error_file:
                error_writer = csv.writer(error_file)
                error_writer.writerow([datetime.datetime.now(), hashtag_name, "繰り返しでエラー(top)"])
            if is_limit_reached(access_token=access_token):
                break
            else:
                pass
        time.sleep(5)

    return True

実行：

Google Colabなら

main()

でOKかと．ファイル実行なら

if __name__ == "__main__":

    main()

を加えておくなど．

注意点：アクセスの上限

アプリレベルのレート制限

データ収集にはまず短期間でのlimitがあります．

get_api_response_header という関数で書きましたが，適当にアクセスしてheaderを見ると，余裕がどれぐらいあるか確認できます．

アプリのダッシュボードからいける「アプリレベルのレート制限」
https://developers.facebook.com/apps/〇〇/rate-limit-details/app/
よりもはっきりとわかるかと．

結果例（一部抜粋）： '{"call_count":9,"total_cputime":0,"total_time":28}'

データを集めていて特にあがりがちだったのが，total_timeです．100を超えたら下がるまでしばらく待たないと弾かれる仕様だったかと．
（体感，アプリのダッシュボードの「レート制限」で「残り」が5割を切ったら気を付ける感じ）

（自信なし・経験）場合によっては，リクエストするfieldsを減らすとtotal_timeの上昇がやや和らげられることもあるかもしれません．

ハッシュタグのクエリの制限

上とは別に，ハッシュタグ検索にも上限があります．

参照： https://developers.facebook.com/docs/instagram-api/guides/hashtag-search?locale=ja_JP

InstagramビジネスアカウントまたはInstagramクリエイターアカウントに代わってクエリできる一意のハッシュタグは、7日のローリング期間で最大30件です。一度ハッシュタグをクエリすると、7日間にわたってこの制限がカウントされます。この時間枠内で同じハッシュタグをクエリした場合、制限に対するカウントは行われず、最初のクエリで起動した7日間のタイマーはリセットされません。

1日の中で時間を空けても，たとえば2,3日以内にたくさん検索すると次のようなエラーが返ってくることがあります．

{'error': {'message': 'This API call could not be completed due to resource limits', 'type': 'OAuthException', 'code': 18, 'error_subcode': 2207034, 'is_transient': False, 'error_user_title': 'ハッシュタグ検索の上限を超えました。1週間後に新しいハッシュタグをもう一度お試しください。', 'error_user_msg': 'Instagramビジネスは、Graph APIを使用して、週ごとに限られた数のユニークなハッシュタグを検索することができます。 開発者向けドキュメントを参照して、アクセス可能なハッシュタグのリストをチェックしてください。https://developers.facebook.com/docs/instagram-api/hashtag-search', 'fbtrace_id': 'ほげほげ'}}