Python

【TwitterAPI×Python】指定アカウント・キーワードのツイートを取得後CSV出力

本稿執筆の目的

  • タイムリーなユーザーインサイトを誰もが簡単に収集可能にすること
  • プログラミング言語の学習コスト削減
  • Twitter運用をアウトソースするための準備

 

本稿を読むメリット

  • Pythonの基礎知識がなくても、ツイートをスクレイピングできる
  • 指定アカウントのツイート直近3,000件取得できる
  • 指定キーワードを含むツイート直近3,000件取得できる

 

本稿の対象読者

  • SNS上のユーザーインサイトを収集したいと考えている人
  • Twitter Developperの登録が完了している人

※2点目のハードルがやや高いので後日記事化します。

 

実行環境

  • Google Colaboratory(Visual Studio Code等でも可)
  • Python 3.6.6
  • mac Catalina

 

実行方法

step
1
Google Colaboratoryにアクセス

Google Colaboratoryにアクセスします。

 

step
2
ノートブックを新規作成

ノートブックを新規作成

ログインを求められるので、Googleアカウントにログインします。

 

step
3
本稿「実装コード_Python」を貼り付け

Pythonプログラムを実行

 

step
4
Twitter Developperから取得したAPIキーを入力

コード内で空欄になっているCK、CS、AT、ASを埋めます。

※Twitter Developperへの登録が完了していない人は、登録申請が必要です。(別途、記事化します)

 

step
5
取得条件「キーワード」または「アカウント名」を入力

# 検索条件を指定して取得 (screen_nameまたはキーワード)
targets = ['recruiter_take','Syukatsu_Muso']


上記のように、取得対象の「キーワード」または「アカウント名」を入力します。

 

注意ポイント

「キーワード」を指定する場合と「アカウント」を指定する場合で呼び出すメソッドが異なるため、以下のように必要に応じてコメントアウトが必要です。

 

キーワードを指定する場合

以下をコメントアウトします。

getter = TweetsGetter.byUser(target) #ユーザー名(screen_name)検索の場合はこっち

 

アカウントを指定する場合

以下をコメントアウトします。

getter = TweetsGetter.bySearch(target) #キーワード検索の場合はこっち

 

step
6
プログラムを実行する

Pythonプログラムを実行

「再生ボタン」のようなアイコンをクリックすると、プログラムが実行されます。

プログラム実行後、スクレイピングしたツイートがログに出力されていきます。

 

step
7
CSVをダウンロード

CSVダウンロード

プログラムを実行すると、指定した「キーワード」または「アカウント」ごとにCSVが出力されます。

Google Colaboratoryの画面左端にあるサイドバーから「フォルダ」アイコンをクリックし、表示されたダウンロードしたいCSVファイルにマウスオーバーし、ファイル右側の3連粒をクリック。「ダウンロード」をクリックすると、自身のPC(ローカル)に対象ファイルが保存されます。

 

当該プログラムによるアウトプットサンプル

アウトプットサンプル

L列以降は私がスプレッドシート上で追加したものです。K列までが本稿で紹介しているプラグラムで出力されるデータです。

 

実装コード_Python

# -*- coding: utf-8 -*-

from requests_oauthlib import OAuth1Session
import json
import datetime, time, sys, calendar
from abc import ABCMeta, abstractmethod
import sys
import urllib.request

# TwitterAPIキー
CK = ''   # Consumer Key(Twitter Developperで取得)
CS = ''   # Consumer Secret(Twitter Developperで取得)
AT = ''    # Access Token(Twitter Developperで取得)
AS = ''   # Accesss Token Secert(Twitter Developperで取得)

# リダイレクトしないハンドラークラス
class NoRedirectHandler(urllib.request.HTTPRedirectHandler):
  # HTTPRedirectHandler.redirect_request をオーバーライド
  def redirect_request(self, req, fp, code, msg, hdrs, newurl):
    self.newurl = newurl # リダイレクト先URLを保持
    return None

# リダイレクト先 URL を取得する関数
def get_redirect_url(src_url):
  # リダイレクトしないハンドラーをセット
  no_redirect_handler = NoRedirectHandler()
  opener = urllib.request.build_opener(no_redirect_handler)
  try:
    with opener.open(src_url) as res:
      return None # リダイレクトしない URL だった
  except urllib.error.HTTPError as e:
    if hasattr(no_redirect_handler, "newurl"):
      return no_redirect_handler.newurl # リダイレクト先 URL を返す
    else:
      raise e # リダイレクト以外で発生した例外なので投げ直す

def YmdHMS(created_at):
    time_utc = time.strptime(created_at, '%a %b %d %H:%M:%S +0000 %Y')
    unix_time = calendar.timegm(time_utc)
    time_local = time.localtime(unix_time) 
    return time.strftime("%Y/%m/%d %H:%M:%S", time_local)

