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

适配与去除不必要依赖

Browse Source
This commit is contained in:
San Liy 2024-02-03 23:36:12 +08:00
parent f3eb236b28
commit 5a50b40dbd
4 changed files with 200 additions and 176 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

60
cmds.py
View File

@ -4,18 +4,18 @@ from casbin import AsyncEnforcer
from sanic import SanicException
from models import User, GiteaUser
from sio_model import SioDecorator, Message, SioRequest, ReplyMessage
from sio_model import SioDecorator, SendMessage, SioRequest, ReplyMessage
def cmds(app, data):
# 此处会检查注册的指令是否重复
sio_decorator = SioDecorator(app, data)
sio_decorator = SioDecorator()
@sio_decorator.cmd('ping')
async def ping(sqt: SioRequest):
msg = Message(content='pong', room_id=sqt.room_id)
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
@ -35,29 +35,30 @@ def cmds(app, data):
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&'
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 = Message(content=f'click: \n{url}', room_id=sqt.room_id, reply_to=reply)
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 = Message(content=f'您的Gitea账号是：{g_user.name}', room_id=sqt.room_id, reply_to=reply)
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 = Message(content=f'绑定状态：{result}', room_id=sqt.room_id, reply_to=reply)
msg = SendMessage(content=f'绑定状态：{result}', room_id=sqt.room_id, reply_to=reply)
except SanicException:
msg = Message(content=f'你还没有绑定呢', room_id=sqt.room_id, reply_to=reply)
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
@ -76,28 +77,31 @@ def cmds(app, data):
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, reply_to=reply)
msg = SendMessage(content=f'添加成功：p, {args.group}, {args.command}', room_id=sqt.room_id,
reply_to=reply)
else:
msg = Message(content='添加失败，用户组已存在或其它错误', room_id=sqt.room_id, reply_to=reply)
msg = SendMessage(content='添加失败，用户组已存在或其它错误', room_id=sqt.room_id, reply_to=reply)
else:
msg = Message(content='缺失参数，请使用-h查看帮助', room_id=sqt.room_id, reply_to=reply)
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 = Message(content=f'添加成功：g, {args.user or sqt.room_id}, {args.group}', room_id=sqt.room_id, reply_to=reply)
msg = SendMessage(content=f'添加成功：g, {args.user or sqt.room_id}, {args.group}', room_id=sqt.room_id,
reply_to=reply)
else:
msg = Message(content='添加失败，用户已在组内或其它错误', room_id=sqt.room_id, reply_to=reply)
msg = SendMessage(content='添加失败，用户已在组内或其它错误', room_id=sqt.room_id, reply_to=reply)
else:
msg = Message(content='缺失参数，请使用-h查看帮助', room_id=sqt.room_id, reply_to=reply)
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 = Message(content='参数错误，请使用-h查看帮助', room_id=sqt.room_id, reply_to=reply)
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
@ -115,26 +119,26 @@ def cmds(app, data):
if args.rmgroup:
if args.group and args.command:
if await e.remove_policy(args.group, args.command):
msg = Message(content=f'移除成功：p, {args.group}, {args.command}', room_id=sqt.room_id, reply_to=reply)
msg = SendMessage(content=f'移除成功：p, {args.group}, {args.command}', room_id=sqt.room_id,
reply_to=reply)
else:
msg = Message(content='移除失败，用户组已存在或其它错误', room_id=sqt.room_id, reply_to=reply)
msg = SendMessage(content='移除失败，用户组已存在或其它错误', room_id=sqt.room_id, reply_to=reply)
else:
msg = Message(content='缺失参数，请使用-h查看帮助', room_id=sqt.room_id, reply_to=reply)
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 = Message(content=f'移除成功：g, {args.user or sqt.room_id}, {args.group}', room_id=sqt.room_id, reply_to=reply)
msg = SendMessage(content=f'移除成功：g, {args.user or sqt.room_id}, {args.group}', room_id=sqt.room_id,
reply_to=reply)
else:
msg = Message(content='移除失败，用户已在组内或其它错误', room_id=sqt.room_id, reply_to=reply)
msg = SendMessage(content='移除失败，用户已在组内或其它错误', room_id=sqt.room_id, reply_to=reply)
else:
msg = Message(content='缺失参数，请使用-h查看帮助', room_id=sqt.room_id, reply_to=reply)
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 = Message(content='参数错误，请使用-h查看帮助', room_id=sqt.room_id, reply_to=reply)
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()
return sio_decorator

14
requirements.txt
View File

@ -1,8 +1,10 @@
aiohttp==3.9.1
# casbin==1.33.0
# casbin_tortoise_adapter==1.2.1
pydantic==2.5.2
pynacl==1.5.0
python-socketio==5.10.0
aiohttp==3.9.3
asynccasbin==1.1.8
casbin==1.35.0
casbin_tortoise_adapter==2.0.0
lib_not_dr==0.3.18
pydantic==2.6.0
PyNaCl==1.5.0
python-socketio==5.11.0
sanic==23.6.0
tortoise==0.1.1

7
server.py
View File

