adk23333/gitea_push2qq
Template
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

Compare commits

base: adk23333:5a331e773d88fd79cdd7048288a139deb0c9b64a
Branches Tags
adk23333:main
..
compare: adk23333:a3688bad160fe5fc814ea4dc145975be36c85fac
Branches Tags
adk23333:main

No commits in common. "5a331e773d88fd79cdd7048288a139deb0c9b64a" and "a3688bad160fe5fc814ea4dc145975be36c85fac" have entirely different histories.
5a331e773d ... a3688bad16

2 changed files with 14 additions and 75 deletions
Show Stats Expand all files Collapse all files

59
cmds.py
View File

@ -1,9 +1,6 @@
import secrets import secrets
from casbin import AsyncEnforcer from models import User
from sanic import SanicException
from models import User, GiteaUser
from sio_model import SioDecorator, Message, SioRequest from sio_model import SioDecorator, Message, SioRequest
@ -21,8 +18,6 @@ def cmds(app, data):
parser = sqt.parser parser = sqt.parser
parser.add_argument('-vd', '--validate', help='绑定QQ与gitea', parser.add_argument('-vd', '--validate', help='绑定QQ与gitea',
action="store_true") action="store_true")
parser.add_argument('-ust', '--ustatus', help='查询Gitea绑定状态',
action="store_true")
args = parser.parse_args(sqt.args) args = parser.parse_args(sqt.args)
if args.validate: if args.validate:
state = secrets.token_urlsafe(16) state = secrets.token_urlsafe(16)
@ -40,59 +35,7 @@ def cmds(app, data):
f'response_type=code&state={state}') f'response_type=code&state={state}')
msg = Message(content=f'click: \n{url}', room_id=sqt.room_id) msg = Message(content=f'click: \n{url}', room_id=sqt.room_id)
await sqt.app.ctx.sio.emit('sendMessage', msg.to_json()) 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: if len(sqt.args) == 0:
parser.parse_args(["-h"]) 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())
return sio_decorator return sio_decorator

30
server.py
View File

@ -69,25 +69,25 @@ async def receive(rqt: Request):
match (rqt.headers['X-Gitea-Event']): match (rqt.headers['X-Gitea-Event']):
case GiteaEvent.issues.value: case GiteaEvent.issues.value:
data = WebHookIssue(**rqt.json) data = WebHookIssue(**rqt.json)
rsp_sender = f"{data.sender.username}"
rsp_title = f"[{data.repository.full_name}][{data.action}] {data.issue.title} (Issue #{data.issue.id})" rsp_title = f"[{data.repository.full_name}][{data.action}] {data.issue.title} (Issue #{data.issue.id})"
rsp_sender = f"By {data.sender.username}"
rsp_ctx = f"{data.issue.body}" rsp_ctx = f"{data.issue.body}"
rsp_link = f"View it: {data.issue.html_url}" rsp_link = f"View it: {data.issue.html_url}"
cancel_subscribe = f"unsubscribe: {data.repository.html_url}" cancel_subscribe = f"unsubscribe: {data.repository.html_url}"
rsp = f"{rsp_sender}\n{rsp_title}\n------\n{rsp_ctx}\n------\n{rsp_link}\n{cancel_subscribe}" rsp = f"{rsp_title}\n\n{rsp_sender}\n------\n{rsp_ctx}\n------\n{rsp_link}\n{cancel_subscribe}"
case GiteaEvent.issue_comment.value: case GiteaEvent.issue_comment.value:
data = WebHookIssueComment(**rqt.json) data = WebHookIssueComment(**rqt.json)
rsp_sender = f"{data.sender.username}"
rsp_title = f"Re:[{data.repository.full_name}][{data.action}] {data.issue.title} (Issue #{data.issue.id})" rsp_title = f"Re:[{data.repository.full_name}][{data.action}] {data.issue.title} (Issue #{data.issue.id})"
rsp_sender = f"By {data.sender.username}"
rsp_ctx = f"{data.comment.body}" rsp_ctx = f"{data.comment.body}"
rsp_link = f"View it: {data.comment.html_url}" rsp_link = f"View it: {data.comment.html_url}"
cancel_subscribe = f"unsubscribe: {data.repository.html_url}" cancel_subscribe = f"unsubscribe: {data.repository.html_url}"
rsp = f"{rsp_sender}\n{rsp_title}\n------\n{rsp_ctx}\n------\n{rsp_link}\n{cancel_subscribe}" rsp = f"{rsp_title}\n\n{rsp_sender}\n------\n{rsp_ctx}\n------\n{rsp_link}\n{cancel_subscribe}"
case _: case _:
rsp = "Unknown webhook type! Please contact the administrator." rsp = "Unknown webhook type! Please contact the administrator."
@ -143,17 +143,17 @@ async def redirect(rqt: Request):
q_user = await User.filter(state=rqt_state).get_or_none() q_user = await User.filter(state=rqt_state).get_or_none()
if q_user is None: if q_user is None:
logger.warn(f'该请求不是从QQ中请求的链接') logger.warn(f'该请求不是从QQ中请求的链接')
return text('请先从QQ处申请绑定') return text('请从QQ处申请绑定')
msg = Message(content='已在绑定中，可能因为意外而失败\n查询结果使用\n/gitea -ust', room_id=q_user.id) msg = Message(content='已在绑定中，可能因为意外而失败', room_id=q_user.id)
await app.ctx.sio.emit('sendMessage', msg.to_json()) await app.ctx.sio.emit('sendMessage', msg.to_json())
rqt.app.add_task(get_gitea_token(app, {'client_id': app.ctx.sio_config.client_id, await app.add_task(get_gitea_token(app, {'client_id': app.ctx.sio_config.client_id,
'client_secret': app.ctx.sio_config.client_secret, 'client_secret': app.ctx.sio_config.client_secret,
'code': code, 'code': code,
'grant_type': 'authorization_code', 'grant_type': 'authorization_code',
'redirect_uri': f'{app.ctx.sio_config.localhost}/redirect'}, 'redirect_uri': f'{app.ctx.sio_config.localhost}/redirect'},
q_user), name=str(q_user.id)) q_user), name=str(q_user.id))
return text('success') return text('success')
@ -240,11 +240,7 @@ def start_sio_listener():
@app.ctx.sio.on('addMessage') @app.ctx.sio.on('addMessage')
async def add_message(data: Dict[str, Any]): async def add_message(data: Dict[str, Any]):
sio_decorator = cmds.cmds(app, data) sio_decorator = cmds.cmds(app, data)
try: await sio_decorator.route2cmd_and_run()
await sio_decorator.route2cmd_and_run()
except IndexError:
# 处理非文本消息
pass
if __name__ == "__main__": if __name__ == "__main__":