100 lines
4.3 KiB
Python
100 lines
4.3 KiB
Python
import secrets
|
||
|
||
from casbin import AsyncEnforcer
|
||
from sanic import SanicException
|
||
|
||
from models import User, GiteaUser
|
||
from sio_model import SioDecorator, Message, SioRequest
|
||
|
||
|
||
def cmds(app, data):
|
||
# 此处会检查注册的指令是否重复
|
||
sio_decorator = SioDecorator(app, data)
|
||
|
||
@sio_decorator.cmd('ping')
|
||
async def ping(sqt: SioRequest):
|
||
msg = Message(content='pong', room_id=sqt.room_id)
|
||
await sqt.app.ctx.sio.emit('sendMessage', msg.to_json())
|
||
|
||
@sio_decorator.cmd('gitea')
|
||
async def gitea(sqt: SioRequest):
|
||
parser = sqt.parser
|
||
parser.add_argument('-vd', '--validate', help='绑定QQ与gitea',
|
||
action="store_true")
|
||
parser.add_argument('-ust', '--ustatus', help='查询Gitea绑定状态',
|
||
action="store_true")
|
||
args = parser.parse_args(sqt.args)
|
||
if args.validate:
|
||
state = secrets.token_urlsafe(16)
|
||
user = await User.filter(id=sqt.sender_id).get_or_none()
|
||
if user:
|
||
user.name = sqt.sender_name
|
||
user.state = state
|
||
await user.save()
|
||
else:
|
||
user = User(id=sqt.sender_id, name=sqt.sender_name, state=state)
|
||
await user.save()
|
||
url = (f'{app.ctx.sio_config.gitea_host}/login/oauth/authorize?'
|
||
f'client_id={app.ctx.sio_config.client_id}&'
|
||
f'redirect_uri={app.ctx.sio_config.localhost}/redirect&'
|
||
f'response_type=code&state={state}')
|
||
msg = Message(content=f'click: \n{url}', room_id=sqt.room_id)
|
||
await sqt.app.ctx.sio.emit('sendMessage', msg.to_json())
|
||
|
||
if args.ustatus:
|
||
g_user = await GiteaUser.filter(qid_id=sqt.sender_id).get_or_none()
|
||
if g_user:
|
||
msg = Message(content=f'您的Gitea账号是：{g_user.name}', room_id=sqt.room_id)
|
||
else:
|
||
try:
|
||
task = await sqt.app.get_task(str(sqt.sender_id))
|
||
result = task.result()
|
||
msg = Message(content=f'绑定状态：{result}', room_id=sqt.room_id)
|
||
except SanicException:
|
||
msg = Message(content=f'你还没有绑定呢', room_id=sqt.room_id)
|
||
await sqt.app.ctx.sio.emit('sendMessage', msg.to_json())
|
||
|
||
if len(sqt.args) == 0:
|
||
parser.parse_args(["-h"])
|
||
|
||
@sio_decorator.cmd('authadd')
|
||
async def auth_add(sqt: SioRequest):
|
||
parser = sqt.parser
|
||
parser.add_argument('-ag', '--addgroup', help='添加用户组',
|
||
action="store_true")
|
||
parser.add_argument('-g', '--group', help='组名')
|
||
parser.add_argument('-m', '--command', help='命令名')
|
||
parser.add_argument('-au', '--adduser', help='添加用户到组',
|
||
action="store_true")
|
||
parser.add_argument('-u', '--user', help='用户QQ号')
|
||
|
||
args = parser.parse_args(sqt.args)
|
||
e: AsyncEnforcer = sqt.app.ctx.e
|
||
msg = ''
|
||
if args.addgroup:
|
||
if args.group and args.command:
|
||
if await e.add_policy(args.group, args.command):
|
||
msg = Message(content=f'添加成功：p, {args.group}, {args.command}', room_id=sqt.room_id)
|
||
else:
|
||
msg = Message(content='添加失败，用户组已存在或其它错误', room_id=sqt.room_id)
|
||
else:
|
||
msg = Message(content='缺失参数，请使用-h查看帮助', room_id=sqt.room_id)
|
||
if args.adduser:
|
||
if args.user and args.group:
|
||
if await e.add_role_for_user(args.user, args.group):
|
||
msg = Message(content=f'添加成功：g, {args.user}, {args.group}', room_id=sqt.room_id)
|
||
else:
|
||
msg = Message(content='添加失败，用户已在组内或其它错误', room_id=sqt.room_id)
|
||
else:
|
||
msg = Message(content='缺失参数，请使用-h查看帮助', room_id=sqt.room_id)
|
||
|
||
if len(sqt.args) == 0:
|
||
parser.parse_args(["-h"])
|
||
else:
|
||
if msg == '':
|
||
msg = Message(content='参数错误，请使用-h查看帮助', room_id=sqt.room_id)
|
||
await sqt.app.ctx.sio.emit('sendMessage', msg.to_json())
|
||
await e.save_policy()
|
||
|
||
return sio_decorator
|