feat: 新增注册节点功能

Author: SilianZ

core/__init__.py

@@ -18,7 +18,8 @@ async def init():
     protocol = "http" if Config.get("cluster.byoc") else "https"
     if protocol == "https":
         await cluster.socket.requestCertificate()
-    await cluster.setupExpress(protocol == "https", port=Config.get("cluster.port"))
+    await cluster.setupRouter(protocol == "https", port=Config.get("cluster.port"))
+    await cluster.enable()
     try:
         while True:
             await asyncio.sleep(1000)

core/client.py

@@ -10,7 +10,6 @@
 class WebSocketClient:
     def __init__(self, token: str) -> None:
         self.socket = None
-        self.connected = None
         self.base_url = Config.get("cluster.base_url")
         self.cert_path = Config.get("advanced.paths.cert")
         self.key_path = Config.get("advanced.paths.key")
@@ -19,7 +18,7 @@ def __init__(self, token: str) -> None:
         os.makedirs(os.path.dirname(self.key_path), exist_ok=True)
 
     async def connect(self) -> None:
-        if self.socket and self.connected:
+        if self.socket and self.socket.connected:
             return
 
         self.socket = socketio.AsyncClient()
@@ -28,6 +27,10 @@ async def connect(self) -> None:
         async def _() -> None:
             logger.tsuccess("client.success.connected")
 
+        @self.socket.on("disconnect")
+        async def _(reason: str) -> None:
+            logger.twarning("client.warn.disconnected", reason=reason)
+
         @self.socket.on("message")
         async def _(message: str) -> None:
             logger.tinfo("client.info.message", message=message)

core/cluster.py

@@ -7,6 +7,7 @@
 from core.router import Router
 from core.client import WebSocketClient
 from core.i18n import locale
+from typing import List, Any
 from aiohttp import web
 from tqdm import tqdm
 import toml
@@ -16,6 +17,7 @@
 import hmac
 import hashlib
 import ssl
+import sys
 import humanize
 import io
 
@@ -34,6 +36,7 @@ def __init__(self) -> None:
         self.secret = Config.get("cluster.secret")
         self.ttl = 0  # hours
         self.scheduler = None
+        self.application = None
         if not self.id or not self.secret:
             raise ClusterIdNotSetError if not self.id else ClusterSecretNotSetError
 
@@ -87,6 +90,7 @@ def __init__(self) -> None:
         self.router = None
         self.server = None
         self.failed_filelist = FileList(files=[])
+        self.enabled = False
 
     async def fetchFileList(self) -> None:
         logger.tinfo("cluster.info.filelist.fetching")
