基本完成QQ指令部分

.gitignore

@@ -3,3 +3,5 @@
 
 /config.toml
 /temp.json
+
+/db.*

.idea/dataSources.xml

@@ -0,0 +1,17 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<project version="4">
+  <component name="DataSourceManagerImpl" format="xml" multifile-model="true">
+    <data-source source="LOCAL" name="db" uuid="a69a780c-7ddf-4eb5-891a-e4864b75ed31">
+      <driver-ref>sqlite.xerial</driver-ref>
+      <synchronize>true</synchronize>
+      <jdbc-driver>org.sqlite.JDBC</jdbc-driver>
+      <jdbc-url>jdbc:sqlite:E:\pythonProject\gitea_push2qq\db.sqlite3</jdbc-url>
+      <working-dir>$ProjectFileDir$</working-dir>
+      <libraries>
+        <library>
+          <url>file://$APPLICATION_CONFIG_DIR$/jdbc-drivers/Xerial SQLiteJDBC/3.39.2/sqlite-jdbc-3.39.2.jar</url>
+        </library>
+      </libraries>
+    </data-source>
+  </component>
+</project>

casbin_data/model.conf

@@ -0,0 +1,14 @@
+[request_definition]
+r = sub, act
+
+[policy_definition]
+p = sub, act
+
+[role_definition]
+g = _, _
+
+[policy_effect]
+e = some(where (p.eft == allow))
+
+[matchers]
+m = g(r.sub, p.sub) && (r.act == p.act || p.act == "*")

cmds.py

@@ -0,0 +1,41 @@
+import secrets
+
+from models import User
+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")
+        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.validate_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 len(sqt.args) == 0:
+            parser.parse_args(["-h"])
+
+    return sio_decorator

model.py

