Twitterの投稿を取得したいと考える人は多いでしょう。PythonとTwitter APIを組み合わせることで、ツイートを手軽に取得することが可能です。

この記事では、取得したいツイートの数に応じて2パターンのコードを記載しています。

Google Colaboratoryを利用すると、環境構築の手間なくコードを実行できるので、ぜひ試してみてください。

なお、Twitter APIの取得方法については言及していませんのでご了承ください。

Twitter APIを取得する方法

この記事が参考になりましたので、この手順に従って申請を出せばOKです。

PythonでTwitter APIを利用する方法を現役エンジニアが解説【初心者向け】

少量のツイートを取得したい場合

取得したいツイートの数が少ない場合はシンプルなコードの記述で済みます。
以下参照してください。

import tweepy
import csv

consumer_key = '**********'	# Consumer Key を入力する
consumer_secret = '**********'	# Consumer Secret を入力する
access_token = '**********'		# Access Token を入力する
accesstokensecret = '**********'	# Accesss Token Secert を入力する
auth = tweepy.OAuthHandler(consumerkey, consumersecret)
auth.setaccesstoken(accesstoken, accesstoken_secret)

api = tweepy.API(auth)

#検索キーワードをqに、取得ツイート数をcountに入力する

q = "任意のキーワードを入力する"

count = 10

#ここからツイート取得

tweet_list=[]

tweets = api.search(q=q, locale="ja", count=count, tweet_mode='extended')

for tweet in tweets:

    tweetlist.append([tweet.user.id, tweet.user.screenname, tweet.user.followerscount, tweet.user.friendscount, tweet.user.description, tweet.id, tweet.user.createdat, tweet.createdat, tweet.full_text])

#csv出力

with open('tweetssearchlist.csv', 'w',newline='',encoding='utf-8') as f:

    writer = csv.writer(f, lineterminator='\n')

    writer.writerow(["userid", "screenname", "follower","follow","description","tw_id", "when created", "when posted", "fulltext"])

    writer.writerows(tweet_list)

    print(tweet_list)

pass

大量のツイートを取得したい場合

大量のツイートを取得したい場合は工夫が必要です。

なぜなら、Twitter APIには取得上限があるためです。

以下のように、待機時間を考慮したコードの記述が求められます。

#Tweepyによるツイート取得方法を解説する

import tweepy
import pandas as pd
from requests_oauthlib import OAuth1Session
import json
import datetime, time, sys
from abc import ABCMeta, abstractmethod

consumer_key = '**********'	# Consumer Key を入力する

consumer_secret = '**********'	# Consumer Secret を入力する

access_token = '**********'		# Access Token を入力する

accesstokensecret = '**********'	# Accesss Token Secert を入力する

auth = tweepy.OAuthHandler(consumerkey, consumersecret)

auth.setaccesstoken(accesstoken, accesstoken_secret)

api = tweepy.API(auth)

