gitea_push2qq/server.py

import tomllib
from typing import Dict, Any, List, Tuple

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 command import command_convert
from gitea_model import WebHookIssueComment, WebHookIssue, GiteaEvent
from model import Ctx, SioConfig, Message
from unit import sio_log_format

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'])


@app.before_server_start
async def setup_before_start(_app):
    _app.ctx.sio_config = get_config()
    _app.ctx.sio = AsyncClient()

    start_sio_listener()

    await _app.ctx.sio.connect(_app.ctx.sio_config.HOST)


@app.post('/receive')
async def receive(rqt: Request):
    match (rqt.headers['X-Gitea-Event']):
        case GiteaEvent.issues.value:
            data = WebHookIssue(**rqt.json)
            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_link = f"View it: {data.issue.html_url}"
            cancel_subscribe = f"unsubscribe: {data.repository.html_url}"

            rsp = f"{rsp_title}\n\n{rsp_sender}\n------\n{rsp_ctx}\n------\n{rsp_link}\n{cancel_subscribe}"

        case GiteaEvent.issue_comment.value:
            data = WebHookIssueComment(**rqt.json)
            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_link = f"View it: {data.comment.html_url}"
            cancel_subscribe = f"unsubscribe: {data.repository.html_url}"

            rsp = f"{rsp_title}\n\n{rsp_sender}\n------\n{rsp_ctx}\n------\n{rsp_link}\n{cancel_subscribe}"

        case _:
            rsp = "Unknown webhook type! Please contact the administrator."

    message = Message(content=rsp, room_id=-777186831)
    await app.ctx.sio.emit('sendMessage', message.to_json())
    return text(rsp)


@app.get('/redirect')
async def redirect(rqt: Request):
    print(rqt.args)
    return text('success')


"""
以下为QQ监听部分
"""


def start_sio_listener():
    @app.ctx.sio.on('connect')
    def connect():
        logger.info(f'{Colors.GREEN}icalingua 已连接{Colors.END}')

    @app.ctx.sio.on('requireAuth')
    async def require_auth(salt: str, versions: Dict[str, str]):
        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))
        signature = sign.sign(bytes.fromhex(salt))

        # 发送数据
        logger.info(f"{len(signature.signature)=} {type(signature.signature)=}{Colors.END}")
        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]):
        logger.debug(sio_log_format('add_message:', data))

        sender_id = data['message']['senderId']
        is_self = (sender_id == app.ctx.sio_config.SELF_ID)

        sender_name = data['message']['username']
        content = data['message']['content']
        room_id = data['roomId']

        if not is_self:
            msg = command_convert(data)
            if msg is not None:
                await app.ctx.sio.emit('sendMessage', msg.to_json())


if __name__ == "__main__":
    app.run(host='0.0.0.0', port=80, dev=True)