fix: 修复了单位，更新了 web dashboard 等等

Author: tianxiu2b2t

core/__init__.py

@@ -5,12 +5,16 @@
 import atexit
 from . import cluster
 from . import scheduler
+from . import web
+from . import dashboard
 
 _WAITLOCK = utils.CountLock()
 
 async def main():
     start = time.monotonic_ns()
     await scheduler.init()
+    await web.init()
+    await dashboard.init()
     await cluster.init()
     _WAITLOCK.acquire()
     end = time.monotonic_ns()
@@ -20,6 +24,7 @@ async def main():
     except:
         logger.tdebug("main.debug.service_unfinish")
     finally:
+        await web.unload()
         await scheduler.unload()
 
 def init():

core/cluster.py

@@ -71,7 +71,9 @@ async def write_file(self, file: File, content: bytes):
         return all(await asyncio.gather(*(asyncio.create_task(storage.write_file(convert_file_to_storage_file(file), content, file.mtime)) for storage in self.available_storages)))
 
     async def get_missing_files(self, files: set[File]) -> set[File | Any]:
-        function = None
+        async def _(file: File, storage: storages.iStorage):
+            return True
+        function = _
         if function is None:
             logger.twarning("cluster.warning.no_check_function")
             function = self._check_exists
@@ -81,21 +83,24 @@ async def get_missing_files(self, files: set[File]) -> set[File | Any]:
             unit="file",
             unit_scale=True,
         ) as pbar:
-            missing_files = set().union(await asyncio.gather(*(self._get_missing_file(function, file, pbar) for file in files)))
-            if None in missing_files:
-                missing_files.remove(None)
+            missing_files = set()
+            waiting_files: asyncio.Queue[File] = asyncio.Queue()
+            
+            for file in files:
+                waiting_files.put_nowait(file)
+            
+            await asyncio.gather(*(self._get_missing_file_storage(function, missing_files, waiting_files, storage, pbar) for storage in self.available_storages))
             return missing_files or set()
-        
-    
-    async def _get_missing_file(self, function: Callable[..., Coroutine[Any, Any, bool]], file: File, pbar: tqdm):
-        if all(await asyncio.gather(*(self._get_missing_file_storage(function, file, storage, pbar) for storage in self.available_storages))):
-            return None
-        return file
 
-    async def _get_missing_file_storage(self, function: Callable[..., Coroutine[Any, Any, bool]], file: File, storage: storages.iStorage, pbar: tqdm):
-        result = await function(file, storage)
-        pbar.update(1)
-        return result
+    async def _get_missing_file_storage(self, function: Callable[..., Coroutine[Any, Any, bool]], missing_files: set[File], files: asyncio.Queue[File], storage: storages.iStorage, pbar: tqdm):
+        while not files.empty():
+            file = await files.get()
+            if await function(file, storage):
+                pbar.update(1)
+            else:
+                missing_files.add(file)
+            files.task_done()
+            await asyncio.sleep(0)
 
     async def _check_exists(self, file: File, storage: storages.iStorage):
         return await storage.exists(file.hash)
@@ -161,7 +166,7 @@ class FileListManager:
     def __init__(self, clusters: 'ClusterManager'):
         self.clusters = clusters
         self.cluster_last_modified: defaultdict['Cluster', int] = defaultdict(lambda: 0)
-        self.sync_sem: utils.SemaphoreLock = utils.SemaphoreLock(10)
+        self.sync_sem: utils.SemaphoreLock = utils.SemaphoreLock(256)
         self.download_statistics = DownloadStatistics()
 
     async def _get_filelist(self, cluster: 'Cluster'):
@@ -205,6 +210,10 @@ async def sync(self):
 
         missing = await self.clusters.storage_manager.get_missing_files(result)
 
+        if not missing:
+            logger.tsuccess("cluster.success.no_missing_files")
+            return
+
         await self.clusters.storage_manager.available()
         configurations: defaultdict[str, deque[OpenBMCLAPIConfiguration]] = defaultdict(deque)
         for configuration in await asyncio.gather(*(asyncio.create_task(self._get_configuration(cluster)) for cluster in self.clusters.clusters)):
@@ -213,7 +222,7 @@ async def sync(self):
         # get better configuration
         configuration = max(configurations.items(), key=lambda x: x[1][0].concurrency)[1][0]
         logger.tinfo("cluster.info.sync_configuration", source=configuration.source, concurrency=configuration.concurrency)
-        self.sync_sem.set_value(configuration.concurrency)
+        #self.sync_sem.set_value(configuration.concurrency)
 
         await self.download(missing)
 
