添加更多QQ监听函数

This commit is contained in:
San Liy 2023-12-19 00:10:34 +08:00
parent c950460235
commit beaef5f7ef

View File

@ -1,5 +1,5 @@
import tomllib import tomllib
from typing import Dict, Any from typing import Dict, Any, List, Tuple
from nacl.signing import SigningKey from nacl.signing import SigningKey
from sanic import Sanic, Request from sanic import Sanic, Request
@ -9,6 +9,7 @@ from socketio import AsyncClient
from gitea_model import WebHookIssueComment, WebHookIssue, GiteaEvent from gitea_model import WebHookIssueComment, WebHookIssue, GiteaEvent
from model import Ctx, SioConfig, Message from model import Ctx, SioConfig, Message
from unit import sio_log_format
app = Sanic('GiteaPush', ctx=Ctx) app = Sanic('GiteaPush', ctx=Ctx)
@ -112,6 +113,41 @@ def start_sio_listener():
def update_room(data: Dict[str, Any]): def update_room(data: Dict[str, Any]):
logger.info(f"update_room: {data}") logger.info(f"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}'))
if __name__ == "__main__": if __name__ == "__main__":
app.run(host='0.0.0.0', port=80, dev=True) app.run(host='0.0.0.0', port=80, dev=True)