なんかいつの間にか作ってたけど運用する気力がないのでgistに置きます
your_misskey_instance_nameの部分をインスタンス名に、bot_tokenのところにトークンを入れれば動きます。
依存関係はbrominecore
とmisskey-py
です。
絵文字更新とかも検出しますが、コメントアウトして機能を制限しています
もし使いたかったらコメントアウトを戻してください
Last active
October 14, 2024 11:29
-
-
Save 35enidoi/e3e15cd535b2a002a59dedee2a460713 to your computer and use it in GitHub Desktop.
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
import asyncio | |
import logging | |
from collections import deque | |
from typing import Union | |
from time import time, sleep | |
from brcore import Bromine | |
from requests.exceptions import Timeout, ConnectionError | |
from misskey import ( | |
Misskey, | |
exceptions as misskey_exceptions, | |
) | |
# インスタンス名 | |
INSTANCE = "your_misskey_instance_name" | |
# トークン 必要な権限: | |
# アカウントの情報を見る, ノートを作成・削除する, 通知を見る | |
TOKEN = "bot_token" | |
brm = Bromine(instance=INSTANCE, token=TOKEN) | |
msk = Misskey(address=INSTANCE, i=TOKEN) | |
# 一分間に何回ノートしたらレートリミットにするか | |
RATELIMIT_QUEUENUM = 5 | |
ratelimit_deque = deque((0 for _ in range(RATELIMIT_QUEUENUM)), maxlen=RATELIMIT_QUEUENUM) | |
def ratelimit_isexceeded() -> bool: | |
"""1分以内に連続して投稿しているか""" | |
return (time() - (sum(ratelimit_deque) / RATELIMIT_QUEUENUM)) < 60 | |
def note(text, reply_id: Union[str] = None) -> bool: | |
"""Misskey.notes_createのラッパー(例外が発生しないようにしただけ)""" | |
while ratelimit_isexceeded(): | |
sleep(1) | |
ratelimit_deque.append(time()) | |
try: | |
msk.notes_create(text, reply_id=reply_id) | |
return True | |
except ( | |
misskey_exceptions.MisskeyAPIException, | |
Timeout, | |
ConnectionError, | |
): | |
return False | |
async def on_main(info: dict) -> None: | |
"""通知が入ってきたとき""" | |
if info["type"] == "mention": | |
# メンションが来た時 | |
note_ = info["body"] | |
if note_["user"]["isBot"]: | |
# ボットのメンションは危ないので却下 | |
return | |
elif "ping" in note_["text"]: | |
print("ping was coming!") | |
await asyncio.to_thread(note, "pong!", reply_id=note["id"]) | |
else: | |
# よくわからん奴の場合 | |
pass | |
async def on_emoji_added(info: dict) -> None: | |
"""絵文字追加時""" | |
emoji = info["emoji"] | |
text = f"`{emoji['name']}`が追加\n:{emoji['name']}:\n\nカテゴリー: {emoji['category']}\nタグ: {', '.join(emoji['aliases'])}" | |
if emoji["isSensitive"]: | |
text += "\nセンシティブな絵文字です!" | |
is_success = await asyncio.to_thread(note, text) | |
if is_success: | |
print("create note success at emoji added") | |
else: | |
print("create note failed at emoji added") | |
async def on_emoji_updated(info: dict) -> None: | |
"""絵文字更新""" | |
text = "絵文字に更新が入りました\n変更が入った絵文字たち↓" | |
for emoji in info["emojis"]: | |
emojiname = emoji["name"] | |
emojialiases = emoji["aliases"] | |
emojisensitive = emoji["isSensitive"] | |
text += "-"*10 + "\n" | |
text += f"`{emojiname}`、:{emojiname}:\nタグ; [{'、 '.join(emojialiases)}]\n" | |
if emojisensitive: | |
text += "センシティブな絵文字です\n" | |
text += "-"*10 + "\n" | |
is_success = await asyncio.to_thread(note, text) | |
if is_success: | |
print("create note success at emoji updated.") | |
else: | |
print("create note failed at emoji updated.") | |
async def on_emoji_deleted(info: dict) -> None: | |
"""絵文字削除時""" | |
text = "絵文字が削除。削除された絵文字達↓\n" | |
for emoji in info["emojis"]: | |
text += f"- `{emoji['name']}`\n" | |
is_success = await asyncio.to_thread(note, text) | |
if is_success: | |
print("create note success at emoji deleted") | |
else: | |
print("create note failed at emoji deleted") | |
if __name__ == "__main__": | |
# 通知 | |
brm.ws_connect("main", on_main) | |
# 絵文字追加 | |
brm._add_ws_type_id("emojiAdded", "ALLMATCH", on_emoji_added) | |
# 絵文字更新 | |
# brm._add_ws_type_id("emojiUpdated", "ALLMATCH", on_emoji_updated) | |
# 絵文字削除 | |
# brm._add_ws_type_id("emojiDeleted", "ALLMATCH", on_emoji_deleted) | |
# ログのレベルを変更 | |
# logging.basicConfig(level=logging.INFO) | |
# brm.loglevel = logging.INFO | |
# 見逃さないためにクールダウンを少なくする | |
# 五回連続で失敗したらこの値によらず30秒待ちます(BromineCoreの仕様) | |
brm.cooltime = 1 | |
print("start...") | |
try: | |
asyncio.run(brm.main()) | |
except KeyboardInterrupt: | |
print("stop...") |
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment