tweepy+bottleを使ってAccout Activity APIに対応する

はじめに

User Sterams APIを前提としたtweepyでbotを作っていました。
User Streams API廃止と聞いて凹んでましたが
bottleを使ってちょっと頑張ればAccount Activity APIに対応できるんじゃないかと思って
実装してみました。

何をするか

1. 開発者アクセス権の登録
2. アプリの登録
3. 開発環境の登録
4. Herokuとかでアプリを登録
5. bottle本体を実装
6. webhook用URLをツイッター認証する
7. webhook用URLにwebhook登録したいユーザーを登録する

1-3はこちらを参考に。（結構めんどい）
https://qiita.com/sbtseiji/items/7957de5db0987d9a6891

4はHeroku + bottleで適当にググると見つかります。

本記事では５以降を説明します。

5.bottle本体の実装

認証用処理はgetで、Account Activity API処理はpostでくるので分けてあげるのがポイント

app.py

# -*- coding: utf-8 -*-
import tweepy, os, bottle, sys, random, urllib, datetime, hashlib, hmac, base64, logging, json
from beaker.middleware import SessionMiddleware
from time import sleep
from pytz import timezone
from http import HTTPStatus

# Challenge-Responseチェック通過
@bottle.get('/webhook')
def twitterCrcValidation():

    crc = bottle.request.query.get('crc_token')

    # webhookと紐付けたCS
    validation = hmac.new(
        key=bytes(CS, 'utf-8'),
        msg=bytes(crc, 'utf-8'),
        digestmod=hashlib.sha256)
    digested = base64.b64encode(validation.digest())
    response = {'response_token': 'sha256=' + format(str(digested)[2:-1])}
    print('responding to CRC call')

    return json.dumps(response)


# webhookへの対応
@bottle.post('/webhook')
def twitterEventReceived():

    requestJson = bottle.request.json

    # 自分へのreplyに対しての処理
    if 'tweet_create_events' in requestJson.keys():

        # tweet_create_eventsの中身は通常のTweet Object
        status = requestJson['tweet_create_events'][0]

        # replyイベントを絞りつつ、自分あての返信でないものは無視をする
        if (status['in_reply_to_user_id_str'] != ID_ME):
            return "OK"

        auth = tweepy.OAuthHandler(CK, CS)
        auth.set_access_token(AT, AS)
        api = tweepy.API(auth)

        # sender_idとsender_nameとsender_screen_nameとtextはこんな風に拾う
        sender_id = status['user']['id']
        sender_name = status['user']['name']
        sender_screen_name = status['user']['screen_name']
        text = status['text']

        # 色んな処理をここに書く
        〜〜〜〜

        # 返信
        api.update_status(
                status=tweet,
                in_reply_to_status_id=status['id'],
                auto_populate_reply_metadata=True)


    elif 'direct_message_events' in requestJson.keys():
        #DM に対しての処理
        eventType = requestJson['direct_message_events'][0].get('type')
        messageObject = requestJson['direct_message_events'][0].get(
            'message_create', {})
        messageSenderId = messageObject.get('sender_id')

        #DM送信イベントでなければ排除
        if (eventType != 'message_create'):
            return "OK"

        #自分からのDMは排除
        if (messageSenderId == ID_ME):
            return "OK"

        auth = tweepy.OAuthHandler(CK, CS)
        auth.set_access_token(AT, AS)
        api = tweepy.API(auth)

        # sender_idとsender_nameとsender_screen_nameとtextはこんな風に拾う
        sender_id = messageObject['sender_id']
        sender_name = requestJson['users'][sender_id]['name']
        sender_screen_name = requestJson['users'][sender_id]['screen_name']
        text = messageObject['message_data']['text']

        # 色んな処理をここに書く
        〜〜〜〜

        # DM返信
        api.send_direct_message(sender_id, text=text)

    elif 'follow_events' in requestJson.keys():
        #自動フォロー
        source = requestJson['follow_events'][0]['source']
        if (source['id'] == ID_ME):
            return "OK"

        auth = tweepy.OAuthHandler(CK, CS)
        auth.set_access_token(AT, AS)
        api = tweepy.API(auth)

        # フォロー返す
        api.create_friendship(source['id'])

    else:
        #サポートしてないイベント
        return "OK"

    return "OK"

6. webhook用URLをツイッター認証する

webhook.py

# !/usr/bin/env python
# -*- coding: utf-8 -*-
import os
from requests_oauthlib import OAuth1Session

# botのAccessToken/AccessSecret/CustomerKey/CustomerSecretを環境変数に登録してあります
AT = os.environ["AT"] # アプリ作成者垢のAT
AS = os.environ["AS"] # アプリ作成者垢のAS
CS = os.environ["CS"]
CK = os.environ["CK"]

# webhook登録用のURL(例として、https://hogehoge.herokuapp.com/webhook というURL、ラベルはfugafugaにしている)
url = "https://api.twitter.com/1.1/account_activity/all/fugafuga/webhooks.json?url=https%3a%2f%2fhogehoge%2eherokuapp%2ecom%2fwebhook"

# OAuth認証
twitter = OAuth1Session(CK, CS, AT, AS)

# webhookの登録
req = twitter.post(url)

# レスポンスを確認
if req.status_code == 200:
    print ("OK")
else:
    print ("Error at webhook: %d" % req.status_code)
    print(req.text)

7. webhook用URLにwebhook登録したいユーザーを登録する

6とほとんど同じ。違うのはAT/ASとURLくらい

webhook.py

# !/usr/bin/env python
# -*- coding: utf-8 -*-
import os
from requests_oauthlib import OAuth1Session

# botのAccessToken/AccessSecret/CustomerKey/CustomerSecretを環境変数に登録してあります
AT = os.environ["AT_SUB"] # 購読したい垢のAT
AS = os.environ["AS_SUB"] # 購読したい垢のAS
CS = os.environ["CS"]
CK = os.environ["CK"]

# subscription登録用のURL ラベルはfugafugaにしている
url = "https://api.twitter.com/1.1/account_activity/all/fugafuga/subscriptions.json"

# OAuth認証
twitter = OAuth1Session(CK, CS, AT, AS)

# Account Activity APIをsubscribe
req = twitter.post(url)

# レスポンスを確認
if req.status_code == 200:
    print ("OK")
else:
    print ("Error at subscribe: %d" % req.status_code)
    print(req.text)

まとめ

5をHerokuにデプロイしつつ、６、７を順番に実行してなんとかなりました。
とりあえず様子見。