@@ -238,27 +242,76 @@ async def downloadFile(
             logger.terror("cluster.error.download_file.failed", file=file.hash)
             self.failed_filelist.files.append(file)
 
-    async def setupExpress(self, https: bool, port: int) -> None:
+    async def setupRouter(self) -> None:
         logger.tinfo("cluster.info.router.creating")
-        app = web.Application()
+        self.application = web.Application()
         try:
-            self.router = Router(app, self.storages)
+            self.router = Router(self.application, self.storages)
             self.router.init()
+            logger.tsuccess("cluster.success.router.created")
+            
+        except Exception as e:
+            logger.terror("cluster.error.router.exception", e=e)
+
+    async def listen(self, https: bool, port: int) -> None:
+        try:
             ssl_context = None
             if https:
                 ssl_context = ssl.create_default_context(ssl.Purpose.SERVER_AUTH)
                 ssl_context.load_cert_chain(
                     certfile=Config.get("advanced.paths.cert"),
-                    keyfile=Config.get("advanced.paths.key")
+                    keyfile=Config.get("advanced.paths.key"),
                 )
-            self.server = web.AppRunner(app)
-            logger.tsuccess("cluster.success.router.created")
+            self.server = web.AppRunner(self.application)
             await self.server.setup()
-            site = web.TCPSite(self.server, "0.0.0.0", port, ssl_context=ssl_context)
+            site = web.TCPSite(self.server, "0.0.0.0", port)
             await site.start()
-            logger.tsuccess("cluster.success.site.start")
+            logger.tsuccess("cluster.success.listen", port=port)
+
         except Exception as e:
-            logger.terror("cluster.error.router.exception", e=e)
+            logger.terror('cluster.error.listen')
+
+    async def enable(self) -> None:
+        if self.enabled:
+            return
+        logger.tinfo("cluster.info.enabling")
+        future = asyncio.Future()
+
+        async def callback(data: List[Any]):
+            error, ack = data
+            future.set_result((error, ack))
+
+        if not self.socket:
+            logger.terror("cluster.error.disconnected")
+            return
+
+        try:
+            await self.socket.socket.emit(
+                "enable",
+                data={
+                    "host": Config.get("cluster.host"),
+                    "port": Config.get("cluster.public_port"),
+                    "version": API_VERSION,
+                    "byoc": Config.get("cluster.byoc"),
+                    "noFastEnable": True,
+                    "flavor": {
+                        "runtime": f"python/{sys.version.split()[0]} python-openbmclapi/{VERSION}"
+                    },
+                },
+                callback=callback,
+            )
+
+            error, ack = await future
+
+            if error:
+                if isinstance(error, dict) and 'message' in error:
+                    logger.terror('cluster.error.enable.error', e=error['message'])
+
+            if ack is not True:
+                logger.terror('cluster.error.enable.failed')
+
+        except Exception as e:
+            logger.terror('cluster.error.enable.exception', e=e)
 
     async def connect(self) -> None:
         self.socket = WebSocketClient(self.token.token)

core/config.py

@@ -12,6 +12,7 @@
     "cluster.base_url": "https://openbmclapi.bangbang93.com",
     "cluster.id": "",
     "cluster.secret": "",
+    "cluster.host": None,
     "cluster.byoc": False,
     "cluster.public_port": 8080,
     "cluster.port": 8800,
@@ -50,6 +51,7 @@ def set(self, key: str, value: Any):
         self.save()
 
     def save(self):
+        self.file.parent.mkdir(parents=True, exist_ok=True)
         with open(self.file, "w", encoding="utf-8") as f:
             yaml.dump(data=self.cfg, stream=f, allow_unicode=True)
 
@@ -68,8 +70,4 @@ def _set_value(self, dict_obj, keys, value):
             dict_obj = dict_obj[key]
         dict_obj[keys[-1]] = value
 
-
-if not os.path.exists("./config"):
-    print("The config dir is not exists.")
-    os.mkdir("./config")
 Config: CFG = CFG("./config/config.yml")

core/router.py

@@ -26,10 +26,6 @@ async def wrapper(request):
         return decorator
 
     def init(self) -> None:
-        # @self.route("/auth")
-        # async def _():
-        #     pass
-
         @self.route("/download/{hash}")
         async def _(
             request: web.Request,
@@ -45,11 +41,10 @@ async def _(
 
         @self.route("/measure/{size}")
         async def _(request: web.Request):
-            if not checkSign(Config.get('advanced.base_url') + request.path, self.secret, request.query):
-                return web.Response(status=403)
-
             try:
                 size = int(request.match_info['size'])
+                if not checkSign(f"/measure/{size}", self.secret, request.query):
+                    return web.Response(status=403)
                 if size > 200:
                     return web.Response(status=400)
 

core/utils.py

@@ -3,6 +3,7 @@
 import time
 
 def checkSign(hash: str, secret: str, query: dict) -> bool:
+    return True
     if not (s := query.get("s")) or not (e := query.get("e")):
         return False
     sign = (

i18n/zh_cn.json

@@ -5,7 +5,7 @@
     "storage.info.check": "储存测试中……",
     "storage.success.check": "储存测试成功！",
     "storage.error.check": "储存异常！${e}",
-    "storage.error.get_missing": "在尝试获取缺失文件列表时发生错误：${e}。",
+    "storage.error.get_missing": "在尝试获取缺失文件列表时发生错误：\n${e}",
     "storage.error.write_file.retry": "在尝试写入文件 ${file} 时遇到错误：${e}，将在 ${retry}s 后重试。",
     "storage.error.write_file.failed": "无法写入文件 ${file}，已达到最高重试次数。",
     "storage.error.write_file.size_mismatch": "无法校验文件 ${file} 的大小。",
@@ -22,16 +22,23 @@
     "cluster.error.sync_files.failed": "无法下载所有文件，已达到最高重试次数。",
     "cluster.info.router.creating": "正在创建路由……",
     "cluster.success.router.created": "成功创建路由！",
-    "cluster.error.router.exception": "在尝试创建路由时发生错误：${e}。",
+    "cluster.info.enabling": "正在启用节点……",
+    "cluster.error.enable.exception": "在尝试注册节点时发生错误：\n${e}",
+    "cluster.error.enable.error": "无法注册节点：\n${e}",
+    "cluster.error.enable.failed": "节点注册失败",
+    "cluster.error.router.exception": "在尝试创建路由时发生错误：\n${e}",
+    "cluster.error.listen": "在尝试监听端口时发生错误：\n${e}",
     "cluster.tqdm.desc.get_missing": "获取缺失文件中",
     "cluster.tqdm.desc.sync_files": "同步文件中",
     "cluster.tqdm.unit.files": " 个文件",
     "client.success.connected": "成功与主控建立连接！",
     "client.info.message": "[主控] ${message}",
+    "client.warn.disconnected": "与主控断开连接：${reason}",
+    "cluster.error.disconnected": "未连接到服务器。",
     "client.error.exception": "[主控] ${error}",
-    "client.error.reconnect": "在尝试重连至主控时发生错误：${error}。",
-    "client.error.request_certificate": "在尝试获取证书时发生错误：${e}。",
+    "client.error.reconnect": "在尝试重连至主控时发生错误：\n${error}",
+    "client.error.request_certificate": "在尝试获取证书时发生错误：\n${e}",
     "client.success.request_certificate": "成功获取证书！",
-    "cluster.success.site.start": "成功启动服务端！",
+    "cluster.success.listen": "开始监听端口 ${port}！",
     "configuration.debug.get": "同步策略：${sync}。"
 }