1
0

Delete article

Deleted articles cannot be recovered.

Draft of this article would be also deleted.

Are you sure you want to delete this article?

[備忘録]Xで指定したテキストを含むハッシュタグ付き等の投稿を検索・取得する

Last updated at Posted at 2024-09-14

細々した話

Basicプランの課金タイミングについての注意

1ヶ月100ドルと書いてあったが、登録したらそこで「1ヶ月分」課金されるわけではなかった(2024年9月時点での報告)。

  • 最初に1週間分課金された(日割り)。
  • 1週間後に1ヶ月分課金された。

クレジットカードの締日と相談して経理系の手続きを行う場合は注意。

事前設定

APPを作ること。

必要な認証系の情報は、 「Apps」の個別アプリの「Keys and tokens」リンク(鍵マーク)で行けるページから作成・取得できる。

  • API Key and Secret (Consumer Keysの下)
  • Bearer Token(Authentication Tokensの下)
  • Access Token and Secret (Authentication Tokensの下)

※ User authentication settingsはデータ取得のみなら設定不要かと思われる。

Searchの注意点(Basicプラン)

https://developer.x.com/en/portal/products/basic
より

GET /2/tweets/search/recent
についての制限(Rate Limit)

  • 60 requests / 15 mins PER USER
  • 60 requests / 15 mins PER APP

※これがあるので、一気に様々なクエリを投げたい場合は注意。

他の情報

  • 10 default results per response
  • 100 results per response
    • 512 query length
    • core operators

mex_resultsは10以上100以下。

関数

tweepyを使うのであらかじめインストールする。
使うライブラリを適当にimportしておく。

from time import sleep

import tweepy

今回は認証にいる情報や使いたいフィールドを別のファイルに切り分けておく。

import a_api_constant

API_Key = a_api_constant.API_KEY
API_Sec = a_api_constant.API_SEC
Token = a_api_constant.TOKEN
Token_Sec = a_api_constant.TOKEN_SEC
BEARER = a_api_constant.BEARER_TOKEN

TWEET_FIELDS_DEFAULT_SEARCHHT = a_api_constant.TWEET_FIELDS_DEFAULT
EXPANSIONS_DEFAULT_SEARCHHT = a_api_constant.EXPANSIONS_DEFAULT

デフォルトのフィールドの例。


TWEET_FIELDS_DEFAULT=[
    "created_at",
    "lang",
    "entities",
    "attachments",
    "conversation_id",
    "geo",
    "edit_history_tweet_ids",
    "public_metrics",
]
EXPANSIONS_DEFAULT = ["attachments.media_keys", "author_id", "in_reply_to_user_id"]

本体。

指定したテキストでハッシュタグ付き・RP(RT)以外の投稿を取得(言語指定可能)

# max_results: between 10 and 100. 
# filter_lang: 'ja', 'en'など。
def api_search_wtHTwoRT(search_text='(#abc)', filter_lang=None, max_results=50):
    
    client = tweepy.Client(
        bearer_token=BEARER,
        consumer_key=API_Key,
        consumer_secret=API_Sec,
        access_token=Token,
        access_token_secret=Token_Sec,
    )

    results = []
    search_query = search_text + ' -is:retweet has:hashtags'
    print(search_query)
    if filter_lang is not None:
        search_query = search_query + " lang:" + filter_lang

    response = client.search_recent_tweets(
        query=search_query,
        tweet_fields=TWEET_FIELDS_DEFAULT_SEARCHHT,
        expansions=EXPANSIONS_DEFAULT_SEARCHHT,
        max_results=max_results,
    )

    tweets = response.data
    
    sleep(1)

    if tweets != None:

        for tweet in tweets:
            obj = {}
            obj["tweet_id"] = tweet.id
            created_at = tweet.created_at
            created_at_str = created_at.isoformat()
            obj["created_at"] = created_at_str
            obj["text"] = tweet.text
            obj["author_id"] = tweet.author_id
            obj["conversation_id"] = tweet.conversation_id
            if tweet.in_reply_to_user_id is not None:
                obj["in_reply_to_user_id"] = tweet.in_reply_to_user_id
            if tweet.attachments is not None:
                obj["attachments"] = tweet.attachments
            if tweet.entities is not None:
                obj["entities"] = tweet.entities
            if tweet.geo is not None:
                obj["geo"] = tweet.geo

            results.append(obj)
    else:
        print('NOT FOUND')

    return results

ハッシュタグ付きの縛りがないバージョン。

def api_search_woRT(search_text='"abc"', filter_lang=None, max_results=50):
    
    client = tweepy.Client(
        bearer_token=BEARER,
        consumer_key=API_Key,
        consumer_secret=API_Sec,
        access_token=Token,
        access_token_secret=Token_Sec,
    )

    search_query = search_text + ' -is:retweet'
    print(search_query)
    if filter_lang is not None:
        search_query = search_query + " lang:" + filter_lang

    response = client.search_recent_tweets(
        query=search_query,
        tweet_fields=TWEET_FIELDS_DEFAULT_SEARCHHT,
        expansions=EXPANSIONS_DEFAULT_SEARCHHT,
        max_results=max_results,
    )

    tweets = response.data

    sleep(1)

    if tweets != None:

        for tweet in tweets:
            obj = {}
            obj["tweet_id"] = tweet.id
            created_at = tweet.created_at
            created_at_str = created_at.isoformat()
            obj["created_at"] = created_at_str
            obj["text"] = tweet.text
            obj["author_id"] = tweet.author_id
            obj["conversation_id"] = tweet.conversation_id
            if tweet.in_reply_to_user_id is not None:
                obj["in_reply_to_user_id"] = tweet.in_reply_to_user_id
            if tweet.attachments is not None:
                obj["attachments"] = tweet.attachments
            if tweet.entities is not None:
                obj["entities"] = tweet.entities
            if tweet.geo is not None:
                obj["geo"] = tweet.geo

            results.append(obj)
    else:
        print('NOT FOUND')

    return results

1
0
0

Register as a new user and use Qiita more conveniently

  1. You get articles that match your needs
  2. You can efficiently read back useful information
  3. You can use dark theme
What you can do with signing up
1
0

Delete article

Deleted articles cannot be recovered.

Draft of this article would be also deleted.

Are you sure you want to delete this article?