twitter apiでのツイート取得ができない

Question

twitter apiを用いてツイートの取得を目指しておりますこちらのサイトにあったサンプルコードを実行しましたが, 400errorを吐きうまくいきません．consumer keyとapi keyを同一の物として扱ってよいのかもわかりません．

エラーは

 if res.status_code != 200:
            raise Exception('Twitter API error %d' % res.status_code)

の部分に出ております．コード全体は文字数制限上記載できないため, エラーの出た部分までを記載しており, 参考元のページのほうが正確です．

エラー内容

Traceback (most recent call last):
  File "conversation_py3.py", line 334, in <module>
    reset  = checkLimit(session) 
  File "conversation_py3.py", line 238, in checkLimit
    raise Exception('Twitter API error %d' % res.status_code)
Exception: Twitter API error 400

400 errorはリクエストが無効になっているらしいのですが, 原因がよくわからず苦戦しています　知識のある方教えていただけると幸いです

                                                                          *
# tweet_idから時刻情報を取得する                                               *
#                                                                              *
#*******************************************************************************    
def tweet_id2time(tweet_id) :
    id_bin = bin(tweet_id>>22)
    tweet_time=int(id_bin,2)
    tweet_time += 1288834974657
    return tweet_time
 
def getTweet(res,start_time,reset):
    res_text = json.loads(res.text)
    url1 = 'https://api.twitter.com/1.1/statuses/user_timeline.json'    #今回こちらは使わない
    url2 = 'https://api.twitter.com/1.1/statuses/lookup.json'

    cnt_req = 0
    max_tweet=start_time

    total_text = []                           # tweet本文（発話／応答）のリスト
    tweet_list = []                           # n_reply_to_status_idと応答tweetの対のリスト      
    for tweet in res_text['statuses']:
        status_id = tweet['in_reply_to_status_id_str']
        tweet_id=tweet['id']                  # 応答tweetのid

        if status_id != None :               # 当該tweetが応答かどうかの判断

            tweet_time = tweet_id2time(tweet_id)
            if tweet_time <= start_time :    # 前回処理より新しいtweetのみ処理する
                continue

            if max_tweet < tweet_time :
                max_tweet = tweet_time

            res_sentence = tweet['text']
            #RTを対象外にする
            if res_sentence[0:3] == "RT " :
                continue

            res_sentence = screening(res_sentence)
            if res_sentence == '' :
                continue

            tweet_list.append([status_id,res_sentence])


    if len(tweet_list) == 0 :
        return max_tweet,cnt_req ,total_text

    #複数status_idを連結する   
    id_list = tweet_list[0][0]
    for i in range(1,len(tweet_list)) :
        id_list += ','
        id_list += tweet_list[i][0]


    #複数status_id指定で発話tweet取得
    unavailableCnt = 0
    while True :
        try :
            req = session.get(url2, params = {'id':id_list ,'count':len(tweet_list)})
        except SocketError as e:
            print('ソケットエラー errno=',e.errno)
            if unavailableCnt > 10:
                raise

            waitUntilReset(time.mktime(datetime.datetime.now().timetuple()) + 30)
            unavailableCnt += 1
            continue

        if req.status_code == 503:
            # 503 : Service Unavailable
            if unavailableCnt > 10:
                raise Exception('Twitter API error %d' % res.status_code)

            unavailableCnt += 1
            print ('Service Unavailable 503')
            waitUntilReset(time.mktime(datetime.datetime.now().timetuple()) + 30)
            continue

        unavailableCnt = 0

        if req.status_code == 200 :
            req_text = json.loads(req.text)
            break
        else :
            raise Exception('Twitter API error %d' % res.status_code)    

    # 発話tweet本文スクリーニング
    for i in range(0,len(tweet_list)) :
        for j in range(0,len(req_text)) :
            if req_text[j]['id_str'] == tweet_list[i][0] :
                req_sentence = req_text[j]['text']

                if len(req_text) <= 0 :
                    print(req_text)
                    continue

                req_sentence = req_text[j]['text'] 
                #RTを対象外にする
                if req_sentence[0:3] == "RT " :
                    continue

                req_sentence = screening(req_sentence)

                #スクリーニングの結果、ブランクだったら対象外
                if req_sentence == '' :
                    continue   
                # 発話tweetと応答tweetを対で書き込み
                if req_sentence != tweet_list[i][1] :      
                    total_text.append("REQ:"+req_sentence)
                    total_text.append('RES:'+tweet_list[i][1])
                    cnt_req += 1

    max_tweet = max(max_tweet,start_time)
    return max_tweet,cnt_req ,total_text


# tweet本文スクリーニング