class TweetsGetter(object):

    metaclass = ABCMeta

    def init(self):

        self.session = OAuth1Session(consumerkey, consumersecret, accesstoken, accesstoken_secret)

    @abstractmethod

    def specifyUrlAndParams(self, keyword):

        ###

        呼出し先 URL、パラメータを返す

        ###

    @abstractmethod

    def pickupTweet(self, res_text, includeRetweet):

        ###

        res_text からツイートを取り出し、配列にセットして返す

        ###

    @abstractmethod

    def getLimitContext(self, res_text):
    def collect(self, total = -1, onlyText = False, includeRetweet = False):

        ###

        ツイート取得を開始する。

        API接続によるツイート取得回数制限を確認する。

        ###

        self.checkLimit()

        ###

        次に、URLおよびパラメータを確認する。

        ###

        url, params = self.specifyUrlAndParams()

        params['include_rts'] = str(includeRetweet).lower()

        
        ###
        
        実際にツイートを取得する。

        ###

        cnt = 0

        unavailableCnt = 0

        while True:

            res = self.session.get(url, params = params)

            if res.status_code == 503:

                if unavailableCnt > 10:

                    raise Exception('Twitter API error %d' % res.status_code)

                unavailableCnt += 1

                print ('Service Unavailable 503')

                self.waitUntilReset(time.mktime(datetime.datetime.now().timetuple()) + 30)

                continue

            unavailableCnt = 0

            if res.status_code != 200:

                raise Exception('Twitter API error %d' % res.status_code)

            tweets = self.pickupTweet(json.loads(res.text))

            if len(tweets) == 0:

                ###
                
                len(tweets) != params['count'] という表現は使えない。
                なぜなら、 count は最大値を取得するため。
                したがって、"== 0" に設定する。

                ###

                break

            for tweet in tweets:

                if (('retweeted_status' in tweet) and (includeRetweet is False)):

                    pass

                else:

                    if onlyText is True:

                        yield tweet['text']

                    else:

                        yield tweet

                    cnt += 1

                    if cnt % 100 == 0:

                        print ('%d件 ' % cnt)

                    if total > 0 and cnt >= total:

                        return

            params['max_id'] = tweet['id'] - 1

            ###

            ヘッダ確認を確認し、回数制限を回避する。
            
            ここでX-Rate-Limit-Remaining の有無をチェックし、無い場合はアラートを表示する
            
            ###

            if ('X-Rate-Limit-Remaining' in res.headers and 'X-Rate-Limit-Reset' in res.headers):

                if (int(res.headers['X-Rate-Limit-Remaining']) == 0):

                    self.waitUntilReset(int(res.headers['X-Rate-Limit-Reset']))

                    self.checkLimit()

            else:

                print ('not found  -  X-Rate-Limit-Remaining or X-Rate-Limit-Reset')

                self.checkLimit()

    def checkLimit(self):

        ###

        回数制限を超過しないようにするため、回数を問い合わせる
        → アクセス可能になるまで wait する

        ###

        unavailableCnt = 0

        while True:

            url = "https://api.twitter.com/1.1/application/ratelimitstatus.json"

            res = self.session.get(url)

            if res.status_code == 503:

                if unavailableCnt > 10:

                    raise Exception('Twitter API error %d' % res.status_code)

                unavailableCnt += 1

                print ('Service Unavailable 503')

                self.waitUntilReset(time.mktime(datetime.datetime.now().timetuple()) + 30)

                continue

            unavailableCnt = 0

            if res.status_code != 200:

                raise Exception('Twitter API error %d' % res.status_code)

            remaining, reset = self.getLimitContext(json.loads(res.text))

            if (remaining == 0):

                self.waitUntilReset(reset)

            else:

                break

    def waitUntilReset(self, reset):

        ###

        reset 時刻まで sleep する

        ###

        seconds = reset - time.mktime(datetime.datetime.now().timetuple())

        seconds = max(seconds, 0)

        print ('\n     =====================')

        print ('     == waiting %d sec ==' % seconds)

        print ('     =====================')

        sys.stdout.flush()

        time.sleep(seconds + 15)  # + 15 秒のマージンを取り、エラーを回避する

    @staticmethod

    def bySearch(keyword):

        return TweetsGetterBySearch(keyword)

    @staticmethod

    def byUser(screen_name):

        return TweetsGetterByUser(screen_name)

class TweetsGetterBySearch(TweetsGetter):

    ###

    キーワードでツイートを検索

    ###

    def init(self, keyword):

        super(TweetsGetterBySearch, self).init()

        self.keyword = keyword

    def specifyUrlAndParams(self):

        ###

        呼出し先 URL、パラメータを返す

        ###

        url = 'https://api.twitter.com/1.1/search/tweets.json'

        params = {'q':self.keyword, 'count':100}

        return url, params

    def pickupTweet(self, res_text):

        ###

        res_text からツイートを取り出し、配列にセットして返却

        ###

        results = []

        for tweet in res_text['statuses']:

            results.append(tweet)

        return results

    def getLimitContext(self, res_text):

        ###

        回数制限の情報を取得する

        ###

        remaining = res_text['resources']['search']['/search/tweets']['remaining']

        reset     = res_text['resources']['search']['/search/tweets']['reset']

        return int(remaining), int(reset)

