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 router = Router() class AuthStates(StatesGroup): waiting_for_token = State() def check_secrets_folder_exists() -> bool: """ 检查 secrets 文件夹是否存在 """ return os.path.exists('secrets') and os.path.isdir('secrets') def check_client_cred_exists(instance: str) -> bool: """ 检查实例的凭据文件是否存在 """ try: with open(f'secrets/realbot_{instance}_clientcred.secret', 'r'): return True except FileNotFoundError: return False def check_user_cred_exists(instance: str, userid: int) -> bool: """ 检查用户凭据文件是否存在 """ try: with open(f'secrets/realbot_{instance}_{userid}_usercred.secret', 'r'): return True except FileNotFoundError: return False async def instance_is_misskey(instance: str) -> bool: """ 检查实例是否是 Misskey 实例 """ try: async with aiohttp.ClientSession() as client: async with client.get(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, to_file=f"secrets/realbot_{instance}_{message.from_user.id}_usercred.secret", scopes=['read:accounts', 'read:statuses', 'write:media', 'write:statuses'] ) async def handle_auth(message: Message, state: FSMContext): """ 处理身份验证 """ if not config.is_feature_enabled('fedi', message.chat.id): return if not message.chat.type == 'private': await message.reply('请在私聊中使用此命令') return instance = message.text.replace('/fauth', '').strip() if instance == '': await message.reply('请输入实例域名，例如：`example.com`') return auth_url = '' session_id = uuid.uuid4() if not check_client_cred_exists(instance): 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,session=session_id) await state.set_state(AuthStates.waiting_for_token) # 创建处理回复的 handler @router.message(AuthStates.waiting_for_token) async def handle_token_reply(message: Message, state: FSMContext): data = await state.get_data() instance = data.get('instance') 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() async def handle_post_to_fedi(message: Message): """ 处理发布到联邦网络的消息 """ if not config.is_feature_enabled('fedi', message.chat.id): return if not message.reply_to_message: await message.reply('请回复要发布的消息') return user_id = message.from_user.id # 查找用户绑定的实例 import glob user_cred_pattern = f'secrets/realbot_*_{user_id}_usercred.secret' user_cred_files = glob.glob(user_cred_pattern) if not user_cred_files: await message.reply('请先使用 /fauth 命令进行身份验证') return arguments = message.text.replace('/post', '', 1).strip().split(' ') specified_instance = None visibility = None if len(arguments) >= 1 and arguments[0]: # 检查第一个参数是否是实例名 first_arg = arguments[0] # 检查是否存在对应的凭据文件 matching_files = [f for f in user_cred_files if first_arg in f] if matching_files: specified_instance = first_arg if len(arguments) >= 2: visibility = arguments[1] else: visibility = arguments[0] # 如果指定了实例，使用指定的实例 if specified_instance: cred_file = next(f for f in user_cred_files if specified_instance in f) filename = os.path.basename(cred_file) parts = filename.split('_') instance = '_'.join(parts[1:-2]) else: # 如果有多个实例，让用户选择 if len(user_cred_files) > 1: from aiogram.types import InlineKeyboardMarkup, InlineKeyboardButton # 提取所有实例名 instances = [] for cred_file in user_cred_files: filename = os.path.basename(cred_file) parts = filename.split('_') instance_name = '_'.join(parts[1:-2]) instances.append(instance_name) # 创建选择按钮 keyboard = [] for instance_name in instances: keyboard.append([InlineKeyboardButton( text=f"{instance_name}", callback_data=f"post:{instance_name}:" )]) # 添加全部发送选项 keyboard.append([InlineKeyboardButton( text="发送到所有实例", callback_data="post_instance:all" )]) reply_markup = InlineKeyboardMarkup(inline_keyboard=keyboard) await message.reply("请选择要发送到的实例：\n（实验性，目前还不能使用）", reply_markup=reply_markup) return else: # 只有一个实例，直接使用 cred_file = user_cred_files[0] filename = os.path.basename(cred_file) parts = filename.split('_') instance = '_'.join(parts[1:-2]) mastodon = Mastodon( access_token=f'{cred_file}', api_base_url=f'https://{instance}' ) # 发布消息到联邦网络 try: status_message = await message.reply('尝试发布消息到联邦网络...') is_misskey = await instance_is_misskey(instance) # 处理图片附件 media_ids = [] if message.reply_to_message.photo: await status_message.edit_text('正在处理图片附件...') # 获取最大尺寸的图片 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() 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: 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:')) async def handle_instance_selection(callback: CallbackQuery): """处理实例选择回调""" await callback.answer() data_parts = callback.data.split(':') selected_instance = data_parts[1] # 获取原始回复消息 # 获取原始 /post 命令消息，然后获取它回复的消息 post_command_message = callback.message.reply_to_message original_message = post_command_message.reply_to_message if post_command_message else None if not original_message: await callback.message.edit_text("错误：找不到原始消息") return user_id = callback.from_user.id mastodon = Mastodon( access_token=f'secrets/realbot_{selected_instance}_{user_id}_usercred.secret', api_base_url=f'https://{selected_instance}' ) # 发布消息到联邦网络 try: status_message = await original_message.reply('尝试发布消息到联邦网络...') # 处理图片附件 media_ids = [] cb_arguments = original_message.text.replace('/post', '', 1).strip().split(' ') if not original_message.reply_to_message: await status_message.edit_text("错误：找不到要发布的消息") return if original_message.reply_to_message.photo: await status_message.edit_text('正在处理图片附件...') # 获取最大尺寸的图片 photo = original_message.reply_to_message.photo[-1] file_info = await callback.message.bot.get_file(photo.file_id) file_data = await callback.message.bot.download_file(file_info.file_path) # 上传图片到Mastodon media = mastodon.media_post(file_data, mime_type='image/png') media_ids.append(media['id']) text = original_message.reply_to_message.text if media_ids: text = original_message.reply_to_message.caption # 发布消息 status = mastodon.status_post( text, media_ids=media_ids if media_ids else None, visibility = cb_arguments[0] if len(cb_arguments) == 1 else None # 默认为账户默认可见性 ) status_url = status['url'] await status_message.edit_text(f'消息已成功发布到联邦网络！\n{status_url}') await callback.message.delete() except Exception as e: await status_message.edit_text(f'发布失败: {str(e)}')