Skip to content

Instantly share code, notes, and snippets.

@35enidoi
Last active October 14, 2024 11:29
Show Gist options
  • Save 35enidoi/e3e15cd535b2a002a59dedee2a460713 to your computer and use it in GitHub Desktop.
Save 35enidoi/e3e15cd535b2a002a59dedee2a460713 to your computer and use it in GitHub Desktop.

なんかいつの間にか作ってたけど運用する気力がないのでgistに置きます
your_misskey_instance_nameの部分をインスタンス名に、bot_tokenのところにトークンを入れれば動きます。
依存関係はbrominecoremisskey-pyです。
絵文字更新とかも検出しますが、コメントアウトして機能を制限しています
もし使いたかったらコメントアウトを戻してください

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