概要
フォロワーたちが今日何で盛り上がったのか、今のトレンドはどのようなものなのかがきになったので、それが1目でわかるWordCloud画像を生成することで、今のタイムラインの流れをある程度掴めるツールを作成した。
環境
MacBookPro 13-inch, M1, 2020
macOS Ventura 13.0.1
目次
- ファイル構成
- プログラム全文
- TwitterAPIの申請
- ライブラリのインポート
ファイル構成
TL
├ .env
├ config.py
├ api.py
├ list.py
├ tweet.py
├ creat_image.py
├ tweet.txt
├ tweet_id.txt
├ stop_words.txt
└ result.png
プログラム全文
# 例
API_KEY='sdS5kjFenFDig8UtMwFm5f3JQ'
API_SECRET='5kAxS7n3uKBRCJVVQmqbDGdXkbPz4ej9SS7xYqT4JqhUyqFdha'
CONSUMER_KEY='yHeQ6fNfEpQZnynhw5xdkZMuRFx6TJ3HSKWzuG3typ2RGDL3yA'
CONSUMER_SECRET='mDKwrCaeZrZmJwWLuP9jpCVgtN4cP8gxaxmqkr8LZnfrqF'
SCREEN_NAME='ABCDEF'
LIST_ID='1234567890123456789'
import os
from dotenv import load_dotenv
load_dotenv()
api_key = os.environ.get("API_KEY")
api_secret = os.environ.get("API_SECRET")
consumer_key = os.environ.get("CONSUMER_KEY")
consumer_secret = os.environ.get("CONSUMER_SECRET")
screen_name = os.environ.get("SCREEN_NAME")
list_ID = os.environ.get("LIST_ID")
font_path = "使いたいフォントの絶対パス"
stop_words_path = "stop_words.txtの絶対パス"
tweet_path = "tweet.txtの絶対パス"
result_path = "result.pngの絶対パス"
tweet_id_path = "tweet_id.txtの絶対パス"
import tweepy
import config
auth = tweepy.OAuth1UserHandler(
config.api_key,
config.api_secret,
config.consumer_key,
config.consumer_secret
)
API = tweepy.API(auth, wait_on_rate_limit=True)
import tweepy
import api
import config
ls = tweepy.Cursor(api.API.get_list_members, list_id=config.list_ID).items()
fs = tweepy.Cursor(api.API.get_friends).items()
add_list = []
lists = [l.screen_name for l in ls]
friends = [f.screen_name for f in fs]
for f in friends:
result = f not in lists
if result:
add_list.append(f)
for a in add_list:
api.API.add_list_member(list_id=config.list_ID, screen_name=a)
import tweepy
from janome.tokenizer import Tokenizer
import api
import config
f = open(config.tweet_id_path, "r")
tweet_id = f.readline()
f.close()
tts = tweepy.Cursor(api.API.list_timeline, owner_screen_name=config.screen_name, slug=config.list_ID, since_id=tweet_id).items()
t_lists = []
text = ""
for tt in tts:
twbe = tt.text
tw = twbe.strip().replace(' ', '')
k = 0
while "@" in tw:
for i, t in enumerate(tw):
if t == "@":
k = i
elif t == " ":
tw = tw.replace(tw[k:i + 1], "")
break
elif i == len(tw)-1:
tw = tw.replace(tw[k:], "")
break
while "#" in tw:
for i, t in enumerate(tw):
if t == "#":
tw = tw.replace(tw[i], "")
break
if "RT" in tw:
tw = tw.replace(tw[:2], "")
while "https" in tw:
for i, t in enumerate(tw):
try:
if t == "h" and tw[i + 1] == "t":
tw = tw.replace(tw[i:i + 23], "")
except IndexError:
if t == "h":
tw = tw.replace(tw[i:i + 23], "")
t_lists.append(tw)
t = Tokenizer()
for tw in t_lists:
tts = [token.surface for token in t.tokenize(tw) if token.part_of_speech.startswith('名詞')]
for ts in tts:
text += " " + ts
f = open(config.tweet_path, "a", encoding='utf-8')
f.write(text)
f.close()
t = api.API.list_timeline(owner_screen_name=config.screen_name, slug=config.list_ID, count=1, since_id=tweet_id)
if len(t) != 0:
f = open(config.tweet_id_path, "w")
f.write(str(t[0].id))
f.close()
from pyknp import Jumanpp
from wordcloud import WordCloud
import api
import config
f = open(config.stop_words_path, "r")
word = f.read()
f.close()
stop_words = word.split()
f = open(config.tweet_path, "r")
tweet = f.readline()
f.close()
wc = WordCloud(font_path=config.font_pass, width=1200, height=675, background_color="white", stopwords=set(stop_words), max_words=500).generate(tweet)
wc.to_file(config.result_path)
f = open(config.tweet_path, "w")
f.write("")
f.close()
api.API.update_status_with_media(status="ツイート内容を入力",
filename=config.result_path)
ts = api.API.list_timeline(owner_screen_name=config.screen_name, slug=config.list_ID, count=1)
f = open(config.tweet_id_path, "w")
f.write(str(ts[0].id))
f.close()
TwitterAPIの申請
TwitterAPIの機能を使用するためには、申請が必要になります。
詳しくは【twitterAPI】超簡単！twitterAPI取得方法まとめてみた〜を見ると良いと思います。
英作文が結構きついので参考として申請が通ったメールを共有します。
以下メール本文
Thanks for reply.
The purpose of my service is to measure the frequency of nouns in all tweets that appear in a timeline and to generate and share an image that summarizes the timeline, changing color and size accordingly.
This allows followers to gain a comprehensive understanding of the timeline that they may have missed.
The collected tweets are processed primarily in Python. Specifically, they are stored in an array, broken down into nouns only with janome, and converted into images with wordcloud. The images are then tweeted through the Twitter API.
The Twitter API will be used to collect tweets for image generation and to share images.
We do not plan to use Twitter API for retweeting.
We do not plan to use Twitter API for registering favorites.
I do not plan to use Twitter API for tweeting except for tweeting generated images about once a day.
I do not plan to use the Twitter API to collect tweets other than from accounts that follow me.
I do not plan to use the API to interact with users and will tweet and retweet in the same way as regular users, logging directly into web and smartphone applications.
The content of tweets will be mainly tweeting about daily life.
I will only be active 1-5 times per day.
My service will only work with Twitter and the code I create, and will not display Twitter content outside of Twitter.
Thanks,
NULL
ライブラリのインポート
普通にpip installすれば大丈夫です。
$ pip install tweepy
$ pip install config
$ pip install python-dotenv
$ pip install wordcloud
$ pip install janome
プログラムの解説
プログラム全文に載せたコードを引用し、細部を解説します。
読み飛ばしていただいても構いません.
.env
gitなどに投げる時に見られたくないものを保存しておくファイルです。
API_KEY='sdS5kjFenFDig8UtMwFm5f3JQ'
TwitterAPIのキーをそれぞれ保存します。表示されているのはランダムで生成したダミーなので自分で取得したものと置き換えてください。
SCREEN_NAME='ABCDEF'
TwitterのIDや使用するリストのIDを記録します。
config.py
.envを読み込むためのファイルです。
共有したい変数もここで定義してしまっていますが、気になる方は分けてください。
load_dotenv()
api_key = os.environ.get("API_KEY")
変数名 = os.environ.get("環境変数名")のように記述します。
list.py
ls = tweepy.Cursor(api.API.get_list_members, list_id=config.list_ID).items()
list_idで指定したリストのメンバーを取得するコードです。
json形式で帰ってくるので扱いに気をつけましょう。
fs = tweepy.Cursor(api.API.get_friends).items()
自分がフォローしている人を取得するコードです。
json形式で帰ってくるので扱いに気をつけましょう。
lists = [l.screen_name for l in ls]
friends = [f.screen_name for f in fs]
上で帰ってきた結果を配列にまとめています。
内包表記で書かれているので少しわかりにくいですが、やっていることは上の結果(json形式)のtext欄を参照して値を配列に入れるという動作をその要素数だけ繰り返しているだけなので理解は簡単かと思います。
add_list = []
for f in friends:
result = f not in lists
if result:
add_list.append(f)
ここで実現したいのは、フォローしている人の中からリストに追加されていない人を見つけることです。
よってfor文の繰り返し回数の基準は friends であるべきで、このような形になります。
add_list がリストに追加するべき人を保存する配列であることに注目すればさほど理解に苦労はしないはずです。
for a in add_list:
api.API.add_list_member(list_id=config.list_ID, screen_name=a)
先ほど追加した add_list の全てのユーザーを list_id で指定したリストに入れるコードです。
tweet.py
リストのタイムラインを取得・処理・保存するためのファイルです。
f = open(config.tweet_id_path, "r")
tweet_id = f.readline()
f.close()
テキストファイルの読み込みをしています。
この記事を読む分には
-
r→読み取り,w→書き込み(上書き),a→書き込み(追加) -
readline()→1行を読み取る,read()→全てを読み取る
とだけ覚えておけば大丈夫です。
tts = tweepy.Cursor(api.API.list_timeline,
owner_screen_name=config.screen_name,
slug=config.list_ID,
since_id=tweet_id).items()
owner_screen_name, slug(おそらくどちらかだけで十分)で指定したリストのタイムラインを取得するコード。オプションで since_id=tweet_id とすることで、重複してツイートを取得することを防いでいます。
ここからは取得したツイートを処理に適した形に変えるコードです。
本来は別ファイルで書くべきですがガサツなのでやりません。
for tt in tts:
tw = tt.text.strip().replace(' ', '')
ここでは取得してきたjsonファイルから必要な情報を抜き出しています。改行や空白があると都合が悪いので削除しています。
strip()で改行を削除しています。
replace(' ', '')で空白を削除しています。
k = 0
while "@" in tw:
for i, t in enumerate(tw):
if t == "@":
k = i
elif t == " ":
tw = tw.replace(tw[k:i + 1], "")
break
elif i == len(tw)-1:
tw = tw.replace(tw[k:], "")
break
ここではメンションを取り除く作業をしています。いい方法が浮かばなかったのでゴリ押しています。
メンションの有無は文章中に@が含まれているかでわかるので条件式は"@" in twとなり、2つ以上のメンションに対応するためにwhile文を使用しています。
次のfor文からは、メンションを削除する動作を書いています。enumerateというのは、今閲覧している要素の住所を整数として返してもらうために設定しています。ここではiという変数に代入していますね。
冗長なenumerateの解説
| 0 | 1 | 2 | 3 | 4 |
|---|---|---|---|---|
| あ | い | う | え | お |
という配列があったら
i, t それぞれの変化は次のようになります。
i→ ０, １, ２, ３, ４
t→ あ, い, う, え, お
最初のif文は、削除するべき箇所(メンション)の始まりを保存しておくためのコードです。
ここで、tはツイート本文中の任意の1文字である点に注意してください。
それが@であるということはメンション部分が始まっているので、この値の住所をkという変数に保存しています。
コードを確認すればわかるように、このkがメンション部を削除するためのフラグとなっています。
1つ目のelif文は、返信をした時にできる自然な形のメンションを削除するためのコードです。
メンションの中には空白が含まれませんが、メンションと本文との間に空白が入っていることを利用して、メンションを削除しています。
削除の際には、ツイート本文から@から半角スペースまでの文字tw[k:i + 1]を全て削除しています。
if文の説明を見返せば何をやっているかは簡単にわかるはずです。
2つ目のelif文は、ユーザーが手打ちでツイートの末尾にIDを埋め込んだ時にできるメンションを削除するコードです。
やっていることは上2つと変わりません。
if t == "#":
tw = tw.replace(tw[i], "")
break
ハッシュタグを削除するコードです。メンションと違って1文字消せばいいので楽ですね。
if "RT" in tw:
tw = tw.replace(tw[:2], "")
RTでタイムラインに表示された際に文頭にくるRTを削除するコードです。stop_words.txtにRTを追加するだけで済むことにこんなに文字数をかけていたことにいまさら気づいて絶望しています。
while "https" in tw:
for i, t in enumerate(tw):
try:
if t == "h" and tw[i + 1] == "t":
tw = tw.replace(tw[i:i + 23], "")
except IndexError:
if t == "h":
tw = tw.replace(tw[i:i + 23], "")
urlを共有した際のhttps::~を削除するコードです。
上とほぼ一緒ですが、Twitterで共有されたurlの文字数が決まっていることを利用して削除している点が異なります。
また、あまりいい解決法が浮かばないがシステムに支障をきたさない程度のエラーが発生したのでtryを使って誤魔化しています。というか and以降いらない気もしてきました。
tryについての解説(読まなくても良い)
try: の下にあるコードでエラーが発生するとexceptに飛びそのまま処理を実行します。
本来は、どのようなエラーなのかを出力したりエラーが発生した時に解決専門の関数に飛ばしたりするために使うものですが、今回は大規模なシステムでもないしバグも大した被害が出るものではないので許容しました。
このような使い方は悪手です。
ここからは形態素解析を行います。
などと難しい言葉で言いましたが、やることは文章を品詞で区切るだけの小学生でもできる処理です。
ライブラリを使えば簡単です。
t = Tokenizer()
Tokenizerを呼び出しています。以下tはTokenizerクラスの関数を使うことができます。
ちなみにこのように同じ名前の変数を多用するとヒューマンエラーの元なのでやらない方が良い。
それなりの規模の開発でチームメンバーがこのようなコードを渡してきたらブチギレる自信がある。
for tw in t_lists:
tts = [token.surface for token in t.tokenize(tw) if token.part_of_speech.startswith('名詞')]
for ts in tts:
text += " " + ts
文章に形態素解析をかけてt.tokenize(tw)その結果名詞と分類されたものについてif token.part_of_speech.startswith('名詞')その単語を配列に保存token.surfaceというように処理されています。
create_image.py
f = open(config.stop_words_path, "r")
word = f.read()
f.close()
stop_words = word.split()
文字列から配列への変換をやっているコードです。
stop_words.txtは半角空白で区切られた単語が並んでいるので、半角から次の半角
までを1つの要素として保存すればstop_wordsという配列ができます。
wc = WordCloud(font_path=config.font_pass,
width=1200,
height=675,
background_color="white",
stopwords=set(stop_words),
max_words=500).generate(tweet)
wc.to_file(config.result_path)
WordCloudを作成するコードです。
WordCloud()が画像の生成。to_fileが画像の保存をしています。
font_pathは設定しないと文字化けするので必ず設定しましょう。
width, heightは生成する画像のサイズ。
background_colorは生成する画像の背景色。
stopwordsは生成する画像に表示しない単語。
max_wordsは生成する画像に表示する単語の数。
を設定しています。
api.API.update_status_with_media(status="ツイート内容を入力",
filename=config.result_path)
画像をツイートするコードです。statusにはツイート内容を記述するため、""""""と囲み複数行に渡ってかけるようにした方が良いと思います。
filenameは投稿したい画像を設定します。
実行
プログラムを定期的に実行し続ける必要がある。→いちいちプログラムを実行するのはめんどくさいので自動で実行してもらいたい。
サーバーを借りるという手もあるが、このプログラムは一銭ももらえないので自前で揃えたいところ(学生なので金がない)
→cronを使う
cronの設定
cronとは？？？となってる人は【cron】Macでcronを設定してみたなどを見るといいかも。cronと検索すれば色々出てきます。
今回の例だと
-
tweet.pyを1分ごとに実行 -
create_image.pyを0:00に実行 -
list.pyを1:00に実行
を実現したいので...
まずcrontabを開くために以下のコマンドを叩きます。
$ crontab -e
viでcrontabが開くのでiを入力して画面下にINSERTと表示されたことを確認してから
*/1 * * * * python3のパス tweet.pyのパス
0 0 * * * python3のパス creat_image.pyのパス
0 1 * * * python3のパス list.pyのパス
と入力。終わったらescから:wqと入力してcrontab: installing new crontabとでたら成功です。
python3のパスは、ファイルの詳細からコピーする(僕の場合はうまくいきませんでしたが)か
where python
というコマンドで帰ってきたものをコピーするようにしてください。
絶対パスにすることが大切です
僕の環境ではこれで動作しました。tweet.txtを観察すれば更新されているかがわかると思うので1分待機して確認してください。
動作が確認できなかった場合は
$ cron -l
としてcronの内容を確認し、python3のパス tweet.pyのパスの部分だけをコピペしてターミナルに貼り付けて実行してください。
python3のパス tweet.pyのパス
これでエラーが出た場合はそちらを修正してください。
エラーが出ないのにダメだった場合は他の記事を調べてみるかこの記事のコメントとして伝えてください。
どうしてもダメならlaunchdを使用する方法もあります。
launchdの活用
/Library/LaunchDaemonsにhogehoge.plistを追加し、
$ sudo chmod 644 /Library/LaunchDaemons/jp.co.sciencepark.sampleAntiVirus.plist
$ sudo chown root:wheel /Library/LaunchDaemonsjp.co.sciencepark.sampleAntiVirus.plist
と権限を変更
<?xml version="1.0" encoding="UTF-8"?>
<!DOCTYPE plist PUBLIC "-//Apple//DTD PLIST 1.0//EN" "http://www.apple.com/DTDs/PropertyList-1.0.dtd">
<plist version="1.0">
<dict>
<key>hogehoge</key>
<string>hoge</string>
<key>ProgramArguments</key>
<array>
<string>python3の絶対パス</string>
<string>-Ku</string>
<string>実行ファイルの絶対パス</string>
</array><key>StartCalendarInterval</key>
<dict>
<key>Hour</key>
<integer>0</integer>
<key>Minute</key>
<integer>0</integer>
</dict>
</dict>
</plist>
というファイルを作成。これを実行したいファイル分作れば良い。
変更するべき場所は
<key>hogehoge</key>
<string>hoge</string>
<array>
<string>python3の絶対パス</string>
<string>-Ku</string>
<string>実行ファイルの絶対パス</string>
<dict>
<key>Hour</key>
<integer>0</integer>
<key>Minute</key>
<integer>0</integer>
</dict>
の3つ。
終わったら
$ launchctl load ~/Library/LaunchAgents/test.plist
とすると実行されるようになるはず。
終了させたい時は
$ launchctl unload ~/Library/LaunchAgents/test.plist
とする。
正直この程度の処理でlaunchを使ってもあまり旨味はない(と思う)。
参考