gitea_push2qq/cmds.py

import secrets

from casbin import AsyncEnforcer
from sanic import SanicException

from models import User, GiteaUser
from sio_model import SioDecorator, SendMessage, SioRequest, ReplyMessage

# 此处会检查注册的指令是否重复
sio_decorator = SioDecorator()


@sio_decorator.cmd('ping')
async def ping(sqt: SioRequest):
    msg = SendMessage(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)
    reply = ReplyMessage(id=sqt.msg_id)
    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'{sqt.app.ctx.sio_config.gitea_host}/login/oauth/authorize?'
               f'client_id={sqt.app.ctx.sio_config.client_id}&'
               f'redirect_uri={sqt.app.ctx.sio_config.localhost}/redirect&'
               f'response_type=code&state={state}')
        msg = SendMessage(content=f'click: \n{url}', room_id=sqt.room_id, reply_to=reply)
        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 = SendMessage(content=f'您的Gitea账号是：{g_user.name}', room_id=sqt.room_id, reply_to=reply)
        else:
            try:
                task = await sqt.app.get_task(str(sqt.sender_id))
                result = task.result()
                msg = SendMessage(content=f'绑定状态：{result}', room_id=sqt.room_id, reply_to=reply)
            except SanicException:
                msg = SendMessage(content=f'你还没有绑定呢', room_id=sqt.room_id, reply_to=reply)
        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('-u', '--user', help='用户名')
    parser.add_argument('-au', '--adduser', help='添加用户/群到组',
                        action="store_true")

    args = parser.parse_args(sqt.args)
    reply = ReplyMessage(id=sqt.msg_id)
    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 = SendMessage(content=f'添加成功：p, {args.group}, {args.command}', room_id=sqt.room_id,
                                  reply_to=reply)
            else:
                msg = SendMessage(content='添加失败，用户组已存在或其它错误', room_id=sqt.room_id, reply_to=reply)
        else:
            msg = SendMessage(content='缺失参数，请使用-h查看帮助', room_id=sqt.room_id, reply_to=reply)
    if args.adduser:
        if args.group:
            if await e.add_role_for_user(str(args.user or sqt.room_id), args.group):
                msg = SendMessage(content=f'添加成功：g, {args.user or sqt.room_id}, {args.group}', room_id=sqt.room_id,
                                  reply_to=reply)
            else:
                msg = SendMessage(content='添加失败，用户已在组内或其它错误', room_id=sqt.room_id, reply_to=reply)
        else:
            msg = SendMessage(content='缺失参数，请使用-h查看帮助', room_id=sqt.room_id, reply_to=reply)

    if len(sqt.args) == 0:
        parser.parse_args(["-h"])
    else:
        if msg == '':
            msg = SendMessage(content='参数错误，请使用-h查看帮助', room_id=sqt.room_id, reply_to=reply)
        await sqt.app.ctx.sio.emit('sendMessage', msg.to_json())
    await e.save_policy()


@sio_decorator.cmd('authrm')
async def auth_add(sqt: SioRequest):
    parser = sqt.parser
    parser.add_argument('-rg', '--rmgroup', help='移除用户组的权限',
                        action="store_true")
    parser.add_argument('-g', '--group', help='组名')
    parser.add_argument('-u', '--user', help='用户名')
    parser.add_argument('-ru', '--rmuser', help='从组移除用户/群',
                        action="store_true")

    args = parser.parse_args(sqt.args)
    reply = ReplyMessage(id=sqt.msg_id)
    e: AsyncEnforcer = sqt.app.ctx.e
    msg = ''
    if args.rmgroup:
        if args.group and args.command:
            if await e.remove_policy(args.group, args.command):
                msg = SendMessage(content=f'移除成功：p, {args.group}, {args.command}', room_id=sqt.room_id,
                                  reply_to=reply)
            else:
                msg = SendMessage(content='移除失败，用户组已存在或其它错误', room_id=sqt.room_id, reply_to=reply)
        else:
            msg = SendMessage(content='缺失参数，请使用-h查看帮助', room_id=sqt.room_id, reply_to=reply)
    if args.rmuser:
        if args.group:
            if await e.delete_role_for_user(str(args.user or sqt.room_id), args.group):
                msg = SendMessage(content=f'移除成功：g, {args.user or sqt.room_id}, {args.group}', room_id=sqt.room_id,
                                  reply_to=reply)
            else:
                msg = SendMessage(content='移除失败，用户已在组内或其它错误', room_id=sqt.room_id, reply_to=reply)
        else:
            msg = SendMessage(content='缺失参数，请使用-h查看帮助', room_id=sqt.room_id, reply_to=reply)

    if len(sqt.args) == 0:
        parser.parse_args(["-h"])
    else:
        if msg == '':
            msg = SendMessage(content='参数错误，请使用-h查看帮助', room_id=sqt.room_id, reply_to=reply)
        await sqt.app.ctx.sio.emit('sendMessage', msg.to_json())
    await e.save_policy()