@ -14,7 +14,7 @@ from tortoise.contrib.sanic import register_tortoise
import cmds
from gitea_model import WebHookIssueComment, WebHookIssue, GiteaEvent
from models import User, GiteaUser
from sio_model import Ctx, SioConfig, Message
from sio_model import Ctx, SioConfig, SendMessage
from unit import sio_log_format, int2str, cas_log_fmt
app = Sanic('GiteaPush', ctx=Ctx)
@ -77,7 +77,7 @@ async def push_webhook2users(_app: Sanic, full_name: str, msg: str):
_q_user = await g_user.qid.get()
q_user.add(_q_user.id)
for i in q_user:
message = Message(content=msg, room_id=i)
message = SendMessage(content=msg, room_id=i)
await app.ctx.sio.emit('sendMessage', message.to_json())
else:
logger.warn(rps)
@ -261,9 +261,8 @@ def start_sio_listener():
@app.ctx.sio.on('addMessage')
async def add_message(data: Dict[str, Any]):
sio_decorator = cmds.cmds(app, data)
try:
await sio_decorator.route2cmd_and_run()
await cmds.sio_decorator.run(app, data)
except IndexError:
# 处理非文本消息
pass

105
sio_model.py
View File

@ -1,10 +1,10 @@
import shlex
from dataclasses import dataclass
from functools import wraps
from typing import Optional, List, Union, Literal, Dict, Any
from typing import Optional, List, Union, Literal, Dict, Any, Callable
from casbin import Enforcer, AsyncEnforcer
from pydantic import BaseModel, Field, model_validator
from lib_not_dr.types import Options
from pydantic import BaseModel, Field
from sanic import Sanic
from sanic.log import logger
from socketio import AsyncClient
@ -13,12 +13,12 @@ from cmdparser import ArgumentParser, PrintMessage
from unit import sio_log_format
class AtElement(BaseModel):
class AtElement(Options):
text: str
id: Union[int, Literal['all']] = 'all'
class ReplyMessage(BaseModel):
class ReplyMessage(Options):
id: str
username: str = ''
content: str = ''
@ -33,22 +33,16 @@ class ReplyMessage(BaseModel):
}
class Message(BaseModel):
class SendMessage(Options):
content: str
room_id: Optional[int] = None
room: Optional[int] = None # room id 和 room 二选一 ( 实际上直接填 room id 就行了 )
file: None = None # 上传文件
reply_to: Optional[ReplyMessage] = None # 源码 给了一个 any 回复消息
b64_img: Optional[str] = None # 发送图片
at: Optional[List[AtElement]] = [] # @某人
sticker: Optional[None] = None # 发送表情
message_type: Optional[str] = None # 消息类型
@model_validator(mode='after')
def check_room_id(self):
if self.room_id is None and self.room is None:
raise ValueError('room id 和 room 二选一 ( 实际上直接填 room id 就行了 )')
return self
file: None = None # TODO: 上传文件
reply_to: Optional[ReplyMessage] = None # 源码 给了一个 any TODO: 回复消息
b64_img: Optional[str] = None # TODO: 发送图片
at: Optional[List[AtElement]] = [] # TODO: @某人
sticker: Optional[None] = None # TODO: 发送表情
message_type: Optional[str] = None # TODO: 消息类型
def to_json(self) -> dict:
return {
@ -63,6 +57,31 @@ class Message(BaseModel):
'messageType': self.message_type
}
def to_content(self, content: str) -> "SendMessage":
self.content = content
return self
class NewMessage(Options):
sender_id: int
sender_name: str
room_id: int
content: str
msg_id: str
data: dict
def init(self, **kwargs) -> None:
data = kwargs.pop('data')
self.sender_name = data["message"]["username"]
self.sender_id = data["message"]["senderId"]
self.content = data["message"]["content"]
self.room_id = data["roomId"]
self.msg_id = data["message"]["_id"]
def is_self(self, self_id: int) -> bool:
return self.sender_id == self_id
class SioConfig(BaseModel):
host: str
@ -81,36 +100,36 @@ class CmdExists(Exception):
class SioDecorator:
def __init__(self, app: Sanic, data: Dict[str, Any]):
def __init__(self):
self.data = None
self.app = None
self._cmd = None
self._content = None
self.cmd_mapper = {}
def cmd(self, cmd_key: str):
def wrapper(func: Callable[[SioRequest], None]):
self._cmds_append(cmd_key, func)
return func
return wrapper
def _cmds_append(self, cmd, func):
if cmd in self.cmd_mapper.keys():
raise CmdExists(
f"Command already registered: /{cmd}"
)
else:
self.cmd_mapper[f'/{cmd}'] = func
async def run(self, app: Sanic, data: Dict[str, Any]):
logger.debug(sio_log_format('add_message:', data))
self._content = data['message']['content']
self._cmd = shlex.split(self._content)
self.app = app
self.data = data
self.cmds = {}
def cmd(self, cmd_key: str):
def decorator(func):
self._cmds_append(cmd_key, func)
@wraps(func)
async def wrapper(*args, **kwargs): # args 无参数名的 kw 有参数名的
...
return wrapper
return decorator
def _cmds_append(self, cmd, func):
if cmd in self.cmds.keys():
raise CmdExists(
f"Command already registered: /{cmd}"
)
else:
self.cmds[f'/{cmd}'] = func
async def route2cmd_and_run(self):
func = self.cmds.get(self._cmd[0])
func = self.cmd_mapper.get(self._cmd[0])
if func:
sender_id = self.data['message']['senderId']
sender_name = self.data['message']['username']
@ -137,7 +156,7 @@ class SioDecorator:
await func(sqt)
except PrintMessage as e:
reply = ReplyMessage(id=message_id)
msg = Message(content=str(e), room_id=room_id, reply_to=reply)
msg = SendMessage(content=str(e), room_id=room_id, reply_to=reply)
await self.app.ctx.sio.emit('sendMessage', msg.to_json())