mirror of
https://github.com/xuthus83/LittlePaimon.git
synced 2025-04-12 23:29:37 +08:00
106 lines
3.7 KiB
Python
106 lines
3.7 KiB
Python
import asyncio
|
||
import datetime
|
||
import os
|
||
import sys
|
||
from pathlib import Path
|
||
|
||
from fastapi.responses import JSONResponse
|
||
from fastapi import APIRouter
|
||
from nonebot import get_bot
|
||
from nonebot.adapters.onebot.v11 import Bot
|
||
from LittlePaimon import SUPERUSERS
|
||
from LittlePaimon.manager.bot_manager.handler import update
|
||
from LittlePaimon.utils.files import save_json
|
||
from LittlePaimon.utils.tool import cache
|
||
from .utils import authentication
|
||
|
||
route = APIRouter()
|
||
|
||
|
||
@route.get('/get_group_list', response_class=JSONResponse, dependencies=[authentication()])
|
||
async def get_group_list():
|
||
bot: Bot = get_bot()
|
||
return await bot.get_group_list()
|
||
|
||
|
||
@route.get('/get_group_members', response_class=JSONResponse, dependencies=[authentication()])
|
||
async def get_group_members(group_id: int):
|
||
bot: Bot = get_bot()
|
||
return await bot.get_group_member_list(group_id=group_id)
|
||
|
||
|
||
@route.get('/get_groups_and_members', response_class=JSONResponse, dependencies=[authentication()])
|
||
@cache(datetime.timedelta(minutes=10))
|
||
async def get_groups_and_members():
|
||
result = []
|
||
try:
|
||
bot: Bot = get_bot()
|
||
except Exception:
|
||
return {
|
||
'status': -100,
|
||
'msg': '获取群和好友列表失败，请确认已连接GOCQ'
|
||
}
|
||
group_list = await bot.get_group_list()
|
||
friend_list = await bot.get_friend_list()
|
||
for group in group_list:
|
||
group_members = await bot.get_group_member_list(group_id=group['group_id'])
|
||
result.append({
|
||
'left_label': f'{group["group_name"]}({group["group_id"]})',
|
||
'right_label': f'{group["group_name"]}(群{group["group_id"]})',
|
||
'value': f'群{group["group_id"]}',
|
||
'children': [{'left_label': f'{m["card"] or m["nickname"]}({m["user_id"]})',
|
||
'right_label': f'群({group["group_name"]}) - {m["card"] or m["nickname"]}({m["user_id"]})',
|
||
'value': f'群{group["group_id"]}.{m["user_id"]}'} for m in group_members if
|
||
str(m['user_id']) != bot.self_id]
|
||
})
|
||
await asyncio.sleep(0.2)
|
||
result = [
|
||
{
|
||
'label': '群组',
|
||
'selectMode': 'tree',
|
||
'searchable': True,
|
||
'children': result
|
||
},
|
||
{
|
||
'label': '好友',
|
||
'selectMode': 'list',
|
||
'searchable': True,
|
||
'children': [{
|
||
'left_label': f'{f["nickname"]}({f["user_id"]})',
|
||
'right_label': f'{f["nickname"]}({f["user_id"]})',
|
||
'value': f'{f["user_id"]}'
|
||
} for f in friend_list if str(f['user_id']) != bot.self_id]
|
||
}]
|
||
return result
|
||
|
||
|
||
@route.get('/get_friend_list', response_class=JSONResponse, dependencies=[authentication()])
|
||
@cache(datetime.timedelta(minutes=10))
|
||
async def get_friend_list():
|
||
try:
|
||
bot: Bot = get_bot()
|
||
friend_list = await bot.get_friend_list()
|
||
return [{'label': f'{friend["nickname"]}({friend["user_id"]})', 'value': friend['user_id']} for friend in
|
||
friend_list]
|
||
except Exception:
|
||
return {'status': 100, 'msg': '获取好友列表失败'}
|
||
|
||
|
||
@route.post('/bot_update', response_class=JSONResponse, dependencies=[authentication()])
|
||
async def bot_update():
|
||
result = await update()
|
||
return {
|
||
'status': 0,
|
||
'msg': result
|
||
}
|
||
|
||
|
||
@route.post('/bot_restart', response_class=JSONResponse, dependencies=[authentication()])
|
||
async def bot_restart():
|
||
save_json({'session_type': 'private',
|
||
'session_id': SUPERUSERS[0]},
|
||
Path() / 'rebooting.json')
|
||
if sys.argv[0].endswith('nb'):
|
||
sys.argv[0] = 'bot.py'
|
||
os.execv(sys.executable, ['python'] + sys.argv)
|