Twitterでいいね(かRT)した画像をひたすら保存する

目的

Twitterで流れてくる絵師さんの画像を、よくRTやお気に入り登録しています。そこで、一期一会ではもったいないので、せっかくだから保存してやろう！ということでpythonのコードを書きました。サーバーで常時動かして、対象ユーザーを監視しながらローカルに保存することを目標にしています。

結果

実際に完成したコードを貼ります。

favsaver.py

from requests_oauthlib import OAuth1Session
from datetime import datetime
from time import sleep
import os.path
import urllib.request
import json
import env

user_id = env.USER_ID
CONSUMER_KEY = env.CONSUMER_KEY
CONSUMER_SECRET = env.CONSUMER_SECRET
ACCESS_TOKEN = env.ACCESS_TOKEN
ACCESS_TOKEN_SECRET = env.ACCESS_TOKEN_SECRET

twitter = OAuth1Session(CONSUMER_KEY, CONSUMER_SECRET, ACCESS_TOKEN, ACCESS_TOKEN_SECRET)

def get_image_url_by_id(id):
    url = "https://api.twitter.com/1.1/statuses/show.json"
    res = twitter.get(url, params={'id': id})
    if res.status_code == 200:
        if 'media_url_https' in res.text:
            r = json.loads(res.text)
            try:
                medias = r["extended_entities"]["media"]
            except KeyError:
                return False
            return [m["media_url_https"] for m in medias]
    return False

def dl_image_from_url(urls):
    for url in urls:
        name = url.split("/")
        pathname = "./dl/" + name[-1]
        print(pathname)
        urllib.request.urlretrieve(url + ":large", pathname)

def get_fav_list():
    url = "https://api.twitter.com/1.1/favorites/list.json"
    params = {'screen_name': user_id, 'count': 100}
    res = twitter.get(url, params = params)
    if res.status_code == 200:
        r = json.loads(res.text)
        return [tweets["id"] for tweets in r]
    return False

def get_rt_list():
    url = "https://api.twitter.com/1.1/statuses/user_timeline.json"
    params = {'screen_name': user_id, 'count': 150}
    res = twitter.get(url, params=params)
    if res.status_code == 200:
        r = json.loads(res.text)
        return [tweet["id"] for tweet in r if 'retweeted_status' in tweet]
    return False



def main():
    old_ids = []
    while True:
        # ids = get_fav_list()
        ids = get_rt_list()
        if ids :
            since_id = ids[0]
            for tweet_id in ids:
                if tweet_id in old_ids:
                    continue
                media_url = get_image_url_by_id(tweet_id)
                if not media_url:
                    continue
                else:
                    dl_image_from_url(media_url)
                    sleep(1)
        old_ids = ids
        sleep(300)
        

if __name__ == "__main__":
    main()

APIキーや対象のユーザーIDは、別途env.pyに隔離しておきます。

env.py

CONSUMER_KEY = "xxxxxxxxxxxx"
CONSUMER_SECRET = "xxxxxxxxxxxx"
ACCESS_TOKEN = "xxxxxxxxxxxx"
ACCESS_TOKEN_SECRET = "xxxxxxxxxxxx"
# 対象のユーザーID
USER_ID = "xxxxx"

画像は別ディレクトリdlに保存されます。

道のり

ツイート取得

ひとまず、いいねしたツイートのリストをGETするところからはじめました。エンドポイントはhttps://api.twitter.com/1.1/favorites/list.jsonで、今回はGETするときのパラメータに以下のものを指定します。

• screen_name
  • ユーザーID
• count
  • 取得するいいねしたツイートの件数

いいねしたツイートの、IDのみ抽出しリストで返します。Pythonのリスト内包表記を初めて使いました。また、帰ってくるのはツイートのID順(つまり、新しいツイート順)になってしまいます。

def get_fav_list():
    url = "https://api.twitter.com/1.1/favorites/list.json"
    params = {'screen_name': user_id, 'count': 100}
    res = twitter.get(url, params = params)
    if res.status_code == 200:
        r = json.loads(res.text)
        return [tweets["id"] for tweets in r]
    return False

同様に、RTしたツイートのリストも取得します。エンドポイントはhttps://api.twitter.com/1.1/statuses/user_timeline.jsonで、RTかどうかはユーザータイムライン取得後、retweeted_statusフィールドがあるかどうかで判断できます。こちらはタイムラインの順番で帰ってくるので、RTした順番に結果が帰ってきます。
こちらも同じパラメータを指定します。

• screen_name
  • ユーザーID
• count
  • 取得するツイートの件数

def get_rt_list():
    url = "https://api.twitter.com/1.1/statuses/user_timeline.json"
    params = {'screen_name': user_id, "count": 150}
    res = twitter.get(url, params=params)
    if res.status_code == 200:
        r = json.loads(res.text)
        return [tweet["id"] for tweet in r if 'retweeted_status' in tweet]
    return False

画像URL取得

ツイートのIDから情報を取得し、画像のURLを抽出します。画像が添付されておらず、URLがない場合はFalseを返します。また、一つのツイートに複数の画像が添付されていることも考え、こちらも取得したURLをリストで返します。

def get_image_url_by_id(id):
    url = "https://api.twitter.com/1.1/statuses/show.json"
    res = twitter.get(url, params={'id': id})
    if res.status_code == 200:
        if 'media_url_https' in res.text:
            r = json.loads(res.text)
            try:
                medias = r["extended_entities"]["media"]
            except KeyError:
                return False
            return [m["media_url_https"] for m in medias]
    return False

画像ダウンロード

urllibのurlretrieveを用いて、配下のdlディレクトリに保存します。画像urlの注意点として、末尾に:largeをつけないと原寸大の画像が取れません。
urlで指定されているファイル名をそのまま用いて、保存します。

def dl_image_from_url(urls):
    for url in urls:
        name = url.split("/")
        pathname = "./dl/" + name[-1]
        print(pathname)
        urllib.request.urlretrieve(url + ":large", pathname)

結合

メイン関数でこれらのものを組み合わせます。今回自分はいいねよりRTを多くすることと、RTしたツイートの古さに関係なく、RTした順番でツイートが取得できるので、RTを取得することにしました。
ツイートのIDを保持しておき、2回目のループでは差分のみダウンロードするようにします。

def main():
    old_ids = []
    while True:
        # ids = get_fav_list()
        ids = get_rt_list()
        if ids :
            since_id = ids[0]
            for tweet_id in ids:
                if tweet_id in old_ids:
                    continue
                media_url = get_image_url_by_id(tweet_id)
                if not media_url:
                    continue
                else:
                    dl_image_from_url(media_url)
                    sleep(1)
        old_ids = ids
        sleep(300)

今後の展望

• おそらく最初にツイートのリストを取得した時点で、画像のURLも取り出せることがわかりました。IDを一回取得しそのIDで再度GETを投げるのは効率が良くなく、APIの回数制限に引っかからないように、修正が必要であることを感じました
• Google Driveなどに保存できたら面白いかなと思ってます

最後まで読んでいただきありがとうございました。