From f591d546a5700d1f71967271c334755a239d63fa Mon Sep 17 00:00:00 2001 From: grassblock Date: Mon, 4 Aug 2025 14:35:19 +0800 Subject: [PATCH] feat: misskey support for posting to fediverse --- core/post_to_fedi.py | 176 ++++++++++++++++++++++++++++++++++--------- 1 file changed, 140 insertions(+), 36 deletions(-) diff --git a/core/post_to_fedi.py b/core/post_to_fedi.py index 594fe6b..427456d 100644 --- a/core/post_to_fedi.py +++ b/core/post_to_fedi.py @@ -1,9 +1,14 @@ +import json import logging import os +import uuid + +import aiohttp from aiogram.types import Message, CallbackQuery from aiogram.fsm.context import FSMContext from aiogram.fsm.state import State, StatesGroup from aiogram import Router +from requests import session from config import config from mastodon import Mastodon @@ -39,6 +44,22 @@ def check_user_cred_exists(instance: str, userid: int) -> bool: except FileNotFoundError: return False +async def instance_is_misskey(instance: str) -> bool: + """ + 检查实例是否是 Misskey 实例 + """ + try: + async with aiohttp.ClientSession() as client: + async with client.post(f"https://{instance}/api/v1/instance", headers={"Content-Type": "application/json"}, + allow_redirects=False) as r: + if r.status != 200: + return True + else: + return False # 如果没有异常，则不是 Misskey 实例 + except Exception as e: + logging.debug(f"检查实例 {instance} 是否为 Misskey 时发生错误: {e}") + return True # 如果发生异常，则认为是 Misskey 实例 + async def handle_token(instance,mastodon, message: Message): mastodon.log_in( code=message.text, @@ -59,25 +80,40 @@ async def handle_auth(message: Message, state: FSMContext): if instance == '': await message.reply('请输入实例域名，例如：`example.com`') return + auth_url = '' + session_id = uuid.uuid4() if not check_client_cred_exists(instance): - try: - if not check_secrets_folder_exists(): - os.mkdir('secrets') - Mastodon.create_app( - 'realbot', - api_base_url='https://{}'.format(instance), - to_file='secrets/realbot_{}_clientcred.secret'.format(instance), - scopes=['read:accounts', 'read:statuses','write:media','write:statuses'] - ) - except Exception as e: - logging.warning(e) - await message.reply(f'创建应用失败：{str(e)}\n请确保实例域名正确并且实例支持 Mastodon API。') - return - mastodon = Mastodon(client_id=f'secrets/realbot_{instance}_clientcred.secret') - auth_url = mastodon.auth_request_url() - await message.reply('请在浏览器中打开链接进行身份验证：\n{}\n验证完成后，请用得到的 token 回复这条消息'.format(auth_url)) + if not await instance_is_misskey(instance): + # 如果是 Misskey 实例，使用不同的创建应用方式 + try: + if not check_secrets_folder_exists(): + os.mkdir('secrets') + Mastodon.create_app( + 'realbot', + api_base_url='https://{}'.format(instance), + to_file='secrets/realbot_{}_clientcred.secret'.format(instance), + scopes=['read:accounts', 'read:statuses','write:media','write:statuses'] + ) + except Exception as e: + logging.warning(e) + await message.reply(f'创建应用失败：{str(e)}\n请确保实例域名正确并且实例支持 Mastodon API。') + return + mastodon = Mastodon(client_id=f'secrets/realbot_{instance}_clientcred.secret') + auth_url = mastodon.auth_request_url() + else: + # 如果是 Misskey 实例，使用不同的创建应用方式 + try: + if not check_secrets_folder_exists(): + os.mkdir('secrets') + auth_url = f'https://{instance}/miauth/{session_id}?name=realbot&permission=read:account,write:notes,write:drive' + except Exception as e: + logging.warning(e) + await message.reply(f'创建应用失败：{str(e)}\n请确保实例域名正确并且实例支持 Misskey API。') + return + + await message.reply('请在浏览器中打开链接进行身份验证：\n{}\n验证完成后，请用得到的 token 回复这条消息。\n注意：对于使用 Akkoma 的用户，可能需要验证两次。对于使用 misskey 的用户，请任意输入文本。'.format(auth_url)) # 在发送消息后设置状态 - await state.update_data(instance=instance) + await state.update_data(instance=instance,session=session_id) await state.set_state(AuthStates.waiting_for_token) # 创建处理回复的 handler @@ -85,10 +121,27 @@ async def handle_auth(message: Message, state: FSMContext): async def handle_token_reply(message: Message, state: FSMContext): data = await state.get_data() instance = data.get('instance') - mastodon = Mastodon(client_id=f'secrets/realbot_{instance}_clientcred.secret') - status = await message.reply('正在处理身份验证，请稍候...') - await handle_token(instance,mastodon,message) - await status.edit_text('身份验证成功！\n现在你可以使用 /post 命令将消息发布到联邦网络。') + session_id = data.get('session') + if not await instance_is_misskey(instance): + mastodon = Mastodon(client_id=f'secrets/realbot_{instance}_clientcred.secret') + status = await message.reply('正在处理身份验证，请稍候...') + await handle_token(instance,mastodon,message) + await status.edit_text('身份验证成功！\n现在你可以使用 /post 命令将消息发布到联邦网络。') + else: + status = await message.reply('正在处理身份验证，请稍候...') + misskey_check_url = f'https://{instance}/api/miauth/{session_id}/check' + logging.debug(misskey_check_url) + async with aiohttp.ClientSession() as client: + async with client.post(misskey_check_url, data="", headers={"Accept":"*/*"},allow_redirects=False) as r: + if r.status == 200: + data = await r.json() + if data['token']: + # 保存用户凭据 + with open(f'secrets/realbot_{instance}_{message.from_user.id}_usercred.secret', 'w') as f: + f.write(data['token']) + await status.edit_text('身份验证成功！\n现在你可以使用 /post 命令将消息发布到联邦网络。') + else: + await status.edit_text('身份验证失败，请确保实例域名正确并且实例支持 Misskey API。\n以下的信息可能有助于诊断问题：\n{}'.format(await r.text())) # 清除状态 await state.clear() @@ -179,6 +232,7 @@ async def handle_post_to_fedi(message: Message): # 发布消息到联邦网络 try: status_message = await message.reply('尝试发布消息到联邦网络...') + is_misskey = await instance_is_misskey(instance) # 处理图片附件 media_ids = [] if message.reply_to_message.photo: @@ -187,23 +241,73 @@ async def handle_post_to_fedi(message: Message): photo = message.reply_to_message.photo[-1] file_info = await message.bot.get_file(photo.file_id) file_data = await message.bot.download_file(file_info.file_path) + if not is_misskey: + # 上传图片到Mastodon + media = mastodon.media_post(file_data, mime_type='image/png') + media_ids.append(media['id']) + else: + misskey_upload_drive_url = f"https://{instance}/api/drive/files/create" + token = '' + with open(f'secrets/realbot_{instance}_{message.from_user.id}_usercred.secret', 'r') as f: + token = f.read() - # 上传图片到Mastodon - media = mastodon.media_post(file_data, mime_type='image/png') - media_ids.append(media['id']) - text = message.reply_to_message.text - if media_ids: - text = message.reply_to_message.caption - # 发布消息 - status = mastodon.status_post( - text, - media_ids=media_ids if media_ids else None, - visibility = visibility - ) - status_url = status['url'] - await status_message.edit_text(f'消息已成功发布到联邦网络！\n{status_url}') + file_info = await message.bot.get_file(photo.file_id) + file_data = await message.bot.download_file(file_info.file_path) + + data = aiohttp.FormData() + data.add_field('i',token) + data.add_field('file', file_data, filename=f"{photo.file_id}.png", content_type='image/png') + #if photo.has_media_spoiler: + # data.add_field('isSensitive', True) + data.add_field('name', f"tg_{photo.file_id}.png") + async with aiohttp.ClientSession() as client: + async with client.post(misskey_upload_drive_url, allow_redirects=False, data=data) as r: + if r.status == 200: + media_info = await r.json() + media_ids.append(media_info['id']) + else: + await status_message.reply(f'上传图片失败，但是仍然可以发布这条消息的文本部分。\n错误信息: {r.status} {await r.text()}') + # 如果有图片附件而且图片有 caption，优先使用图片的 caption 作为文本 + text = message.reply_to_message.caption or message.reply_to_message.text + if not is_misskey: + status = mastodon.status_post( + text, + media_ids=media_ids if media_ids else None, + visibility = visibility + ) + status_url = status['url'] + await status_message.edit_text(f'消息已成功发布到联邦网络！\n{status_url}') + else: + # 对于 Misskey 实例，使用不同的发布方式 + misskey_create_status_url = f'https://{instance}/api/notes/create' + token = '' + with open(f'secrets/realbot_{instance}_{message.from_user.id}_usercred.secret', 'r') as f: + token = f.read() + data = {} + if token and media_ids: + data = json.dumps({ + "visibility": visibility if visibility else "public", + "text": text, + "fileIds": media_ids + }) + elif token: + data = json.dumps({ + "visibility": visibility if visibility else "public", + "text": text + }) + created_note = {} + async with aiohttp.ClientSession() as client: + async with client.post(misskey_create_status_url, allow_redirects=False, data=data,headers={"Authorization": f"Bearer {token}","Content-Type": "application/json"}) as r: + if r.status == 200: + created_note = await r.json() + else: + await status_message.edit_text(f'发布失败: {r.status} - {await r.text()}') + return + status_url = f'https://{instance}/notes/{created_note["createdNote"]["id"]}' + await status_message.edit_text(f'消息已成功发布到联邦网络！\n{status_url}') except Exception as e: - await message.reply(f'发布失败: {str(e)}') + logging.warning('Error posting to fedi:', exc_info=e) + await status_message.edit_text(f'发布失败: {str(e)}') @router.callback_query(lambda c: c.data.startswith('post:'))