@@ -1,63 +0,0 @@
-from dataclasses import dataclass
-from typing import Optional, List, Union, Literal
-
-from lib_not_dr.types import Options
-from socketio import AsyncClient
-
-
-class AtElement(Options):
-    text: str
-    id: Union[int, Literal['all']] = 'all'
-
-
-class ReplyMessage(Options):
-    id: str
-    username: str = ''
-    content: str = ''
-    files: list = []
-
-    def to_json(self) -> dict:
-        return {
-            '_id': self.id,
-            'username': self.username,
-            'content': self.content,
-            'files': self.files
-        }
-
-
-class Message(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  # 消息类型
-
-    def to_json(self) -> dict:
-        return {
-            'content': self.content,
-            'roomId': self.room_id,
-            'room': self.room,
-            'file': self.file,
-            'replyMessage': self.reply_to.to_json() if self.reply_to else None,
-            'b64img': self.b64_img,
-            'at': self.at,
-            'sticker': self.sticker,
-            'messageType': self.message_type
-        }
-
-
-@dataclass
-class SioConfig:
-    HOST: str
-    KEY: str
-    SELF_ID: str
-
-
-@dataclass
-class Ctx:
-    sio: AsyncClient
-    sio_config: SioConfig

models.py

@@ -0,0 +1,16 @@
+from tortoise import fields
+from tortoise.models import Model
+
+
+class User(Model):
+    id = fields.BigIntField(pk=True, autoincrement=False)
+    name = fields.CharField(max_length=60)
+    state = fields.CharField(max_length=60)
+
+
+class GiteaUser(Model):
+    id = fields.BigIntField(pk=True, autoincrement=False)
+    qid = fields.ForeignKeyField('models.GiteaUser')
+    name = fields.CharField(max_length=20)
+    ac_tk = fields.CharField(max_length=256)
+    rfs_tk = fields.CharField(max_length=256)

server.py

@@ -1,14 +1,19 @@
 import tomllib
-from typing import Dict
+from typing import Dict, Any, List, Tuple
 
+import casbin
+from casbin_tortoise_adapter import TortoiseAdapter
 from nacl.signing import SigningKey
 from sanic import Sanic, Request
 from sanic.log import logger, Colors
 from sanic.response import text
 from socketio import AsyncClient
+from tortoise.contrib.sanic import register_tortoise
 
+import cmds
 from gitea_model import WebHookIssueComment, WebHookIssue, GiteaEvent
-from model import Ctx, SioConfig, Message
+from sio_model import Ctx, SioConfig, Message
+from unit import sio_log_format, int2str, cas_log_fmt
 
 app = Sanic('GiteaPush', ctx=Ctx)
 
@@ -16,17 +21,45 @@ app = Sanic('GiteaPush', ctx=Ctx)
 def get_config() -> SioConfig:
     with open('config.toml', 'rb') as f:
         config = tomllib.load(f)
-    return SioConfig(config['host'], config['private_key'], config['self_id'])
+    return SioConfig(**config)
+
+
+SIO_CONFIG = get_config()
+
+register_tortoise(
+    app, db_url=SIO_CONFIG.db_url, modules={"models": ["models", "casbin_tortoise_adapter"]}, generate_schemas=True
+)
 
 
 @app.before_server_start
 async def setup_before_start(_app):
-    _app.ctx.sio_config = get_config()
+    _app.ctx.sio_config = SIO_CONFIG
-    _app.ctx.sio = AsyncClient()
 
+    # 使用casbin策略管理
+    adapter = TortoiseAdapter()
+    e = casbin.AsyncEnforcer('./casbin_data/model.conf', adapter)
+    _app.ctx.e = e
+
+    t1 = await _app.ctx.e.add_policy('admin', '*')
+    t2 = await _app.ctx.e.add_policy('default', 'ping')
+    if t1 is True and t2 is True:
+        logger.info(cas_log_fmt('Init casbin rule success!'))
+    admins = int2str(_app.ctx.sio_config.admin)
+    for qid in admins:
+        if await _app.ctx.e.add_role_for_user(qid, 'admin'):
+            logger.debug(cas_log_fmt(f'Added {Colors.PURPLE}{qid}{Colors.YELLOW} to admin group'))
+    users = await _app.ctx.e.get_users_for_role('admin')
+    rm_user = set(users) ^ set(admins)
+    for u in list(rm_user):
+        if await _app.ctx.e.delete_user(u):
+            logger.debug(f'Delete {Colors.PURPLE}{u}{Colors.YELLOW} for group admin')
+    await _app.ctx.e.save_policy()
+
+    # 初始化sio
+    _app.ctx.sio = AsyncClient()
     start_sio_listener()
 
-    await _app.ctx.sio.connect(_app.ctx.sio_config.HOST)
+    await _app.ctx.sio.connect(_app.ctx.sio_config.host)
 
 
 @app.post('/receive')
@@ -62,6 +95,13 @@ async def receive(rqt: Request):
     return text(rsp)
 
 
+@app.get('/redirect')
+async def redirect(rqt: Request):
+    print(rqt.args)
+    print(rqt.ctx.state)
+    return text('success')
+
+
 """
 以下为QQ监听部分
 """
@@ -77,7 +117,7 @@ def start_sio_listener():
         logger.info(
             f"{Colors.BLUE}versions: {Colors.PURPLE}{versions} {Colors.BLUE}with type {type(salt)}|{salt=}{Colors.END}")
         # 准备数据
-        sign = SigningKey(bytes.fromhex(app.ctx.sio_config.KEY))
+        sign = SigningKey(bytes.fromhex(app.ctx.sio_config.key))
         signature = sign.sign(bytes.fromhex(salt))
 
         # 发送数据
@@ -85,6 +125,83 @@ def start_sio_listener():
         await app.ctx.sio.emit('auth', signature.signature)
         logger.info(f"{Colors.BLUE}send auth emit{Colors.END}")
 
+    @app.ctx.sio.on('auth')
+    async def auth(data: Dict[str, Any]):
+        logger.info(f"auth: {data}")
+
+    @app.ctx.sio.on('authFailed')
+    async def auth_failed():
+        logger.warn(f"authFailed")
+        await app.ctx.sio.disconnect()
+
+    @app.ctx.sio.on('authSucceed')
+    def auth_succeed():
+        logger.info(f"authSucceed")
+
+    @app.ctx.sio.on('connect_error')
+    def connect_error(*args, **kwargs):
+        logger.warn(f"连接错误 {args}, {kwargs}")
+
+    @app.ctx.sio.on('updateRoom')
+    def update_room(data: Dict[str, Any]):
+        logger.debug(sio_log_format('update_room:', data))
+
+    @app.ctx.sio.on('deleteMessage')
+    def delete_message(message_id: str):
+        logger.debug(sio_log_format('delete_message:', message_id))
+
+    @app.ctx.sio.on('setMessages')
+    def set_messages(data: Dict[str, Any]):
+        logger.debug(f"{sio_log_format('set_messages:', data)}"
+                     f"{sio_log_format('message_len:', len(data['messages']))}")
+
+    @app.ctx.sio.on('setAllRooms')
+    def set_all_rooms(rooms: List[Dict[str, Any]]):
+        logger.debug(f"{sio_log_format('set_all_rooms:', rooms)}"
+                     f"{sio_log_format('len:', len(rooms))}")
+
+    @app.ctx.sio.on('setAllChatGroups')
+    def set_all_chat_groups(groups: List[Dict[str, Any]]):
+        logger.debug(f"{sio_log_format('set_all_chat_groups:', groups)}"
+                     f"{sio_log_format('len:', len(groups))}")
+
+    @app.ctx.sio.on('notify')
+    def notify(data: List[Tuple[str, Any]]):
+        logger.debug(sio_log_format('notify:', data))
+
+    @app.ctx.sio.on('closeLoading')
+    def close_loading(_):
+        logger.debug(sio_log_format('close_loading', ''))
+
+    @app.ctx.sio.on('onlineData')
+    def online_data(data: Dict[str, Any]):
+        logger.debug(sio_log_format('online_data:', data))
+
+    @app.ctx.sio.on('*')
+    def catch_all(event, data):
+        logger.debug(sio_log_format('catch_all:', f'{event}|{data}'))
+
+    @app.ctx.sio.on('addMessage')
+    async def add_message(data: Dict[str, Any]):
+        sio_decorator = cmds.cmds(app, data)
+        await sio_decorator.route2cmd_and_run()
+
+    # @app.ctx.sio.on('addMessage')
+    # @SioDecorator.cmd('gitea', app)
+    # async def gitea(sqt: SioRequest):
+    #     parser = sqt.parser
+    #     parser.add_argument('-vd', help='绑定QQ与gitea')
+    #     args = parser.parse_args(sqt.args)
+    #     if args.vd:
+    #         state = secrets.token_urlsafe(16)
+    #         user = User(id=sqt.sender_id, name=sqt.sender_name, state=state)
+    #         await user.save()
+    #         url = (f'{app.ctx.sio_config.validate_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}')
+    #         return Message(content=f'click: \n{url}')
+
 
 if __name__ == "__main__":
     app.run(host='0.0.0.0', port=80, dev=True)

sio_model.py

@@ -0,0 +1,156 @@
+import shlex
+from dataclasses import dataclass
+from functools import wraps
+from typing import Optional, List, Union, Literal, Dict, Any
+
+from casbin import Enforcer
+from pydantic import BaseModel, Field, model_validator
+from sanic import Sanic
+from sanic.log import logger
+from socketio import AsyncClient
+
+from cmdparser import ArgumentParser, PrintMessage
+from unit import sio_log_format
+
+
+class AtElement(BaseModel):
+    text: str
+    id: Union[int, Literal['all']] = 'all'
+
+
+class ReplyMessage(BaseModel):
+    id: str
+    username: str = ''
+    content: str = ''
+    files: list = []
+
+    def to_json(self) -> dict:
+        return {
+            '_id': self.id,
+            'username': self.username,
+            'content': self.content,
+            'files': self.files
+        }
+
+
+class Message(BaseModel):
+    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
+
+    def to_json(self) -> dict:
+        return {
+            'content': self.content,
+            'roomId': self.room_id,
+            'room': self.room,
+            'file': self.file,
+            'replyMessage': self.reply_to.to_json() if self.reply_to else None,
+            'b64img': self.b64_img,
+            'at': self.at,
+            'sticker': self.sticker,
+            'messageType': self.message_type
+        }
+
+
+class SioConfig(BaseModel):
+    host: str
+    key: str = Field(alias='private_key')
+    self_id: int
+    admin: List[int]
+    validate_host: str
+    client_id: str
+    client_secret: str
+    localhost: str
+    db_url: str
+
+
+class CmdExists(Exception):
+    ...
+
+
+class SioDecorator:
+    def __init__(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])
+        if func:
+            sender_id = self.data['message']['senderId']
+            sender_name = self.data['message']['username']
+            room_id = self.data['roomId']
+
+            if sender_id != self.app.ctx.sio_config.self_id:
+                parser = ArgumentParser(self._cmd[0])
+
+                sqt = SioRequest(app=self.app,
+                                 parser=parser,
+                                 args=self._cmd[1:],
+                                 key=self._cmd[0],
+                                 sender_id=sender_id,
+                                 sender_name=sender_name,
+                                 content=self._content,
+                                 room_id=room_id,
+                                 data=self.data)
+                try:
+                    await func(sqt)
+                except PrintMessage as e:
+                    msg = Message(content=str(e), room_id=room_id)
+                    await self.app.ctx.sio.emit('sendMessage', msg.to_json())
+
+
+@dataclass
+class Ctx:
+    sio: AsyncClient
+    sio_config: SioConfig
+    e: Enforcer
+    sio_decorator: SioDecorator
+
+
+@dataclass
+class SioRequest:
+    app: Sanic = None
+    parser: Optional[ArgumentParser] = None
+    args: Optional[list] = None
+    key: Optional[str] = None
+    data: Optional[Dict[str, Any]] = None
+    sender_id: Optional[int] = None
+    sender_name: str = ''
+    content: str = ''
+    room_id: Optional[int] = None

unit.py

@@ -0,0 +1,18 @@
+from typing import Any
+
+from sanic.log import Colors
+
+
+def sio_log_format(text: str, data: Any = ''):
+    return f"{Colors.GREEN}{text} {Colors.PURPLE}{data}{Colors.END}"
+
+
+def cas_log_fmt(text: str, data: Any = ''):
+    return f'{Colors.YELLOW}{text} {Colors.PURPLE}{data}{Colors.END}'
+
+
+def int2str(li: list):
+    t = []
+    for i in li:
+        t.append(str(i))
+    return t