gb/bot
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

feat: misskey support for posting to fediverse

Browse source
This commit is contained in:
草师傅 2025-08-04 14:35:19 +08:00
parent 2384c6203c
commit f591d546a5
1 changed files with 140 additions and 36 deletions
Download patch file Download diff file Expand all files Collapse all files

120
core/post_to_fedi.py
View file

@ -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,7 +80,11 @@ 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):
if not await instance_is_misskey(instance):
# 如果是 Misskey 实例，使用不同的创建应用方式
try:
if not check_secrets_folder_exists():
os.mkdir('secrets')
@ -75,9 +100,20 @@ async def handle_auth(message: Message, state: FSMContext):
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))
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')
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,14 +241,35 @@ 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'])
text = message.reply_to_message.text
if media_ids:
text = message.reply_to_message.caption
# 发布消息
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,
@ -202,8 +277,37 @@ async def handle_post_to_fedi(message: Message):
)
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:'))