@@ -315,7 +324,7 @@ def report(self, file: File, error: Exception, resp: Optional[aiohttp.ClientResp
                 responses.append(f"{r.status} | {r.url}")
             responses.append(f"{resp.status} | {resp.url}")
             host = resp.host
-        hash = msg[0] if len(msg) > 0 else None
+        hash = msg[0] if len(msg) > 0 and type == "file" else None
         logger.debug(error)
         logger.terror(f"clusters.error.downloading", type=type, file_hash=file.hash, file_size=units.format_bytes(file.size), host=host, file_path=file.path, hash=hash, responses="\n".join(("", *responses)))
 
@@ -351,6 +360,8 @@ async def start(self):
         # check files
         await self.file_manager.sync()
 
+        # start job
+
 
 
 class Cluster:
@@ -455,7 +466,10 @@ async def init():
             continue
         logger.tsuccess("cluster.success.load_cluster", id=cluster.id, host=cluster.host, port=cluster.port)
         clusters.add_cluster(cluster)
-    
+    if len(clusters.clusters) == 0:
+        logger.terror("cluster.error.no_cluster")
+        utils.pause()
+        return
     config_storages = config_clusters = config.Config.get("storages")
     for cstorage in config_storages:
         type = cstorage['type']

core/config.py

@@ -13,14 +13,16 @@
     "advanced.sync_interval": 60,
     "advanced.base_url": "https://openbmclapi.bangbang93.com",
     "advanced.threads": 128,
+    "web": {
+        "port": -1,
+        "public_port": 6000
+    },
     "clusters": [
         {
             "id": "",
             "secret": "",
             "host": "",
             "byoc": False,
-            "public_port": -1,
-            "port": 8800,
             "cert": "./cert/cert.pem",
             "key": "./cert/key.pem",
         }
@@ -93,4 +95,8 @@ def base_url(self) -> str:
     def threads(self):
         return Config.get("advanced.threads", 128)
 
+    @property
+    def public_port(self):
+        return Config.get("web.public_port", 6000)
+    
 const = Const()

core/dashboard.py

@@ -0,0 +1,17 @@
+from .web import routes as route
+from aiohttp import web
+
+@route.get('/dashboard/')
+@route.get("/dashboard/{tail:.*}")
+async def _(request: web.Request):
+    return web.FileResponse("./assets/index.html")
+
+@route.get('/')
+async def _(request: web.Request):
+    return web.HTTPFound('/dashboard/')
+
+route.static("/assets", "./assets")
+
+
+async def init():
+    ...

core/units.py

@@ -3,60 +3,64 @@
 NUMBER_UNITS = (
     ('', 1),
     ('K', 1e3),
-    ('M', 1e6),
-    ('G', 1e9),
-    ('T', 1e12),
-    ('P', 1e15),
-    ('E', 1e18),
-    ('Z', 1e21),
-    ('Y', 1e24),
+    ('M', 1e3),
+    ('G', 1e3),
+    ('T', 1e3),
+    ('P', 1e3),
+    ('E', 1e3),
+    ('Z', 1e3),
+    ('Y', 1e3),
 )
 
 TIME_UNITS = (
     ('ns', 1),
     ('μs', 1e3),
-    ('ms', 1e6),
-    ('s', 1e9),
-    ('min', 6e10),
-    ('h', 3.6e12),
+    ('ms', 1e3),
+    ('s', 1e3),
+    ('min', 60),
+    ('h', 60),
 )
 # 1024
 BYTES_UNITS = (
     ('iB', 1),
     ('KiB', 1024),
-    ('MiB', 1024**2),
-    ('GiB', 1024**3),
-    ('TiB', 1024**4),
-    ('PiB', 1024**5),
-    ('EiB', 1024**6),
-    ('ZiB', 1024**7),
-    ('YiB', 1024**8)
+    ('MiB', 1024),
+    ('GiB', 1024),
+    ('TiB', 1024),
+    ('PiB', 1024),
+    ('EiB', 1024),
+    ('ZiB', 1024),
+    ('YiB', 1024)
 )
 
 def format_bytes(n: float) -> str:
     i = 0
-    while n >= 1024 and i < len(BYTES_UNITS) - 1:
-        n /= 1024
+    for u, un in BYTES_UNITS[1:]:
+        if n / un < 1:
+            break
+        n /= un
         i += 1
     return f'{n:.2f}{BYTES_UNITS[i][0]}'
 
 
 def format_number(n: float) -> str:
-    i = 0   
-    while n >= 1000 and i < len(NUMBER_UNITS) - 1:
-        n /= 1000
+    i = 0
+    for u, un in NUMBER_UNITS[1:]:
+        if n / un < 1:
+            break
+        n /= un
         i += 1
     return f'{n:.2f}{NUMBER_UNITS[i][0]}'
 
 
-def format_count_time(n: float) -> str:
+def format_count_time(n: float, round: int = 2) -> str:
     i = 0
-    for unit, t in TIME_UNITS:
-        if n < t:
+    for u, un in TIME_UNITS[1:]:
+        if n / un < 1:
             break
-        n /= t
+        n /= un
         i += 1
-    return f'{n:.2f}{TIME_UNITS[i][0]}'
+    return f'{n:.{round}f}{TIME_UNITS[i][0]}'
 
 def format_count_datetime(secs: float) -> str:
     ms = int(secs * 1000)

core/utils.py

@@ -121,4 +121,11 @@ def get_hash_hexdigest(origin: str, content: bytes):
     h = hashlib.sha1
     if len(origin) == 32:
         h = hashlib.md5
-    return h(content).hexdigest()
+    return h(content).hexdigest()
+
+def pause():
+    try:
+        input("Press Enter to continue...")
+    except KeyboardInterrupt:
+        exit()
+        pass

core/web.py

@@ -0,0 +1,78 @@
+import asyncio
+import time
+from typing import Any, Optional
+from aiohttp import web
+from aiohttp.web_urldispatcher import SystemRoute
+
+from . import units
+from . import config
+from .logger import logger
+
+@web.middleware
+async def middleware(request: web.Request, handler: Any) -> web.Response:
+    with request.match_info.set_current_app(app):
+        start = time.monotonic_ns()
+        resp = None
+        try:
+            resp = await handler(request)
+            return resp
+        finally:
+            status = 500
+            if isinstance(request.match_info.route, SystemRoute):
+                status = request.match_info.route.status
+            if resp is not None:
+                if isinstance(resp, web.StreamResponse):
+                    status = resp.status
+            end = time.monotonic_ns()
+            logger.tdebug("web.debug.request_info", time=units.format_count_time(end - start, 4).rjust(16), host=request.host, address=(request.remote or "").rjust(16), user_agent=request.headers.get("User-Agent"), real_path=request.raw_path, method=request.method.ljust(9), status=status)
+
+
+routes = web.RouteTableDef()
+
+app = web.Application(
+    middlewares=[
+        middleware
+    ]
+)
+runner: Optional[web.AppRunner] = None
+site: Optional[web.TCPSite] = None
+public_server: Optional[asyncio.Server] = None
+
+async def get_free_port():
+    async def _(_, __):
+        ...
+    server = await asyncio.start_server(_, port=-1)
+    port = server.sockets[0].getsockname()[1]
+    server.close()
+    await server.wait_closed()
+    return port
+
+async def handle(reader: asyncio.StreamReader, writer: asyncio.StreamWriter):
+    ...
+
+async def init():
+    global runner, site, public_server, routes, app
+
+    app.add_routes(routes)
+
+    runner = web.AppRunner(app)
+    await runner.setup()
+
+    port = await get_free_port()
+
+    site = web.TCPSite(runner, '0.0.0.0', port)
+    await site.start()
+
+    logger.tdebug("web.debug.local_port", port=site._port)
+
+
+    public_server = await asyncio.start_server(
+        handle, port=config.const.public_port
+    )
+    logger.tsuccess("web.success.public_port", port=config.const.public_port)
+
+
+async def unload():
+    global app
+    await app.cleanup()
+    await app.shutdown()

i18n/zh_cn.json

@@ -13,5 +13,9 @@
     "cluster.processbar.download_files": "下载文件",
     "i18n.prompt.failed": " 未找到 i18n 字符串",
     "cluster.warning.no_check_function": "没有找到检查函数，使用类型为 exists",
-    "clusters.error.downloading": "下载时候出错，原因 [${type}]，文件 [${file_path} ${file_hash}(${file_size})]，主机 [${host}] 下载时候的历史记录：${responses}"
+    "clusters.error.downloading": "下载时候出错，原因 [${type}]，文件 [${file_path} ${file_hash}(${file_size})] 下载后的哈希值 [${hash}]，主机 [${host}] 下载时候的历史记录：${responses}",
+    "web.debug.local_port": "内部开放端口为 [${port}]",
+    "web.success.public_port": "当前 Web 服务器在 [${port}] 端口上开启",
+    "web.error.any": "Web 服务器出错 ${e}",
+    "web.debug.request_info": "${host} | ${time} | ${address} | ${method} ${status} | ${real_path} - ${user_agent}"
 }