Twitter Streaming APIで省略されていないツイートテキストを取得する

Twitter Streaming APIでツイートテキストを取得するときに、省略されていないテキスト全文を取得する方法をまとめておく。

環境

WSL2(Ubuntu20.04)。

$ lsb_release -dr
Description:    Ubuntu 20.04.5 LTS
Release:        20.04
$ python3 -V
Python 3.8.10

省略されていないテキストを取得する方法

Twitter Streaming APIでツイートテキストを収集する場合、レスポンスデータのtextフィールドに、最大140文字までのテキストが格納されている。

ただし、2017年に半角文字の言語では280文字までに文字数制限が拡張されたために、ツイートテキストが140文字を超える場合がある。ツイートテキストが140文字を超える場合、textフィールドの最後が三点リーダー（…）となり、テキストの一部が省略されてしまう。

拡張された文字数のツイートテキストを取得するには、extended_tweetフィールドを使う。このextended_tweetフィールドはツイートテキストの文字数が140文字を超えた場合に存在し、このフィールドのfull_textフィールドにツイートテキストが格納されている。

ただこれでもまだ省略されてしまう場合がある。それはリツイートの場合で、ツイートテキストにRTとメンションが付与されるので、元のツイートが文字数制限いっぱいだと、ツイートテキストは一部省略されてしまう。

リツイートの場合、レスポンスデータのextended_tweetフィールドに元ツイートの情報があるので、そのツイートテキストを取得することで、省略されていないテキストを取得することができる。

省略されていないテキストを取得するコード

上記をふまえて、PythonでTwitter Streaming APIのレスポンスから省略されていないツイートテキスト全文を取得するコードを作成する。Twitter Streaming APIのデータ取得には、PythonライブラリのTweepyを使う。TweepyでStreaming APIを使う方法はTweepy4でリアルタイムツイートを取得するを参照。以下はStreaming APIでランダムサンプリングを取得して、スクリーンネームとツイートテキストを表示するコード。

import tweepy

# Twitter APIの認証情報
# Twitterの開発者向けのページで取得したキーとトークンを使う
CONSUMER_KEY = 'Consumer Key'
CONSUMER_SECRET = 'Consumer Secret'
ACCESS_TOKEN = 'Access Token'
ACCESS_TOKEN_SECRET = 'Access Token Secret'

class TwStream(tweepy.Stream):
    def _get_text(self, status):
        # 拡張ツイートの場合
        if hasattr(status, 'extended_tweet'):
            return status.extended_tweet['full_text']
        else:
            return status.text

    def on_status(self, status):
        # status(APIで取得できるデータ)を受け取ったときにon_statusが実行される
        print('-------------------------')
        # リツイートの場合は「RT」+ メンション分だけ文字数オーバーするので省略されることがある
        if hasattr(status, 'retweeted_status'):
            text = self._get_text(status.retweeted_status)
        else:
            text = self._get_text(status)
        # スクリーンネームとツイートを表示
        print(f'@{status.author.screen_name} {text}')

def main():
    twstream = TwStream(CONSUMER_KEY, CONSUMER_SECRET, ACCESS_TOKEN, ACCESS_TOKEN_SECRET)

    # 日本語ツイートに限定してランダムサンプリングを取得
    twstream.sample(languages=['ja'])

if __name__ == '__main__':
    main()