Go to Qiita Advent Calendar 2024 Top
LoginSignup
4
2

Delete article

Deleted articles cannot be recovered.

Draft of this article would be also deleted.

Are you sure you want to delete this article?

More than 5 years have passed since last update.

@y_uehara1011(Yasuhiko UEHARA)

Twython3.7.0でDM取得してみる（廃止されたTwitterのUserStreamをpollingで書き直してみる）

Last updated at Posted at 2018-10-17

背景

以前の記事

で書いたように、Twitter APIでリアルタイムにタイムライン（ダイレクトメッセージ含む）を取得するために使用していたUser Streamが廃止になってしまった。（Twitter社の方針）

新たにAccount Activity APIというのを使えば同様の動作もできるらしいので、Twitter社にAPI使用の申請も済ませたけど考えてみたら自分しか使わないBotにそんな応答性は必要ない気がしてきた。

→ Pollingで1分くらいおきにダイレクトメッセージを見に行って、DMが届いていたら応答するような簡単Botに改造してみようと思う

現状

Botを作成したときの記事 → Raspberry PiをTwitter botにして家のグローバルIPを調べさせる

現状のソースコードはTwythonを使ってこんな感じ。

#!/usr/bin/python3

from twython import TwythonStreamer
from twython import Twython

import requests

consumer_key = '*****'
consumer_secret = '*****'
access_token = '*****'
access_token_secret = '*****'

my_screen_name = 'aoemon1'

twitter = Twython(consumer_key, consumer_secret, access_token, access_token_secret)


class MyStreamer(TwythonStreamer):
  # 何かイベントが起きるとon_success()が呼ばれるのでオーバーライドしておく
  def on_success(self, data):
    # DMだった場合、"direct_message"というkeyが含まれる
    if 'direct_message' in data:
      #print(data.keys())
      msg = data['direct_message']
      #print(msg.keys())
      # 送り主が自分だった場合は"ハウリング"するので除外
      if msg['sender_screen_name'] != my_screen_name:
        ret_msg = self.dispatch(msg['text'])
        # コマンド解釈部に送る
        self.send_dm(msg['sender_screen_name'], ret_msg)

  def on_error(self, status_code, data):
    print(status_code, data)

  # DMを送り返す
  def send_dm(self, target, msg_text):
    twitter.send_direct_message(screen_name = target, text = msg_text)

  # コマンド解釈部。将来はここを拡張する
  def dispatch(self, cmd_text):
    # bot終了
    if cmd_text == 'exit':
      exit()
    # デバッグ用。echoで始まるメッセージはオウム返しする
    elif cmd_text.startswith('echo '):
      return cmd_text[5:]
    # "ip"と聞かれたらグローバルIPを調べて返答する
    elif cmd_text == 'ip':
      return requests.get('http://inet-ip.info/ip').text

if __name__ == "__main__":
  stream = MyStreamer(consumer_key, consumer_secret, access_token, access_token_secret)
  stream.user()

User Streamingが廃止された現状の動作はどうかなと実行してみると,

410 b'The Site Streams and User Streams endpoints have been turned off. Please migrate to alternate APIs. See https://t.co/usss\r\n'
410 Unable to decode response,                                       not valid JSON.

うむ。やっぱりだめみたい。

Twythonでpollingしてみる

pollingで使うならわざわざライブラリを噛まさなくてもいいとは思うが、とりあえずTwythonでDMを取りに行くAPIがあるかどうか公式（ https://twython.readthedocs.io/en/latest/api.html ）を当たってみる。

ざっと見た感じget_direct_messages()でいけそう。
と思ったら

twython.exceptions.TwythonError: Twitter API returned a 404 (Not Found), Sorry, that page does not exist.

このAPI廃止されたのかも？と思って色々調べてみたら、古いDirect MessageのAPIは2018年6月を持って廃止されたらしい。で、Twythonは最新版の3.7.0でもまだ追随できていない。
これは結構めんどくさいことになりそうだけど公式のドキュメント見ながら色々やってみることにする。

Twitter公式のドキュメントを読んでみる

Twitter公式
https://developer.twitter.com/en/docs/direct-messages/api-features

GET direct_messages/events/list
https://developer.twitter.com/en/docs/direct-messages/sending-and-receiving/api-reference/list-events

この辺を読むと、

https://api.twitter.com/1.1/direct_messages/events/list.json

に取りに行けば普通にDMが取得できる気がする。従来のDMも似たような形式だったはずだし単にURLが変わっただけでは？と推測。
TwythonのDM取得関数のURLだけ書き換えればいけるような。

Twythonのソースを探ってみる

pip経由でインストールしたPythonライブラリのソースは~/.local/lib/python3.5/site-packages/twythonにあるので、ここを探っていく

ユーザが叩くAPIはendpoints.pyにあるので、使おうと思っていたget_direct_messages()を見ると

endpoints.py
    # Direct Messages
    def get_direct_messages(self, **params):
        """Returns the 20 most recent direct messages sent to the authenticating user.
        Docs:
        https://developer.twitter.com/en/docs/direct-messages/sending-and-receiving/api-reference/get-messages
        """
        return self.get('direct_messages', params=params)
    get_direct_messages.iter_mode = 'id'

こんな感じ。get()は各APIが内部で使う共通関数で、api.pyに定義されている。

api.py
    self.api_url = 'https://api.twitter.com/%s'

    ### 略

    def get(self, endpoint, params=None, version='1.1'):
        """Shortcut for GET requests via :class:`request`"""
        return self.request(endpoint, params=params, version=version)

    def request(self, endpoint, method='GET', params=None, version='1.1'):
        ### 略
        # In case they want to pass a full Twitter URL
        # i.e. https://api.twitter.com/1.1/search/tweets.json
        if endpoint.startswith('https://'):
            url = endpoint
        else:
            url = '%s/%s.json' % (self.api_url % version, endpoint)

        content = self._request(url, method=method, params=params,
                                api_call=url)

仕組みは結構単純で、get()にURLを渡して取りに行っているだけの模様。
Twythonの最新版（3.7.0）はget_direct_messages()のURLが最終的には

https://api.twitter.com/1.1/direct_messages.json

になるので、現在のTwitter側の仕様と合わない。

ライブラリ修正

修正は、get_direct_messages()をさくっと

endpoints.py
    # Direct Messages
    def get_direct_messages(self, **params):
        """Returns the 20 most recent direct messages sent to the authenticating user.
        Docs:
        https://developer.twitter.com/en/docs/direct-messages/sending-and-receiving/api-reference/get-messages
        """
        return self.get('direct_messages/events/list', params=params)
    get_direct_messages.iter_mode = 'id'

と書き換えればOK。
get_direct_message()なども同様に動かせると思うけど未確認。

Twitter botの改造

と、ここまで調べてきたが、いざbot本体の改造を進めてみると、
URL以外にもTwitter API側の仕様が変更されてTwythonではそのまま使えない機能が多々あった。
（DM送信など）

Twythonのソースを見てると、これRequests使って直接書いても大して変らないんじゃない・・・？と思い始める。
というわけでTwythonやめてRequestsで直接Twitter API叩いてみました

4
2
0

Register as a new user and use Qiita more conveniently

  1. You get articles that match your needs
  2. You can efficiently read back useful information
  3. You can use dark theme
What you can do with signing up
Sign upLogin
4
2

Delete article

Deleted articles cannot be recovered.

Draft of this article would be also deleted.

Are you sure you want to delete this article?