feat: 完善所有基础功能

Author: SilianZ

core/__init__.py

@@ -28,6 +28,9 @@ async def syncFiles():
         await cluster.setupRouter()
         await cluster.listen(protocol == "https", Config.get("cluster.port"))
         await cluster.enable()
+        if not cluster.enabled:
+            raise asyncio.CancelledError
+        scheduler.add_job(cluster.keepAlive, IntervalTrigger(seconds=Config.get("advanced.keep_alive")))
         scheduler.start()
         logger.tsuccess('main.success.scheduler')
         while True:
@@ -47,12 +50,7 @@ async def syncFiles():
 
 
 def init():
-    loop = asyncio.get_event_loop()
-    main_task = loop.create_task(main())
     try:
-        loop.run_until_complete(main_task)
+        asyncio.run(main())
     except KeyboardInterrupt:
-        main_task.cancel()
-        loop.run_until_complete(asyncio.shield(main_task))
-    finally:
-        loop.close()
+        pass

core/classes.py

@@ -24,6 +24,10 @@ class AgentConfiguration:
     source: str
     concurrency: int
 
+@dataclass
+class Counters:
+    hits: int
+    bytes: int
 
 class Storage(ABC):
     @abstractmethod

core/cluster.py

@@ -3,7 +3,7 @@
 from core.scheduler import *
 from core.exceptions import ClusterIdNotSetError, ClusterSecretNotSetError
 from core.storages import getStorages, LocalStorage
-from core.classes import FileInfo, FileList, AgentConfiguration
+from core.classes import FileInfo, FileList, AgentConfiguration, Counters
 from core.router import Router
 from core.client import WebSocketClient
 from core.i18n import locale
@@ -15,6 +15,7 @@
 import aiohttp
 import asyncio
 import hmac
+import datetime
 import hashlib
 import ssl
 import sys
@@ -314,11 +315,59 @@ async def callback(data: List[Any]):
                 return
 
             self.enabled = True
-            logger.tsuccess("cluster.success.enable.enabled", id=self.id, port=Config.get("cluster.public_port"))
+            logger.tsuccess(
+                "cluster.success.enable.enabled",
+                id=self.id,
+                port=Config.get("cluster.public_port"),
+            )
 
         except Exception as e:
             logger.terror("cluster.error.enable.exception", e=e)
 
+    async def keepAlive(self) -> bool:
+        if not self.enabled:
+            logger.terror("cluster.error.keepalive.cluster_not_enabled")
+        if not self.socket:
+            logger.terror("cluster.error.keepalive.socket_not_setup")
+
+        future = asyncio.Future()
+
+        async def callback(data: List[Any]):
+            future.set_result(data)
+
+        counter = self.router.counters
+
+        try:
+            await self.socket.socket.emit(
+                "keep-alive",
+                data={
+                    "time": datetime.datetime.now(datetime.UTC)
+                    **counter,
+                },
+                callback=callback,
+            )
+
+            response = await future
+            error, date = (
+                (response + [None, None])[:2]
+                if isinstance(response, list)
+                else (None, None)
+            )
+            
+            if error:
+                logger.terror("cluster.error.keep_alive.error", e=error)
+                return False
+
+            logger.tsuccess("cluster.success.keep_alive.success", hits=humanize.intcomma(counter.hits), bytes=humanize.naturalsize(counter.bytes, binary=True))
+
+            self.router.counters.bytes -= counter.bytes
+            self.router.counters.hits -= counter.hits
+
+            return bool(date)
+
+        except Exception as e:
+            logger.terror("cluster.error.keep_alive.error", e=e)
+
     async def disable(self) -> None:
         if not self.socket or not self.enabled:
             return

core/config.py

@@ -9,6 +9,7 @@
     "advanced.debug": False,
     "advanced.retry": 5,
     "advanced.delay": 15,
+    "advanced.keep_alive": 10,
     "advanced.sync_interval": 60,
     "cluster.base_url": "https://openbmclapi.bangbang93.com",
     "cluster.id": "",

core/router.py

@@ -1,7 +1,6 @@
 from functools import wraps
 from core.config import Config
-from core.classes import Storage
-from core.logger import logger
+from core.classes import Storage, Counters
 from core.utils import checkSign
 from typing import List, Union
 from aiohttp import web
@@ -13,6 +12,7 @@ def __init__(self, app: web.Application, storages: List[Storage]) -> None:
         self.app = app
         self.secret = Config.get("cluster.secret")
         self.storages = storages
+        self.counters = Counters(hits=0, bytes=0)
 
     def route(self, path, method="GET"):
         def decorator(func):
@@ -38,7 +38,8 @@ async def _(
             data = await random.choice(self.storages).express(
                 file_hash, request, response
             )
-            logger.debug(data)
+            self.counters.bytes += data["bytes"]
+            self.counters.hits += data["hits"]
             return response
 
         @self.route("/measure/{size}")

i18n/zh_cn.json

@@ -29,6 +29,9 @@
     "cluster.info.disabling": "正在禁用节点……",
     "cluster.error.disable.exception": "在尝试禁用节点时发生错误：${e}",
     "cluster.error.enable.exception": "在尝试启用节点时发生错误：${e}",
+    "cluster.error.keep_alive.cluster_not_enabled": "无法保活：节点未启用。",
+    "cluster.error.keep_alive.socket_not_setup": "无法保活：WebSocket 客户端未创建。",
+    "cluster.error.keep_alive.error": "保活失败：${e}",
     "cluster.error.disable.error": "无法禁用节点：${e}",
     "cluster.error.enable.error": "无法启用节点：${e}",
     "cluster.error.enable.failed": "节点启用失败。",