# In[2]:

    
def screening(text) :
    s = text

    #RTを外す
    if s[0:3] == "RT " :
        s = s.replace(s[0:3],"")
    #@screen_nameを外す
    while s.find("@") != -1 :
        index_at = s.find("@")
        if s.find(" ") != -1  :
            index_sp = s.find(" ",index_at)
            if index_sp != -1 :
                s = s.replace(s[index_at:index_sp+1],"")
            else :
                s = s.replace(s[index_at:],"")
        else :
            s = s.replace(s[index_at:],"")

    #改行を外す
    while s.find("\n") != -1 :
        index_ret = s.find("\n")
        s = s.replace(s[index_ret],"")

    #URLを外す
    s = re.sub(r'https?://[\w/:%#$&?()~.=+\-…]+', "", s)
    #絵文字を「。」に置き換え その１
    non_bmp_map = dict.fromkeys(range(0x10000, sys.maxunicode + 1), '。')
    s = s.translate(non_bmp_map)
    #絵文字を「。」に置き換え　その２
    s=''.join(c if c not in emoji.UNICODE_EMOJI else '。' for c in s  )

    #置き換えた「。」が連続していたら１つにまとめる
    while s.find('。。') != -1 :
        index_period = s.find('。。')
        s = s.replace(s[index_period:index_period+2],'。')

    #ハッシュタグを外す
    while s.find('#') != -1 :
        index_hash = s.find('#') 
        s = s[0:index_hash]

    return s


# 回数制限を問合せ、アクセス可能になるまで wait する

# In[3]:



def checkLimit(session):
    unavailableCnt = 0
    url = "https://api.twitter.com/1.1/application/rate_limit_status.json"

    while True :
        try :
            res = session.get(url)
        except SocketError as e:
            print('erron=',e.errno)
            print('ソケットエラー')
            if unavailableCnt > 10:
                raise

            waitUntilReset(time.mktime(datetime.datetime.now().timetuple()) + 30)
            unavailableCnt += 1
            continue

        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')
            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_search,remaining_user, remaining_limit ,reset = getLimitContext(json.loads(res.text))
        if remaining_search <= 1 or remaining_user <=1 or remaining_limit <= 1:
            waitUntilReset(reset+30)
        else :
            break

    sec = reset - time.mktime(datetime.datetime.now().timetuple())
    print(remaining_search,remaining_user, remaining_limit ,sec)
    return reset


def waitUntilReset(reset):
    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 秒

Answer 1

原因はtwitter apiが2021年にv2が主要になり, 最近登録した自分もv2だったが, サンプルコードはv1用のものであったことでした．主要な関数の仕様が変更されており, 実行できません．

Answer 2

「リクエストが無効になっている」と記載されておりますが…
厳密には「無効なリクエストが投げられている」です。

つまり、リクエストを投げるところまではよくて、投げられたリクエストの内容に問題がありそうな気がしています。

■よくあること
①アクセストークンや、セキュリティキーを誤って設定している。
②存在しないオプションを設定している。（APIのアップデートなどが原因）

もとの記事を見たところかなり古いものと思われます。
もしかしたら実行の仕方が変わっているのかもしれません。

■対応策
・アクセストークン、セキュリティキーなどの固有情報の見直し。
→誤字脱字もそうですが、指定の仕方も合わせて確認する。

・まずは簡単な1行コードから動作を確認する
→状況の切り分けが必要、本当にこのコマンドが原因か、など。

・実行コマンドのフォーマットを確認する
→コマンドでググれば公式に飛べるでしょう、実行時に指定が必要なものをご確認ください。

Comments

[退会済みユーザー]

回答ありがとうございます！keyを修正したところ, errorが403に変わりました．またtwitter apiを使ったもっと簡易的なコードをこちらの（https://www.learning-nao.com/?p=1985）サイトでみつけ, 実行した際にも同様のエラーが出ました.調べるとCallback URLsというものが関係しているらしいのですが, 修正の方法がわからず, 苦戦しています.　api keyを使うコードにはエラーが出て, bearer keyを使うコードでは出ないため, それらも何かしら関係しているのかなと調べています.　良ければ教えていただけると幸いです

[haihaikazuma]

すいませんが私はPythonも使ったことがなく知識に明るいわけではありません。

共有いただいたサイトを確認したところ、サンプルコードの8行目に以下があります。
auth = tweepy.OAuthHandler(API_Key, API_Sec)

こちらに、callback_urlを追加して…
auth = tweepy.OAuthHandler(API_Key, API_Sec, Callback_Url)

…の形になるのではないでしょうか。

私はTwitter APIの開発者ページを知らないので存在するか分かりませんが、アクセストークンを発行する開発者ページのようなものはありますか？もしかしたらそこからcallback_urlが分かるかもしれません。