class TweetsGetter(object):
    __metaclass__ = ABCMeta

    def __init__(self):
        self.session = OAuth1Session(CK, CS, AT, AS)

    @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):
        '''
        ツイート取得を開始する
        '''

        #----------------
        # 回数制限を確認
        #----------------
        self.checkLimit()

        #----------------
        # URL、パラメータ
        #----------------
        url, params = self.specifyUrlAndParams()
        params['include_rts'] = str(includeRetweet).lower()
        # include_rts は statuses/user_timeline のパラメータ。search/tweets には無効

        #----------------
        # ツイート取得
        #----------------
        cnt = 0
        unavailableCnt = 0
        while True:
            res = self.session.get(url, params = params)
            if res.status_code == 503:
                # 503 : Service Unavailable
                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" にする
                # https://dev.twitter.com/discussions/7513
                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/rate_limit_status.json"
            res = self.session.get(url)

            if res.status_code == 503:
                # 503 : Service Unavailable
                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 + 10)  # 念のため + 10 秒

    @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, 'tweet_mode':"extended"}
        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.screen_name = screen_name
        
    def specifyUrlAndParams(self):
        '''
        呼出し先 URL、パラメータを返す
        '''
        url = 'https://api.twitter.com/1.1/statuses/user_timeline.json'
        params = {'screen_name':self.screen_name, 'count':200, 'tweet_mode':"extended"}
        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 = res_text['resources']['statuses']['/statuses/user_timeline']['remaining']
        reset     = res_text['resources']['statuses']['/statuses/user_timeline']['reset']

        return int(remaining), int(reset)


if __name__ == '__main__':

    #ツイート取得日
    today = datetime.date.today()

    # 検索条件を指定して取得 (screen_nameまたはキーワード)
    targets = ['recruiter_take','Syukatsu_Muso','shukatsu_mirai','es_shibamon','saiyoking','beyond_mylife','f_shukatsu25','fight_arai','sasukecareer','eruto1121','topsyuchannel','dossun_jack','gaishi_syukatu1','shukatu_man','unistyleyoppi','unistylehyt','esquestion','unistyleinc','KHsyukatu','syukatsu_mikata','Start_OfferBox','jobhun_app','badassceo','osatoexcel']
    for target in targets:
        getter = TweetsGetter.byUser(target) #ユーザー名(screen_name)検索の場合はこっち
        # getter = TweetsGetter.bySearch(target) #キーワード検索の場合はこっち

        # プロフィール蘭のURL取得フラグ(Trueで取得)
        url_get_flg = False

        # ファイルに保存する
        fname = str(today) + "_twitter_search_target_" + target + ".csv"
        with open(fname, 'w') as fs:
            #ヘッダー部分
            fs.write('ツイートID' + '\t'
            + 'ツイートURL' + '\t'
            + 'ツイート日時' + '\t'
            + 'ツイート本文' + '\t'
            + 'いいね数' + '\t'
            + 'RT数' + '\t'
            + 'ツイートユーザー名' + '\t'
            + 'ツイートユーザーID' + '\t'
            + 'フォロワー数' + '\t'
            + '場所' +  '\t'
            + 'URL' +  '\n')
            cnt = 0
            for tweet in getter.collect(total = 3000):
                if tweet['user']['url'] is None:
                    prof_url = ""
                elif not url_get_flg:
                    prof_url = "非処理"
                elif get_redirect_url(tweet['user']['url']) is None:
                    prof_url = ""
                else:
                    prof_url = get_redirect_url(tweet['user']['url'])
                fs.write(tweet['id_str'] + '\t'
                + 'https://twitter.com/' + tweet['user']['screen_name'] + '/status/' + tweet['id_str'] + '\t'
                + YmdHMS(tweet['created_at']) + '\t'
                + tweet['full_text'].replace('\n', '[改行]') + '\t'
                + str(tweet['favorite_count']) + '\t'
                + str(tweet['retweet_count']) + '\t'
                + tweet['user']['name'] + '\t'
                + tweet['user']['screen_name'] + '\t'
                + str(tweet['user']['followers_count']) + '\t'
                + tweet['user']['location'] + '\t'
                + prof_url +  '\n')
                cnt += 1
                print ('------ %d' % cnt)
                print ('{} {} {}'.format(tweet['id'], tweet['created_at'], '@'+tweet['user']['screen_name']))
                print (tweet['full_text'])
                #

 

  • この記事を書いた人

おさむ

【データ×プログラミングで稼ぐ】Pythonで機械学習を勉強中の30歳。FXや株で死ぬほど稼げる魔法のロジックを作るのが目標。ヘッポコの私には凄く難しいので、ブログ書きながらコツコツやります。趣味は筋トレとダンス。外交型ぽい内向型。今は就活生向けのアプリ開発中。

-Python
-,

© 2021 Osamu Blog Powered by AFFINGER5