###いらっしゃいませ、今日も楽しいMastodonライフをお過ごしでしょうか☕###
本日は私、香風智乃が恥ずかしながらも誕生日を迎えてしまいました。ささやかに祝ってくださると嬉しい限りです。
さて、昨日の深夜(3日の1時45分あたりですね)ではadminの長である神崎おにいさんの古巣、friends.nicoで大量作成垢の集団による連投スパムが起こりました。friends.nicoで24時間挨拶営業をしているももなさんが連投スパムに巻き込まれてしまいましたね。今すぐ住所特定して消費されたID達の供養謝罪文を書かせながらダイナマイトを送りつけて世間に見放されて爆死してほしいものですね。
複数のIPを駆使したとの噂も聞きますが、ある程度MastodonのAPIを熟知してしまうと、あっという間に大量の新規作成Botが出来てしまったり複数のアカウントから同じ文章をスパム投稿できたりしてしまいます。本当にどうしようもないセンスのないあたまわるわるな愉快犯ですね。
このように使い慣れてる言語の力と個人の発想によって、やりたいことはなんでも出来てしまいます。Botの良し悪しも製作者によって決まってしまいます。Botを乱雑に扱ったりするのは心が痛みます。丁重に育て、時間を消費する覚悟でコードを修正する努力が愛にあふれるBotができあがります。
という長い前置きはさておき、2日に引き続き応用を利かせた、StreamingAPIを使用してのBotの作り方をしていきましょう。
#色んな機能を実装するための基本形#
自分のために、もしくはローカルタイムラインにいる人のために、メンションを頂く人のために……形がそれぞれだと思いますが、運用していくと機能をどんどん実装していきたいことでしょう。特に今回もPythonを使用するので、numpyなどで機械学習をさせたりしたいこともあるでしょう。私はしたいです。
ということなので私達のアイドルこおりちゃんの中身(意味深)をベースにコードを組み立てましょう。
# -*- coding: utf-8 -*-
# あらかじめ役に立つモジュールを追加してますが好きなようにカスタマイスしてください。
# from mastodon import * 及び import threading は必須となります。
from mastodon import *
from time import sleep
from time import time
import re, sys, os, csv, json, codecs, io, gc
import threading, random, traceback
# プロンプトで起動したい人のための装置
sys.stdout = io.TextIOWrapper(sys.stdout.buffer,
encoding=sys.stdout.encoding,
errors='backslashreplace',
line_buffering=sys.stdout.line_buffering)
url_ins = 'https://knzk.me'
mastodon = Mastodon(
client_id='client.secret',
access_token='user.secret',
api_base_url=url_ins)
class User(StreamListener): # ホームでフォローした人と通知を監視します
def on_notification(self, notification): # 通知を監視します
"""ここに機能を追加します"""
print(str(notification["type"]))
pass
class Local(StreamListener): # ここではLTLを監視する継承クラスになります。
def on_update(self, status): # StreamingAPIがリアルタイムにトゥート情報を吐き出してくれます。
"""ここに機能を追加します"""
print("update "+str(status["id"]))
pass
def on_delete(self, status_id): # トゥー消し警察の監視場になります。
print("del "+str(status_id))
pass
class Loading():
def go_local(): # listenerオブジェクトには監視させるものを(続く)
try:
listener = Local()
mastodon.stream_local(listener)
except:
print("例外情報\n" + traceback.format_exc())
pass
def go_user(): # (続き)継承で組み込んだものを追加するようにします。
try:
listener = User()
mastodon.stream_user(listener)
except:
print("例外情報\n" + traceback.format_exc())
sleep(180)
pass
if __name__ == '__main__': # ファイルから直接開いたら動くよ!
try:
user = threading.Thread(target=Loading.go_user)
user.start()
local = threading.Thread(target=Loading.go_local)
local.start()
except:
sleep(3)
bot.toot("すみません、ログアウトするかもしれません。")
これで通知の監視、及びローカルタイムライン監視の体制が出来ました。このままですと本当に監視するだけぼーっとしてるBotとなります。
解説をしますと、まずはストリーミング処理をthreading.Thread
でそれぞれのThreadを処理させていきます。
そしてそのdefの中身の基本となるのが
listener = Local() # ローカルタイムライン
mastodon.stream_local(listener)
listener = User() # ホーム及び通知
mastodon.stream_user(listener)
となります。連合ですとmastodon.stream_public(listener)
になって、ハッシュタグのStreamingですとstream_hashtag(tag, listener)
になります。tagは監視するハッシュタグ名が入ると思います。
*こおりちゃんの場合、古いMastodon.py対応のままなのでStreaming関数がmastodon.user_stream``````mastodon.local_stream
と書かれています。もし動かなった場合にmastodon.stream_XXX
からmastodon.XXX_stream
へ書き方に換えてみてください。
listenerに代入させるオブジェクトにはStreamListenerを継承したクラスが鉄則です。中身に関してはMastodon.pyのStreaming.pyを覗いてみればどのような役割をしているかわかりやすいと思います。え、わからないですか?……ご親切に英語でかかれているので頑張って英語を覚えてください。
……もう、しょうがないお兄ちゃんやお姉ちゃんですね。
基本的にお手本のpyに使用されてる関数だけを簡単に紹介しますと、
・on_update(self, status)
タイムラインで読み込むtootを拾います。基本形ですね。
statusという、toot内容やID番号、メディア情報やユーザーの情報までがっしり読み込んでくれます。・on_notification(self, notification)
こちらは通知読み込み専用となります。user_streamだけが使用できる関数です。
ふぁぼベルやブースト、返信などの様々なタイプが分かれます。・on_delete(self, status_id)
こちらは削除通知です。何が消されたかID番号を知らせてくれます。
ローカルで繋いでも他鯖から流れます。なので他鯖が垢消しすると大量のtootが消える通知によって重くなります。
となります。
仕組みがわかりましたら試しに機能実装の実践です。
##機能実装を追加する例##
Botクラスを作成して、何か適当なdef関数を用意してみましょう。
class bot():
def func01(status):
content = status['content']
if re.compile("コーヒー(ひとつ|くださ[あ~ー]*い|お(願|ねが)い"
"|ちょ[うぉお~ー]*だい)").search(content): #前回より正規表現を駆使して強化です
post = 'どうぞ☕'
time.sleep(3)
return mastodon.status_post(post,visibility='public')
pass
次にon_update
の中で関数を通しましょう☕
def on_update(self, status): # StreamingAPIがリアルタイムにトゥート情報を吐き出してくれます。
if if account["acct"] != 'ここにアカウントのID名': # 自分のtootに反応しないようにするための処理です
bot.func01(status)
pass
これでコーヒーが欲しい人のtootが読み込まれると空リプしてくれます。
メンション限定でお返ししたい場合には特殊な形になります。
class User(StreamListener): # ホームでフォローした人と通知を監視するStreamingAPIの継承クラスです。
def on_notification(self, notification): # 通知を監視します。
account = notification["account"]
if notification["type"] == "mention": # 通知がリプだった場合です。
status = notification["status"]
if account["acct"] != "ここにBotのアカウントID名":
post = bot.func(status)
bot.men(post, status)
pass
class bot():
def func01(status):
content = status['content']
if re.compile("コーヒー(ひとつ|くださ[あ~ー]*い|お(願|ねが)い"
"|ちょ[うぉお~ー]*だい)").search(content):
return 'どうぞ☕'
else:
return None
def men(post, status):
account = status["account"]
toot = '@' + account["acct"] + ' ' + post
time.sleep(3)
return mastodon.status_post(toot,
visibility=status["visibility"],
in_reply_to_id=status["id"])
これで相手からのメンションが来た時に、送られた公開範囲にあわせて3秒後でお返ししてくれます。便利ですね☕
リプを頂いた拍子でお気に入りボタンを押してあげたい時は、
class User(StreamListener):
def on_notification(self, notification):
account = notification["account"]
if notification["type"] == "mention":
status = notification["status"]
time.sleep(2)
mastodon.status_favourite(status["id"]) # ←これを追加します
if account["acct"] != "ここにBotのアカウントID名":
post = bot.func(status)
bot.men(post, status)
pass
となります。
フォロー返しがしたい時には、
class User(StreamListener):
def on_notification(self, notification):
account = notification["account"]
if notification["type"] == "follow": # 通知がフォローだった場合はフォロバします。
sleep(2)
mastodon.account_follow(account["id"])
if文でTypeを"follow"に指定してmastodon.account_follow()を使用します。
他にも様々なことが出来たりしますが、丁寧に説明するとすごく長くなってしまうので今回の解説はここまでとなります。
StreamingAPIを使えばAPIを無駄に消費せずにtootを拾うことが出来ます。過去を遡ることは出来ませんので、その場合はAPIでタイムラインを拾ってあげることになるかと思います。
Bot制作に慣れましたら、あとは自由に実装していくだけですのでどんどん自分だけのBotをカスタマイズしていきましょう。例に出した私のコードでは使いにくいと感じましたら改造していっても構いませんし、他の言語で動かしてみたいと思ったときになにかと参考になることでしょう。
knzk.me Advent Calendar 2017の参加をきっかけにQiitaに登録してみましたが、実際にやってみると記事を書くのはすごく楽しいですね。
なにか面白い話題や記事が浮かびましたら張り切って公開記事を作成してみたいと思います。もし今回のBot作成に関しての記事が好評でしたらシリーズ化させていたいと思います。そして微力ながらお役に立てられるのなら光栄です。
ということで、ここまで読んでいただいてありがとうございます。記事の内容におかしなところがあっても目をつむってください。~~その時は改めて記事を立てるかと思います。~~それでは楽しいひととき、Mastodonライフをお楽しみください。またのお越しをお待ちしております☕