Pythonで特定のツイートを定期取得する方法について(今回はチェンソーマン関連のツイート！)【無料版】
はじめに
どうも、pipotteruと申します。
初投稿になります。
今回の記事は、ずばり「Pythonで特定のツイートを取得する方法」になります。
ネットをみればいくらでもやられているネタだと思いますが、ご容赦ください...。
今回はチェンソーマンと書かれたツイートを取得していきたいと思います。
なぜチェンソーマンなのかと言いますと、私がめちゃくちゃ好きだからです！！！
アニメ化まじで楽しみ、MAPPA様よろしく頼むぜ...!!!(PV最高でした)
今回の内容は無料版(本記事)と有料版があります。
無料版では次の段落以降で取得する方法を全てお伝えします。
有料版(100円)では、①ツイート取得用のスクリプトとデータ整形用のスクリプトと②チェンソーマンのツイートデータ(約2週間分)を共有いたします。
実際のツイートデータがどうなっているのか興味ある方・チェンソーマンのツイートにご興味ある方・私におやつを買ってくれる方は是非とも購入をご検討ください！100円ですので！！（切実）
有料版の記事はコチラ⇛【記事完成次第記載します】
※重ね重ねになりますが、無料版のこの記事でも特定のツイートを取得する方法はまとめております！
本題
手順は大まかに４つあります。
１．TwitterAPIを利用するために、各種キーを申請する。
２．TwitterAPIを用いて投稿データを取得する。
３．取得したデータをCSV形式で保存する。
４．定期取得するためにscheduleライブラリを駆使する。
手順１：TwitterAPIを利用するために、各種キーを申請する
初っ端から先人のお力をお借りします笑
コチラの記事がわかりやすいので、参考にしてください！
https://techacademy.jp/magazine/51411
以下４種類のキーを取得してください
- Access Token
- Access Token Secret
- API Key
- API Secret
手順２：TwitterAPIを用いて投稿データを取得する
キーを用意できたら早速取得していきましょう！！
下記のコードに各種のTokenを入れて実行すると、早速1500件分のツイートを取得してCSV形式で保存できます。
import pandas as pd
import json
from time import sleep
from requests_oauthlib import OAuth1Session
import datetime
# 事前に取得したKeyを入力
Consumer_key = 'hogehogehogehoge'
Consumer_secret = 'hogehogehogehoge'
Access_token = 'hogehogehogehoge'
Access_secret = 'hogehogehogehoge'
keyword = "チェンソーマン exclude:retweets" # 検索キーワード リツイートを除く
url = "https://api.twitter.com/1.1/search/tweets.json"
twitter = OAuth1Session(Consumer_key, Consumer_secret, Access_token, Access_secret)
# 収集に使うパラメタ
max_id = -1
count = 100 # 一度に100件ずつ
params = {'q' : keyword, 'count' : count, 'max_id' : max_id, 'lang' : 'ja', 'tweet_mode' : 'extended'}
# 保存先データフレームの初期化
df_out = pd.DataFrame()
# 1500件のツイートを取得
for l in range(15):
if max_id != -1: # ツイートidを既に格納したツイートよりも遡る
params['max_id'] = max_id - 1
req = twitter.get(url, params = params)
if req.status_code == 200: # 正常に取得できていれば
search_timeline = json.loads(req.text)
if search_timeline['statuses'] == []: # 全ツイートを取り終えたら
print("break", datetime.datetime.now())
break
for tweet in search_timeline['statuses']:
# DFへ格納
df = pd.DataFrame(tweet.values(), index=tweet.keys()).T
df["created_at"] = pd.to_datetime(df['created_at']) + datetime.timedelta(hours=9)
df["created_at"] = datetime.datetime.strftime(df["created_at"][0], "%Y/%m/%d %a %H:%M:%S")
df_out = df_out.append(df)
max_id = search_timeline['statuses'][-1]['id']
else: # アクセス頻度制限に引っかかった場合15分待つ
print("Total", df_out.shape[0], "tweets were extracted", sep=" ")
print("break", datetime.datetime.now())
break
df_out = df_out.reset_index(drop=True)
df_out.to_csv("./sample_CSM.csv", index=False, encoding="utf-8-sig")検索ワードは下記の様に設定しています
keyword = "チェンソーマン exclude:retweets" # 検索キーワード リツイートを除く検索ワードは"チェンソーマン"にしました。
これにより"チェンソーマン"という単語を含むツイートを取得してくれます。
また、"exclude:retweets"を入れることでリツイートされたツイートは取得しないようになりますので、必須です。
加えて、tweet_mode=’extended’も重要です。
ツイートの全文を取得できない不具合があるらしいのですが、この'extended'にすることで、テキストを全文取得できます。
次にツイートの取得部分についてです。
該当部分になります。(上のコードの後半部分です)
# 1500件のツイートを取得
for l in range(15):
if max_id != -1: # ツイートidを既に格納したツイートよりも遡る
params['max_id'] = max_id - 1
req = twitter.get(url, params = params)
if req.status_code == 200: # 正常に取得できていれば
search_timeline = json.loads(req.text)
if search_timeline['statuses'] == []: # 全ツイートを取り終えたら
print("break", datetime.datetime.now())
break
for tweet in search_timeline['statuses']:
# DFへ格納
df = pd.DataFrame(tweet.values(), index=tweet.keys()).T
df["created_at"] = pd.to_datetime(df['created_at']) + datetime.timedelta(hours=9)
df["created_at"] = datetime.datetime.strftime(df["created_at"][0], "%Y/%m/%d %a %H:%M:%S")
df_out = df_out.append(df)
max_id = search_timeline['statuses'][-1]['id']
else: # アクセス頻度制限に引っかかった場合15分待つ
print("Total", df_out.shape[0], "tweets were extracted", sep=" ")
print("break", datetime.datetime.now())
break一回に取得できる件数は100件程度と上限があるため、
for文で100件ずつのツイートを15回取得することで合計1500件のツイートを取得しています。
またAPIの呼び出し回数にも制限があるため、制限時の対処としてelse文以下があります。
ツイートを取得できた際は、(=status_codeが200の際は)
Pandasライブラリを使ってデータフレームにデータを格納します。
手順３：取得したデータをCSV形式で保存する。
ツイート情報を格納してCSV形式で出力するのは以下の部分です。
df_out = df_out.reset_index(drop=True)
df_out.to_csv("./sample_CSM.csv", index=False, encoding="utf-8-sig")最後の行でCSV形式で出力します。
encoding="utf-8-sig"が重要で、これをつけないと日本語が文字化けします。
CSVの中身はこんな感じです。
各カラムの内容は下記の神記事を参照してください。
https://syncer.jp/Web/API/Twitter/REST_API/Object/Tweet/
手順４：定期取得するためにscheduleライブラリを駆使する。
手順２と手順３で手動実行ができるようになりました。
最後に定期実行の方法をお伝えいたします。
下記のコードになります。
import pandas as pd
import json
import schedule
from time import sleep
from requests_oauthlib import OAuth1Session
import datetime
from datetime import date, timedelta
import sys
def job():
"""
main内で繰り返すプログラム
"""
NOW = datetime.datetime.now()
keyword = "チェンソーマン exclude:retweets" # 検索キーワード リツイートを除く
print("start!", NOW)
print("検索語", keyword)
DIR = '../data/get_rawdata/' # 事前に用意したディレクトリに保存
# 事前に取得したKeyを入力
Consumer_key = 'hogehogehogehoge'
Consumer_secret = 'hogehogehogehoge'
Access_token = 'hogehogehogehoge'
Access_secret = 'hogehogehogehoge'
url = "https://api.twitter.com/1.1/search/tweets.json"
twitter = OAuth1Session(Consumer_key, Consumer_secret, Access_token, Access_secret)
# 収集に使うパラメタ
max_id = -1
count = 100
params = {'q' : keyword, 'count' : count, 'max_id' : max_id, 'lang' : 'ja', 'tweet_mode' : 'extended'}
# 保存先データフレームの初期化
df_out = pd.DataFrame()
# while(True):
for l in range(15):
if max_id != -1: # ツイートidを既に格納したツイートよりも遡る
params['max_id'] = max_id - 1
req = twitter.get(url, params = params)
if req.status_code == 200: # 正常に取得できていれば
search_timeline = json.loads(req.text)
if search_timeline['statuses'] == []: # 全ツイートを取り終えたら
print("break", datetime.datetime.now())
break
for tweet in search_timeline['statuses']:
# DFへ格納
df = pd.DataFrame(tweet.values(), index=tweet.keys()).T
df["created_at"] = pd.to_datetime(df['created_at']) + datetime.timedelta(hours=9)
df["created_at"] = datetime.datetime.strftime(df["created_at"][0], "%Y/%m/%d %a %H:%M:%S")
df_out = df_out.append(df)
max_id = search_timeline['statuses'][-1]['id']
else: # アクセス頻度制限に引っかかった場合15分待つ
print("Total", df_out.shape[0], "tweets were extracted", sep=" ")
print("break", datetime.datetime.now())
break
df_out = df_out.reset_index(drop=True)
thistime = datetime.datetime.strftime(NOW, '%Y-%m-%d-%H-%M-%S')
df_out.to_csv(DIR + thistime + keyword.replace(" exclude:retweets","") +".csv", index=False, encoding="utf-8-sig")
print(thistime, "Total", df_out.shape, "tweets were extracted!! next start at")
def main():
print("約15分おきに実行、1500件のデータを収集!")
schedule.every(15).minutes.do(job) # データを取得する間隔を変更したければここの行を変更する
try:
while True:
schedule.run_pending()
except KeyboardInterrupt:
print('interrupted!')
# なにか特別な後片付けが必要ならここに書く
sys.exit(0)
if __name__ == '__main__':
main()長文になりましたが、やっていることは以下の２つ。
・定期実行するためにscheduleライブラリを使うmain関数
・これまでの手動実行の内容がメインのjob関数
main関数の処理を簡単に説明します。
※job関数は手順２と３とほぼ同じ内容なので割愛とさせていただきます。
scheduleライブラリを使うことでjob関数の内容について定期実行の詳細な設定ができます。
main関数内でschedule.every(15).minutes.do(job)とすることで15分毎にjob関数を実行します。
また、下記のコードでjob関数の実行監視、指定次官になったらjob関数を実行してくれます。
while True:
schedule.run_pending()
time.sleep(1)TwitterAPIを使う上でのポイントはAPIの使用制限に引っ掛からないようにすることです。
確か15分で150回程度のリクエストしかできなく(定かではないのでかなり怪しい話です、すみません....)、制限に掛かると15分程度使用できなくなります。
一回の実行で取得できるツイートに限度があるということは覚えといてください。
私の体感としては15分程度の定期実行で約2万ツイートを取得しようとする辺りから使用制限に引っかかった記憶があります。
最後に
今回の記事は以上になります。
この記事を読んで、ツイート取得の参考になれば幸いです。
好きな漫画やアニメのツイートをどしどし取得しちゃいましょう。
冒頭でも述べましたが、
有料版の記事では①ツイート取得用のスクリプトとデータ整形用のスクリプトと②チェンソーマンのツイートデータ(約2週間分)を共有いたしますので、そちらもよろしければご購読いただければと思います！(有料版の記事に記載したOneDriveのURLで共有する予定です。)
ではでは、またネタが出来上がったら出没したいと思います！
(まずは有料版用の記事を書かねば,,,)
今回の記事のヘッダー画像はgolchiki様から拝借いたしました！ありがとうございました！！
-----------------------------------------------------------------------------
【自己紹介】
IT系のことをメインに呟きます。初心者です。
PythonとかWordPressネタとかが多いと思います。
オードリーの動画まとめサイト(https://matome-odozehi.com)を運営しています。日々開発中。
Twitterもやってます ⇛ https://twitter.com/pipotteru
作曲用のアカウントはコチラ ⇛ https://twitter.com/tterusic
-----------------------------------------------------------------------------
この記事が気に入ったらサポートをしてみませんか？