第1回目のBlog記事と第2回目のBlog記事で、TwitterのAccess TokenとRefresh Tokenを取得し、Refresh Tokenを使ってAccess Tokenを更新するコードを紹介しました。
今回は最終的に行いたかった自分のTweetのアナリティクス情報を取得するコードを紹介します。今回のコードを実行すれば、一般的に公開されていないカウント数「プロフィールへのアクセス数」や「リンククリック数」も取得できます。
※Tweet毎の「エンゲージメント」「新しいフォロワー数」の取得方法が現段階では分かっていません。わかり次第、記事にしたいと思います。
第1回目のBlog記事を事前に行っておく必要があります。また、Access Tokenを取得後、時間がたった場合、第2回目のBlog記事も実行する必要があります。
第1回目のBlog記事
第2回目のBlog記事
取得できる内容
- tweet内容(tweet_txt)
- impression数(impression_count)
- プロフィールへのアクセス数(user_profile_clicks)
- リンククリック数(url_link_clicks)
- リツーイト数(retweet_count)
- リプライ数(reply_count)
- お気に入り数(like_count)
- quote_count(これがよくわからん)
上記内容を「output」フォルダに「tweet_data_(日時).csv」というcsvファイルとして出力します。
コードサンプル
import os
import json
import requests
import datetime
import csv
import time
ini_json_file = open('./settings.json', 'r')
ini_json_data = json.load(ini_json_file)
os.makedirs("./token/", exist_ok=True)
os.makedirs("./output/", exist_ok=True)
store_token_place = "./token/"
refresh_token_file = store_token_place + "refresh_token.txt"
with open(refresh_token_file,'r', encoding='utf-8-sig',newline="") as f:
refresh = f.read()
def store_tokens(access_token, refresh_token):
store_token_place = "./token/"
access_token_file = store_token_place + "access_token.txt"
refresh_token_file = store_token_place + "refresh_token.txt"
with open(access_token_file,'w', encoding='utf-8-sig',newline="") as f:
f.write(access_token)
with open(refresh_token_file,'w', encoding='utf-8-sig',newline="") as f:
f.write(refresh_token)
dt_now = datetime.datetime.now()
output_file_date = dt_now.strftime('%Y%m%d_%H%M')
output_file = "./output/tweet_data_{}.csv".format(output_file_date)
with open(output_file,'w') as f:
writer = csv.DictWriter(f,["tweet_id","tweet_txt",\
"impression_count","user_profile_clicks","url_link_clicks",\
"retweet_count","reply_count","like_count","quote_count"])
writer.writeheader()
class SaveTweet:
def __init__(self,id,txt,impre,prof_click,link_click,retweet,reply,like,quote,output_file):
self.id = id
self.txt = txt
self.impression = impre
self.prof_click = prof_click
self.link_click = link_click
self.retweet = retweet
self.reply = reply
self.like = like
self.quote = quote
self.output_file = output_file
def csvoutput(self):
contents = {
"tweet_id":self.id,"tweet_txt":self.txt,\
"impression_count":self.impression,"user_profile_clicks":self.prof_click,\
"url_link_clicks":self.link_click,"retweet_count":self.retweet,\
"reply_count":self.reply,"like_count":self.like,\
"quote_count":self.quote}
with open(self.output_file,'a', encoding='utf-8-sig',newline='') as f:
fieldnames = ["tweet_id","tweet_txt",\
"impression_count","user_profile_clicks","url_link_clicks",\
"retweet_count","reply_count","like_count","quote_count"]
writer = csv.DictWriter(f,fieldnames=fieldnames)
writer.writerow(contents)
client_id = ini_json_data["client_id"]
client_secret = ini_json_data["client_secret"]
token_url = "https://api.twitter.com/2/oauth2/token"
data = {
'refresh_token': refresh,
'grant_type': 'refresh_token',
}
response = requests.post(token_url, data=data, auth=(client_id, client_secret))
jsondata = response.json()
#print(jsondata["access_token"], jsondata["refresh_token"])
store_tokens(jsondata["access_token"], jsondata["refresh_token"])
#自分のtwitter ID取得
my_params = {"user.fields": "created_at,description"}
my_headers = {
"Authorization": "Bearer {}".format(jsondata["access_token"]),
}
my_base_url = "https://api.twitter.com/2/users/me"
my_id_res = requests.get(my_base_url, params=my_params, headers=my_headers)
my_data = my_id_res.json()
#print(my_data["data"]["id"])
tweets_params = {
'expansions' : 'author_id',
'tweet.fields': 'created_at,public_metrics',
'user.fields' : 'name',
'max_results' : 15,
}
tweets_headers = {
"Authorization": "Bearer {}".format(jsondata["access_token"]),
}
base_url = "https://api.twitter.com/2/users/{}/tweets".format(my_data["data"]["id"])
get_tweet_res = requests.get(base_url, params=tweets_params, headers=tweets_headers)
get_tweet_data = get_tweet_res.json()
for tweet_data in get_tweet_data["data"]:
tweet_id = tweet_data["edit_history_tweet_ids"]
params = {"tweet.fields":"public_metrics,created_at,non_public_metrics"}
headers = {
"Authorization": "Bearer {}".format(jsondata["access_token"]),
}
analytics_url = 'https://api.twitter.com/2/tweets/{}'.format(tweet_id[0])
analytics_res = requests.get(analytics_url, params=params, headers=headers)
analytics_data = analytics_res.json()
time.sleep(1)
if (analytics_data.get("data")) != None:
txt = analytics_data["data"]["text"]
impre = analytics_data["data"]["non_public_metrics"]["impression_count"]
prof_click = analytics_data["data"]["non_public_metrics"]["user_profile_clicks"]
link_click = analytics_data["data"]["non_public_metrics"].get("url_link_clicks","link_click is None")
retweet = analytics_data["data"]["public_metrics"]["retweet_count"]
reply = analytics_data["data"]["public_metrics"]["reply_count"]
like = analytics_data["data"]["public_metrics"]["like_count"]
quote = analytics_data["data"]["public_metrics"]["quote_count"]
""" debug用
print("text is ",txt)
print("impression is ",impre)
print("prof_click is ",prof_click)
print("link_click is ",link_click)
print("retweet is ",retweet)
print("reply is ",reply)
print("like is ",like)
print("quote is ",quote)
"""
tweet_data = SaveTweet(tweet_id,txt,impre,prof_click,link_click,retweet,reply,like,quote,output_file)
tweet_data.csvoutput()
else:
#print(list(analytics_data)[0])
print("tweet id <{}> can't get tweet".format(tweet_id))
注意点
- 第1回目のBlog記事と同様にsettings.jsonファイルを準備しておいてください。
- 一般的に公開されていないカウント数が取得できるのは、自分のTweetのみなので、ほかの人のTweetや、ほかの人のTweetに対する返信などはデータが取得できません。
- 他の人へのリプライ含めて15記事取得するようにしています。「tweets_params」の「max_results」の数で調整してください。
最後にPythonの書籍のご紹介です。
この書籍、最後になるにつれて、内容が難しくなったりしているのですが、「Pythonとは」ということだけでなく、プログラミングをしながら仕事をするための基礎について網羅的に書かれているのは参考になります。ただ。初心者にはちょっと難しい気がするので、Pythonを触って少し経ってからのほうが良いかもしれません。
いかがでしたでしょうか? 参考になったら嬉しいです。
コメント