質問をすることでしか得られない、回答やアドバイスがある。

15分調べてもわからないことは、質問しよう!

ただいまの
回答率

90.52%

  • Python 3.x

    6387questions

    Python 3はPythonプログラミング言語の最新バージョンであり、2008年12月3日にリリースされました。

PythonでTwitterのツイートを取得したい

受付中

回答 0

投稿

  • 評価
  • クリップ 0
  • VIEW 102

gymgym

score 67

PythonでTwitterのツイートを取得しようと考えています。

その際には、制限があるのでその制限ギリギリまでツイートを取得したいです。

現在、以下のようなコードを用いています。
参考:http://ailaby.com/twitter_api/

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

from requests_oauthlib import OAuth1Session
import json
import datetime
import time
import sys
import csv
import config
from abc import ABCMeta, abstractmethod

CK = config.CONSUMER_KEY
CS = config.CONSUMER_SECRET
AT = config.ACCESS_TOKEN
AS = config.ACCESS_TOKEN_SECRET

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}
        query = '任天堂 exclude:retweets since:2018-07-05'
        params = {
            'q': query, '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.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}
        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__':

    # キーワードで取得
    #getter = TweetsGetter.bySearch(u'任天堂')

    # ユーザーを指定して取得 (screen_name)
    getter = TweetsGetter.byUser('@fashionpressnet')

    cnt = 0
    f = open('test.csv', 'a')
    for tweet in getter.collect(total=10000):
        cnt += 1
        tweet_list = [tweet['created_at'], tweet['text'], tweet['retweet_count'], tweet['favorite_count']]
        writer = csv.writer(f, lineterminator='\n')  # 行末は改行
        writer.writerow(tweet_list)


以下の部分で、ツイートが10000件取得すると終了してしまう仕様となっているのですが、
制限ギリギリいっぱいまで取得できるようにしたいです。

   f = open('test.csv', 'a')
    for tweet in getter.collect(total=10000):
        cnt += 1
        tweet_list = [tweet['created_at'], tweet['text'], tweet['retweet_count'], tweet['favorite_count']]
        writer = csv.writer(f, lineterminator='\n')  # 行末は改行
        writer.writerow(tweet_list)

取得できる範囲で限界まで取得すると終了となるようにしたいのですが、どのようにしたら良いでしょうか。
よろしくお願いいたします。

  • 気になる質問をクリップする

    クリップした質問は、後からいつでもマイページで確認できます。

    またクリップした質問に回答があった際、通知やメールを受け取ることができます。

    クリップを取り消します

  • 良い質問の評価を上げる

    以下のような質問は評価を上げましょう

    • 質問内容が明確
    • 自分も答えを知りたい
    • 質問者以外のユーザにも役立つ

    評価が高い質問は、TOPページの「注目」タブのフィードに表示されやすくなります。

    質問の評価を上げたことを取り消します

  • 評価を下げられる数の上限に達しました

    評価を下げることができません

    • 1日5回まで評価を下げられます
    • 1日に1ユーザに対して2回まで評価を下げられます

    質問の評価を下げる

    teratailでは下記のような質問を「具体的に困っていることがない質問」、「サイトポリシーに違反する質問」と定義し、推奨していません。

    • プログラミングに関係のない質問
    • やってほしいことだけを記載した丸投げの質問
    • 問題・課題が含まれていない質問
    • 意図的に内容が抹消された質問
    • 広告と受け取られるような投稿

    評価が下がると、TOPページの「アクティブ」「注目」タブのフィードに表示されにくくなります。

    質問の評価を下げたことを取り消します

    この機能は開放されていません

    評価を下げる条件を満たしてません

    評価を下げる理由を選択してください

    詳細な説明はこちら

    上記に当てはまらず、質問内容が明確になっていない質問には「情報の追加・修正依頼」機能からコメントをしてください。

    質問の評価を下げる機能の利用条件

    この機能を利用するためには、以下の事項を行う必要があります。

まだ回答がついていません

15分調べてもわからないことは、teratailで質問しよう!

  • ただいまの回答率 90.52%
  • 質問をまとめることで、思考を整理して素早く解決
  • テンプレート機能で、簡単に質問をまとめられる

関連した質問

  • 解決済

    ツイートの投稿

    大学で卒業研究としてTwitterAPIを利用したアプリケーションを開発しています。 その過程で以下のようにツイート投稿のところで行き詰まりました。 同じようなコードでタイムラ

  • 解決済

    Ruby on Railsでツイートを取得する方法

    最近ツイートを取得するアプリを開発しています。 ログイン機能等はできたのですが、肝心のツイートを取得する部分でハマってしまいました。 # app/controllers

  • 受付中

    twitter4jでリプライ先のツイートを取得したい。

    まずやりたいことは、公式ツイッターアプリのように、ツイートをクリックときに、会話の履歴を出したいです。 ☆例☆ 今起きた。 | | @abcde おはようございま

  • 解決済

    Twitter4Jで取得した値をMySQLに格納したい

    Twitter4Jで特定のツイートを取得し、MySQLへ格納したいのですが文字列のまま格納されてしまいます。 https://oshiete.goo.ne.jp/qa/30555

  • 解決済

    Devise内でomniauthのtwitter認証が完了してもレコードが格納されない

    前提・実現したいこと ///0222追記 すみません!そもそも「実現したい事」を書くのを忘れていました! エラーは後述の通り「Twitter認証後データがテーブルに格納されな

  • 解決済

    TwitterKitでツイートできない

    iOS11からTwitterKitを使わないとツイートできなくなったので、インストールしました。 ツイートのためのウィンドウは表示されるのですが、ツイートボタンを押しても「ツイート

  • 解決済

    TwitterAPIとGASで画像つきツイートをしたい

    前提・実現したいこと 下記のサイトを参考にgoogle App Script(GAS)を使用してBotを制作しています.サイトの丁寧な説明もあり,テキストのツイートまではできたので

  • 解決済

    PythonでTwitterのタイムラインを取得

    お世話になります。 PythonでTwitterのタイムラインを取得したいです。 しかし下記のようにtwitter.getが働きません。 タイムラインの取得方法を教えて

同じタグがついた質問を見る

  • Python 3.x

    6387questions

    Python 3はPythonプログラミング言語の最新バージョンであり、2008年12月3日にリリースされました。