推しの画像を集めてTLに持ってくるtwitterBOT Python3
めんどくさいが原動力
TLに推しの画像をかき集めるために絵師様のフォロー専門のアカウントを用意するのがめんどくさかったし、保存すればスマホのストレージも圧迫されるしで、普段使い用のアカウントのTLになんとかして自動で人気の絵ツイを持ってこれないかと考えました。なので、twitterのAPIをしばいて、人気の画像をRTしてついでにローカルに保存するBOTを作りました。やったぜ。プログラムはどこのご家庭にもあるラズパイサーバーで毎日24時間動いています。現在、@agctoolとして稼働しています。好きなVtuberの画像を中心に集めているので、TLに放流したい方は是非。
BOTに対してメンションにコマンドをつけてツイートすることで、設定を追加したり、検索および回収&放流を手動で開始することができます。
例によって内容はアレなコードです。よくわかんないエラーが起きてexceptに引っかかってもpassでガン無視しちゃってるので、いつ死んでもおかしくありません(稼働始めてから半年間まだ死んだことないけど）。コードの使用は自己責任で。コピペで動くかどうかは知りません。悪しからず。保存した画像を他所のSNSに垂れ流したり配布等はしないように（規約に引っかかります）。
ファイルは４つあります。すべて同じ階層に配置しています。
・key ：各種APIキーを保存するファイル
・config.json：設定ファイル
・saved ：画像を保存済みのtweetIDを保存するファイル(最初は空のファイルを用意してください)
・app.py ：BOT本体
上記のファイルと同じ階層にあるdata_storageというフォルダに保存した画像が順次格納されていきます。
仕組みを知りたかったら頑張ってクソ雑コードを読んでください。
key
{ "consumer_key":"*************************", "consumer_secret":"************************************************", "access_token_key":"**************************************************", "access_token_secret":"*********************************************" }
config.json
{ "save_path": "data_storage/", "search": [ { "keyword": "**************", "min_faves": 810, "download": true }, { "keyword": "***********:", "min_faves": 114, "download": true }, //以下同形式のデータ { "keyword": "*********", "min_faves": 514, "download": "true" } ] }
saved
{ "data": [ { "id": *******************, "date": "2021-04-10" }, { "id": *******************, "date": "2021-04-10" }, //以下同形式のデータ { "id": *******************, "date": "2021-04-16" } ], "number": 11149 }
4/16日の時点で11149件の画像を保存しているらしい....
app.py
import tweepy import json import datetime import requests import threading import time #基本設定項目 #管理者のID OPID = ******************* OPNAME = '[使いたい人のアカウントの名前 例：Agchan_Luice]' BOTNAME = '[BOTにするアカウントの名前]' #APIキー読み込み kf = open('key',encoding='utf-8') keys = json.load(kf) kf.close() #設定ファイル読み込み cf = open('config.json',encoding='utf-8') cfg = json.load(cf) save_path = cfg['save_path'] cf.close() #保存済み画像IDリスト読み込み sf = open('saved',encoding='utf-8') saved = json.load(sf) sf.close() #API認証 auth = tweepy.OAuthHandler(keys['consumer_key'], keys['consumer_secret']) auth.set_access_token(keys['access_token_key'], keys['access_token_secret']) api = tweepy.API(auth) #画像収集及びリツイート def collect(): #日付取得 date = datetime.date.today().strftime("%Y-%m-%d") todaytime = datetime.datetime.strptime(date,"%Y-%m-%d") limit = todaytime - datetime.timedelta(days=7) #設定ファイルから読み込み for word in cfg['search']: que = word['keyword'] + ' -RT min_faves:' + str(word['min_faves']) + ' since:' + limit.strftime("%Y-%m-%d") #print(que) #検索 search_result = api.search(q = que) #写真を含むツイートのみ選択 for result in search_result: #保存したかどうか確認 isSavedTweet = False if 'media' in result.entities: for media in result.entities['media']: if media['type'] == 'photo': #保存済みでないツイートだけ選択 f = True #f = False for s in saved['data']: if result.id == s['id']: f = False if f: url = media['media_url_https'] #保存する if word['download'] == 'true': response = requests.get(url) img = response.content filename = save_path + date + '-' + str(saved['number']) + '.jpeg' with open(filename,"wb") as f: f.write(img) saved['number'] += 1 isSavedTweet = True if isSavedTweet: #セーブしたらリツイート try: api.retweet(result.id) except: pass d = {"id":result.id,"date":date} saved['data'].append(d) #一週間以前の古いデータを削除 saved['data'] = [s for s in saved['data'] if datetime.datetime.strptime(s['date'],"%Y-%m-%d") > limit] #書き込み wd = json.dumps(saved,indent=4) ow = open("saved","w",encoding='utf-8') ow.write(wd) ow.close() #管理者のツイートをストリーミング class cmdlistener(tweepy.StreamListener): def on_data(self, data): #ツイート保存いらん #global dir_idx tweet = json.loads(data) if not tweet['retweeted'] and 'RT @' not in tweet['text']: # retweetは取得しない #save_path = "%s/%s.json" % ( query_dir, tweet["id_str"] ) #f = open(save_path, "w") #json.dump(tweet, f) #f.close() if tweet['user']['id'] == OPID: cmdfunc(tweet) return True return True def on_error(self, status): #print( status ) pass def cmd_streaming(): que = ['@' + BOTNAME] l = cmdlistener() def start_stream(): while True : try: stream = tweepy.Stream(auth, l) stream.filter(track = que) except: pass #print( 'target query:', que ) start_stream() def reply(tweet,text:str): try: dst_tw_id = tweet['id'] except: pass try: api.update_status(status = '@' + tweet['user']['screen_name'] + "\n" + text,in_reply_to_status_id = str(dst_tw_id)) except: pass def cmdfunc(tweet): cmd = tweet['text'].split() #print(cmd) try: if cmd[1] == 'srwadd': try: q = cmd[2] f = int(cmd[3]) if type(q) != str: reply(tweet,'error:invalid type' + type(q)) return -1 if type(f) != int: reply(tweet,'error:invalid type' + type(f)) return -1 cft = open('config.json',"w",encoding='utf-8') tf = False if len(cmd) >= 5: if cmd[4] == 'false': tf = True if tf: cfg['search'].append({'keyword':q,'min_faves':f,'download':'false'}) else: cfg['search'].append({'keyword':q,'min_faves':f,'download':'true'}) cft.write(json.dumps(cfg,indent=4)) cft.close() reply(tweet,"設定を追加しました。") return 0 except: reply(tweet,"usage : srwadd \'queue\' \'min_faves\'") return -1 if cmd[1] == 'searchm': reply(tweet,"更新を実行します。") collect() if cmd[1] == 'force-shutdown': reply(tweet,"強制終了します。") threading.stop() except: reply(tweet,"unknwon error occurred.") #各スレッド設定 class once_h(threading.Thread): def __init__(self, thread_name): self.thread_name = str(thread_name) threading.Thread.__init__(self) def __str__(self): return self.thread_name def run(self): while True: try: collect() except: print('error?') #4時間に１回動作 time.sleep(14400) class cmd_getstream(threading.Thread): def __init__(self, thread_name): self.thread_name = str(thread_name) threading.Thread.__init__(self) def __str__(self): return self.thread_name def run(self): cmd_streaming() def main(): t1 = once_h(thread_name='time') t2 = cmd_getstream(thread_name='every') t1.start() t2.start() thread_list = [t1,t2] for thread in thread_list: thread.join() main()
keyファイルには各種APIキーを記入します。savedファイルの中身は勝手に管理されます。期間を決めて検索するので、その期間中にすでに保存した画像を重複して保存しないためにあるファイルです。
config.jsonの中に、所定の形式で設定を書き込むと、その設定項目すべての検索を４時間に一度行います(あんまり検索項目多いと途中で止まるかも）。"keyword"キーには、検索したい文字列（例:#xxアート)を入れます。downloadの項目は、trueにしておくとRTと画像の保存をします。falseにすると、RTのみを行って画像は保存しません。min_favesの項目には、検索に引っ掛ける最低いいね数を0以上の数値で設定します。怒られそう。申し訳なさすぎるけどこれがないとヒット数多すぎるので勘弁して......。config.jsonの設定は、コマンドで追加できます。
例:srwadd #xxxArt 1000 true #xxxArtのうちいいねが1000以上で検索、画像の保存を有効（trueをfalseにすると画像が保存されずRTのみ行われます)
TL上でのコマンドは、app.pyの上の方で設定してるOPIDと名前が一致しているユーザーの言うことしか聞きません。コマンドは増やそうと思えば増やせるけどめんどくさくて放置しています。今のままで十分当初ほしかった機能を果たしてくれています。
みなさんも任意のBOTを書いてタイムラインを豊かにしましょう。
