feat(qq_api): ✨ use aiocqhttp as sdk

qq_api/qq_api/__init__.py

@@ -1,99 +1,88 @@
-# -*- coding: utf-8 -*-
-import requests
-import os
+import logging
 from threading import Thread
 
-from mcdreforged.api.command import *
+from uvicorn import Config, Server
+from aiocqhttp import CQHttp, Event
+from mcdreforged.api.utils import Serializable
 from mcdreforged.api.event import LiteralEvent
+from mcdreforged.api.types import PluginServerInterface
+
+__all__ = [
+    "get_bot"
+]
+
+__bot: CQHttp
+__uvicorn_server: Server
+__mcdr_server: PluginServerInterface
+
+
+class PluginConfig(Serializable):
+    api_host: str = "127.0.0.1"
+    api_port: int = 5700
+    post_host: str = "127.0.0.1"
+    post_port: int = 5701
+
+
+def on_load(server: PluginServerInterface, old):
+    global __bot, __uvicorn_server, __mcdr_server
+
+    # mcdr init
+    __mcdr_server = server
+    config = server.load_config_simple(target_class=PluginConfig)
+
+    # cqhttp init
+    __bot = CQHttp(api_root=f"http://{config.api_host}:{config.api_port}")
+    __uvicorn_server = Server(Config(
+        __bot.server_app,
+        host=config.post_host,
+        port=config.post_port,
+        log_level=logging.CRITICAL
+    ))
+
+    @__bot.on_message
+    async def on_message(event: Event):
+        server.logger.debug(f"on message: {event}")
+        server.dispatch_event(
+            LiteralEvent("qq_api.on_message"),
+            (__bot, event)
+        )
+
+    @__bot.on_notice
+    async def on_notice(event: Event):
+        server.logger.debug(f"on notice: {event}")
+        server.dispatch_event(
+            LiteralEvent("qq_api.on_notice"),
+            (__bot, event)
+        )
+
+    @__bot.on_request
+    async def on_request(event: Event):
+        server.logger.debug(f"on request: {event}")
+        server.dispatch_event(
+            LiteralEvent("qq_api.on_request"),
+            (__bot, event)
+        )
 
-from qq_api.constant import NAME, VERSION, RELEASE_URL, FilePath
-from qq_api.functions import touch_folder, version_compare
-from qq_api.factory import Factory
-
-HELP_MESSAGE = '§6!!cq update §7检查并自动更新'
-
-touch_folder(FilePath.DATA_DIR)
-factory = Factory()
-
-
-def on_load(server, old):
-    server.register_help_message('!!cq', 'CoolQAPI插件帮助')
-    server.register_command(
-        Literal('!!cq').
-            runs(lambda src: src.reply(HELP_MESSAGE)).
-            then(
-            Literal('update').
-                runs(
-                lambda src: Thread(
-                    target=check_update,
-                    args=(src.get_server(), src),
-                    name=NAME
-                ).start()
-            )
+    @__bot.on_meta_event
+    async def on_message(event: Event):
+        server.logger.debug(f"on meta event: {event}")
+        server.dispatch_event(
+            LiteralEvent("qq_api.on_meta_event"),
+            (__bot, event)
         )
-    )
-
-    Thread(target=check_update, args=(server,), name=NAME).start()
-    factory.injection(server)
-    server.dispatch_event(LiteralEvent('cool_q_api.on_qq_load'), (factory.bot,))
-    factory.post_server.start()
-
-
-def on_unload(server):
-    stop()
-
-
-def on_mcdr_stop(server):
-    stop()
-
-
-def stop():
-    requests.post(
-        f'http://{factory.config["post_host"]}:{factory.config["post_port"]}/'
-        f'{factory.config["post_path"]}',
-        json={'shutdown': True}
-    )
-
-
-def check_update(server, src=None):
-    def download(link, path):
-        if not os.path.isdir(FilePath.UPDATE_DIR):
-            os.mkdir(FilePath.UPDATE_DIR)
-        with open(path, 'wb') as f:
-            f.write(requests.get(link).content)
-
-    try:
-        server.logger.info('检测更新中')
-        r = requests.get(RELEASE_URL).json()
-        try:
-            compare = version_compare(r['tag_name'], VERSION)
-        except ValueError:
-            server.logger.error('您的版本号错误, 自动为您下载最新版中')
-            compare = 1
-        if compare == 0:
-            server.logger.info('qq_api 已为最新版')
-        elif compare == 1:
-            server.logger.info('发现新版本: ' + r['tag_name'])
-            file_path = os.path.join(
-                FilePath.UPDATE_DIR,
-                f'{NAME}-{r["tag_name"]}.zip'
-            )
-            download(r['assets'][0]['browser_download_url'], file_path)
-            server.logger.info(f'更新下载完成, 文件位于 {file_path}')
-        elif compare == -1:
-            server.logger.info('检测到 qq_api 为开发版')
-    except:
-        if src is None:
-            server.logger.error('qq_api 更新失败')
-        else:
-            src.reply('§cCoolQAPI 更新失败')
-
-
-def get_bot():
-    """return bot object"""
-    return factory.bot
-
-
-def get_config():
-    """get the api config"""
-    return factory.config
+
+    Thread(target=__uvicorn_server.run, name="QQ API Server").start()
+    server.logger.info("Bot listener server started.")
+
+
+def on_unload(server: PluginServerInterface):
+    __mcdr_server.logger.info("Exiting bot listener server.")
+    __uvicorn_server.should_exit = True
+
+
+def get_bot() -> CQHttp:
+    """
+    Get CQHttp instance.
+    :return: CQHttp.
+    """
+    return __bot