備忘録: PythonでMisskey bot作成した

難しかった。
asyncioとかwebsocketとか初めて使ったので、使い方とか調べながらやった。

コード

正直サンプルコードを劣化コピーしただけのクソコードだから晒すの恥ずかしいけどどうせこんなクソブログ誰も見てないだろうし、気にしないことにした。

import time
import random
import asyncio
import json
import websockets
from misskey import Misskey

from golf import golf #自作モジュール

TOKEN = "トークン"
msk = Misskey("自分のMisskey鯖", i=TOKEN)
MY_ID = msk.i()["id"]
WS_URL = "wss://自分のMisskey鯖/streaming?i="+TOKEN


async def WatchTimeLine():
    async with websockets.connect(WS_URL) as ws:

# 将来的にHTLを監視して何かする機能用
#
#        await ws.send(json.dumps({
#            "type": "connect",
#            "body": {
#                "channel": "homeTimeline",
#                "id": "HTL"
#            }
#        }))
        
        await ws.send(json.dumps({
            "type": "connect",
            "body": {
                "channel": "main",
                "id": "main"
            }
        }))

        while 1:
            data = json.loads(await ws.recv())
            
            # このprint文のコメントアウトを外すと
　　　　　　 # どんな通信してるのかが見れて面白い
            # print(data)

            if data["type"] == "channel":
                if data["body"]["type"] == "mention":
                    note = data["body"]["body"]
                    await reply(note)

                if data['body']['type'] == 'followed':
                    user = data['body']['body']
                    await followback(user)
            
async def SaySomething():
    src = 'vocabulary/something.txt'
    with open(src, 'r', encoding='utf-8-sig') as f:
        lines = f.readlines()

    while 1:            
        text = random.choice(lines)
        msk.notes_create(text=text, visibility='followers')
        
        siesta = 86400 - 150 + random.randint(0, 300)
        await asyncio.sleep(siesta)

                    
async def reply(note):
    if note.get("mentions"):
        if MY_ID in note["mentions"]:
            if 'ゴルフ' in note["text"]:
                time.sleep(1)
                text = golf.golf(note["user"]["username"])
                msk.notes_create(text=text, reply_id=note["id"])
            if '占って' in note["text"]:
                time.sleep(1)
                text = "そんな機能無いよ"
                msk.notes_create(text=text, reply_id=note["id"])
            if 'フォローして' in note["text"]:
                try:
                    msk.following_create(note["userId"])
                except:
                    pass
                              
async def followback(user):
    try:
        msk.following_create(user["id"])
    except:
        pass

async def main():
    await asyncio.gather(
        WatchTimeLine(),
        SaySomething(),
    )

if __name__ == '__main__':

    asyncio.run(main())

asyncioの使い方、表面的にしか理解してないんだけど、
・asyncを付けた関数は『コルーチン』と呼ばれ、非同期処理出来る
・コルーチンはメイン関数などでラップする必要あり（上の例だとasyncio.gather()の部分）ちなみにラップの意味は分かってない
・ラップしないと逐次処理になる

っとことらしい。

ついでに

HTLを監視させてる時に読み取ったデータ。こういう構造になってるのね。おもろい。

{
	'type': 'channel', 
	'body': {
		'id': 'HTL',
		'type': 'note', 
		'body': {
			'id': 'ノートのID', 
			'createdAt': '2024-01-07T13:08:18.314Z', 
			'userId': '発言者のID(機械が決めるヤツ)', 
			'user': {
				'id': '発言者のID(機械が決めるヤツ)', 
				'name': '垢名', 
				'username': '垢ID(自分で決められるヤツ)', 
				'host': None, 
				'avatarUrl': 'アイコンURL',
				'avatarBlurhash':'わかんない、何これ', 
				'avatarColor': None, 
				'isCat': True, 
				'emojis': [], 
				'onlineStatus': 'online', 
				'driveCapacityOverrideMb': None
				}, 
			'text': 'ほっぉ', 
			'cw': None, 
			'visibility': 'specified', 
			'visibleUserIds': ['DMにした場合、ここに閲覧可能なユーザーのID'], 
			'renoteCount': 0, 
			'repliesCount': 0, 
			'reactions': {}, 
			'emojis': [], 
			'fileIds': [], 
			'files': [], 
			'replyId': None, 
			'renoteId': None, 
			'mentions': ['メンション先ID']
		}
	}
}

参考

MisskeyボットをPythonで作る - Qiita