class TweetsGetterByUser(TweetsGetter):

    ###

    ユーザーを指定してツイートを取得する

    ###

    def init(self, screen_name):

        super(TweetsGetterByUser, self).init()

        self.screenname = screenname

    def specifyUrlAndParams(self):

        ###

        呼出し先 URL、パラメータを返す

        ###

        url = 'https://api.twitter.com/1.1/statuses/user_timeline.json'

        params = {'screenname':self.screenname, 'count':200}

        return url, params

    def pickupTweet(self, res_text):

        ###

        res_text からツイートを取り出し、配列にセットして返却

        ###

        results = []

        for tweet in res_text:

            results.append(tweet)

        return results

    def getLimitContext(self, res_text):

        ###

        回数制限の情報を取得 （起動時）

        ###

        remaining = restext['resources']['statuses']['/statuses/usertimeline']['remaining']

        reset     = restext['resources']['statuses']['/statuses/usertimeline']['reset']

        return int(remaining), int(reset)

if name == 'main':

###

★★★ここに必要事項を入力する★★★

    # キーワードで取得

    getter = TweetsGetter.bySearch(u'Python')

    # ユーザーを指定して取得 （screen_name）

    #getter = TweetsGetter.byUser('@realDonaldTrump')

※ユーザーを指定して取得したい場合、上記'#'を削除し、キーワード取得側に'#'を付記する

###

    list_text = []

    list_id = []

    listuserscreenname = []

    listcreatedat = []

    for tweet in getter.collect(total = 3000):

        list_text.append(tweet['text'])

        list_id.append(tweet['id'])

        listuserscreenname.append(tweet['user']['screen_name'])

        listcreatedat.append(tweet['created_at'])

df = pd.DataFrame(columns=['text', 'id', 'user', 'created_at'])

dfnew = df.assign(text=listtext, id=listid, user=listuserscreenname, createdat=listcreatedat)

print(df_new)

取得したツイートは感情分析や共起ネットワークの描画に使える

ツイートは取得しただけでは宝の持ち腐れです。

きちんと分析に使ってこそ、力を発揮します。

たとえば感情分析が良い例です。以下の記事が参考になりましたので、実際にコードを書いてみると良いでしょう。

PythonとTwitter API v2で感情分析（センチメント分析）する手順をわかりやすく解説

「暇」ツイートのテキストマイニングによる潜在的ニーズの発見法に関する一考察

また、共起ネットワークの描画については、以下の記事が参考になりました。

Twitter データを利用した「新型コロナウィルス」関連語句の分析

共起ネットワークの作り方と考察方法｜KH CoderとPythonでの手順を丁寧に解説します

まとめ

Twitter APIの取得が少し面倒くさいのですが、一度発行してしまえばあとは簡単にデータ取得ができるので便利です！

Twitter分析にPythonを活用してみてはいかがでしょうか。ご参考になれば幸いです。

エンジニアのワークスタイル、ライフプランについて

自宅でフリーランス、個人事業主として活躍されているエンジニアは少なくないでしょう。
請求書・契約書などに自宅の住所を使用するのがはばかられるが致し方ないと諦めていませんか？
そんな方にオススメしたいのがバーチャルオフィスの利活用。
バーチャルオフィスは一言でいうと、ビジネス用の住所を格安で利用できるサービスです。
とはいえ「バーチャルオフィスとは？利用すべき？他にメリットは？」と感じる人も少なくないでしょう。
事務所を借りるほどではないが、プライバシーは保護したいという方にオススメのサービスです。開業届や法人登記にも使えるなど、コスパに優れているのでぜひ検討してみることをオススメします。
また、在宅勤務が中心のエンジニアにとって出会いの機会は貴重です。最近ではマッチングアプリが（データ分析の文脈でも）注目度が高いですが、オススメなのが結婚相談所の利用です。
結婚相談所と聞くと古めかしい響きに感じるかもしれませんが、いまやデータマッチングをもとに効率的・完全オンラインでの出会い完結も可能になっています。
自分が利用者にならずとも、副業として結婚相談所を開業することで、「こういうステータス、振る舞いが成婚につながりやすいのか」というデータ収集にも繋がります。結婚相談所最大手のIBJは利用のみならず副業での開業においても人気が高いです。初心者からでも始められ、データ分析の素養も生かせる副業手段としてオススメです。