fix: 修了点东西

Author: tianxiu2b2t

core/__init__.py

@@ -13,12 +13,27 @@
 _WAITLOCK = utils.CountLock()
 _START_RUNTIME = time.monotonic()
 
+async def call(module, func: str):
+    try:
+        init = getattr(module, func)
+        if asyncio.iscoroutinefunction(init):
+            await init()
+        else:
+            await asyncio.get_event_loop().run_in_executor(None, init)
+    except:
+        logger.traceback()
+
 async def main():
     start = time.monotonic_ns()
-    await scheduler.init()
-    await web.init()
-    await dashboard.init()
-    await cluster.init()
+    await asyncio.gather(*[
+        call(m, "init") for m in (
+            scheduler,
+            web,
+            database,
+            dashboard,
+            cluster,
+        )
+    ])
     _WAITLOCK.acquire()
     end = time.monotonic_ns()
     logger.tsuccess("main.success.start_service_done", time=f"{((end-start) / 1e9):.2f}")
@@ -27,9 +42,14 @@ async def main():
     except:
         logger.tdebug("main.debug.service_unfinish")
     finally:
-        await cluster.unload()
-        await web.unload()
-        await scheduler.unload()
+        await asyncio.gather(*[
+            call(m, "unload") for m in (
+                scheduler,
+                web,
+                cluster,
+                database
+            )
+        ])
 
 def read_version():
     with open("VERSION", "r") as f:

core/cluster.py

@@ -49,7 +49,7 @@ def __init__(self, clusters: 'ClusterManager'):
         self.cache_filelist: defaultdict[storages.iStorage, defaultdict[str, storages.File]] = defaultdict(defaultdict) # type: ignore
 
     def init(self):
-        scheduler.run_repeat(self._check_available, 120)
+        scheduler.run_repeat_later(self._check_available, 1, 120)
 
     async def _check_available(self):
         for storage in self.storages:
@@ -61,6 +61,7 @@ async def _check_available(self):
                 )
             if await storage.get_size(CHECK_FILE_MD5) == len(CHECK_FILE_CONTENT) and storage not in self.available_storages:
                 self.available_storages.append(storage)
+        logger.debug(f"Available storages: {len(self.available_storages)}")
         if len(self.available_storages) > 0:
             self.check_available.release()
         else:
@@ -88,6 +89,7 @@ async def get_missing_files(self, files: set[File]) -> set[File | Any]:
             logger.twarning("cluster.warning.no_check_function")
             function = self._check_exists
 
+        await self.available()
         with tqdm(
             total=len(self.available_storages) * 256,
             desc="List Files",
@@ -108,7 +110,7 @@ async def get_missing_files(self, files: set[File]) -> set[File | Any]:
                 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()
+            return missing_files
 
     async def get_storage_filelist(self, storage: storages.iStorage, pbar: tqdm):
         result = await storage.list_files(pbar)
@@ -736,28 +738,28 @@ async def keepalive(self):
         ping = (time.time() - timestamp) // 0.0002
         logger.tsuccess("cluster.success.keepalive", cluster=self.id, hits=units.format_number(total_counter.hits), bytes=units.format_bytes(total_counter.bytes), ping=ping)
 
-        storage_data = []
-        for storage, counter in self.counter.items():
-            storage_data.append(
-                db.StorageStatistics(
-                    storage=storage.unique_id,
-                    hits=counter.hits,
-                    bytes=counter.bytes,
-                )
-            )
-        storage_data.append(
-            db.StorageStatistics(
-                None,
-                commit_no_storage_counter.hits,
-                commit_no_storage_counter.bytes
-            )
-        )
-        db.add_statistics(
-            db.Statistics(
-                self.id,
-                storage_data
-            )
-        )
+        #storage_data = []
+        #for storage, counter in self.counter.items():
+        #    storage_data.append(
+        #        db.StorageStatistics(
+        #            storage=storage.unique_id,
+        #            hits=counter.hits,
+        #            bytes=counter.bytes,
+        #        )
+        #    )
+        #storage_data.append(
+        #    db.StorageStatistics(
+        #        None,
+        #        commit_no_storage_counter.hits,
+        #        commit_no_storage_counter.bytes
+        #    )
+        #)
+        #db.add_statistics(
+        #    db.Statistics(
+        #        self.id,
+        #        storage_data
+        #    )
+        #)
 
     async def disable(self, exit: bool = False):
         scheduler.cancel(self.keepalive_task)
@@ -1055,6 +1057,12 @@ async def _(request: aweb.Request):
         start = request.http_range.start or 0
         end = request.http_range.stop or file.size
         size = end - start + 1
+        storage_name = file.storage.unique_id if file.storage is not None else None
+        db.add_file(
+            cluster.id,
+            storage_name,
+            size
+        )
 
         cluster.hit(file.storage, size)
         # add database

core/config.py

@@ -4,10 +4,9 @@
 import os
 
 defaults = {
-    "advanced.api_version": "1.11.0",
     "advanced.lang": "zh_cn",
     "advanced.debug": False,
-    "advanced.sync_interval": 60,
+    "advanced.sync_interval": 600,
     "advanced.base_url": "https://openbmclapi.bangbang93.com",
     "advanced.threads": 128,
     "advanced.ssl_dir": ".ssl",
@@ -137,6 +136,6 @@ def check_type(self):
 
     @property
     def sync_interval(self):
-        return Config.get("advanced.sync_interval", 600)
+        return max(Config.get("advanced.sync_interval", 600), 600)
 
 const = Const()

core/database.py

@@ -1,10 +1,13 @@
+from collections import defaultdict
 from dataclasses import dataclass
 import time
 from typing import Optional
-from sqlalchemy import create_engine, Column, Integer, String
+from sqlalchemy import create_engine, Column, Integer, String, LargeBinary
 from sqlalchemy.ext.declarative import declarative_base
 from sqlalchemy.orm import sessionmaker
 
+from core import logger, scheduler
+
 @dataclass
 class StorageStatistics:
     storage: Optional[str] = None
@@ -16,6 +19,22 @@ class Statistics:
     cluster_id: str
     storages: list[StorageStatistics]
 
+@dataclass
+class FileStatisticsKey:
+    hour: int
+    cluster_id: str
+    storage_id: Optional[str]
+
+    def __hash__(self):
+        return hash((self.hour, self.cluster_id, self.storage_id))
+
+@dataclass
+class FileStatistics:
+    hits: int = 0
+    bytes: int = 0
+
+
+
 engine = create_engine('sqlite:///database.db')
 Base = declarative_base()
 
@@ -35,6 +54,18 @@ class StorageStatisticsTable(Base):
     hits = Column(String, nullable=False)
     bytes = Column(String, nullable=False)
 
+class ResponseTable(Base):
+    __tablename__ = 'Responses'
+    id = Column(Integer, primary_key=True)
+    hour = Column(Integer, nullable=False)
+    cluster = Column(String, nullable=False)
+    storage = Column(String, nullable=False)
+    success = Column(String, nullable=False)
+    not_found = Column(String, nullable=False)
+    error = Column(String, nullable=False)
+    redirect = Column(String, nullable=False)
+    ip_tables = Column(LargeBinary, nullable=False)
+
 class Session:
     def __init__(self):
         self.session = sessionmaker(bind=engine)()
@@ -47,8 +78,11 @@ def get_session(self):
         return self.session
 
 SESSION = Session()
+FILE_CACHE: defaultdict[FileStatisticsKey, FileStatistics] = defaultdict(lambda: FileStatistics())
 
+@DeprecationWarning
 def add_statistics(data: Statistics):
+    return
     session = SESSION.get_session()
     hits, bytes = 0, 0
     hour = get_hour()
@@ -78,10 +112,87 @@ def add_statistics(data: Statistics):
         }
     )
     session.commit()
-    
 
+def add_file(cluster: str, storage: Optional[str], bytes: int):
+    global FILE_CACHE
+    try:
+        key = FileStatisticsKey(get_hour(), cluster, storage)
+        FILE_CACHE[key].bytes += bytes
+        FILE_CACHE[key].hits += 1
+    except:
+        logger.traceback()
 
 def get_hour():
     return int(time.time() // 3600)
 
-Base.metadata.create_all(engine)
+def _commit_storage(hour: int, storage: Optional[str], hits: int, bytes: int):
+    if hits == bytes == 0:
+        return False
+    session = SESSION.get_session()
+    q = session.query(StorageStatisticsTable).filter_by(hour=hour, storage=storage)
+    r = q.first() or StorageStatisticsTable(hour=hour, storage=storage, hits=str(0), bytes=str(0))
+    if q.count() == 0:
+        session.add(r)
+    q.update(
+        {
+            'hits': str(int(r.hits) + hits), # type: ignore
+            'bytes': str(int(r.bytes) + bytes) # type: ignore
+        }
+    )
+    return True
+
+def _commit_cluster(hour: int, cluster: str, hits: int, bytes: int):
+    if hits == bytes == 0:
+        return False
+    session = SESSION.get_session()
+    q = session.query(ClusterStatisticsTable).filter_by(hour=hour, cluster=cluster)
+    r = q.first() or ClusterStatisticsTable(hour=hour, cluster=cluster, hits=str(0), bytes=str(0))
+    if q.count() == 0:
+        session.add(r)
+    q.update(
+        {
+            'hits': str(int(r.hits) + hits), # type: ignore
+            'bytes': str(int(r.bytes) + bytes) # type: ignore
+        }
+    )
+    return True
+
+def commit():
+    global FILE_CACHE
+    total_hits = 0
+    total_bytes = 0
+    total_storages = 0
+    cache = FILE_CACHE.copy()
+    session = SESSION.get_session()
+    clusters: defaultdict[tuple[int, str], FileStatistics] = defaultdict(lambda: FileStatistics(0, 0))
+    for key, value in FILE_CACHE.items():
+        hour = key.hour
+        cluster = key.cluster_id
+        storage = key.storage_id
+        hits = value.hits
+        bytes = value.bytes
+        if _commit_storage(hour, storage, hits, bytes):
+            total_hits += hits
+            total_bytes += bytes
+            total_storages += 1
+            clusters[(hour, cluster)].hits += hits
+            clusters[(hour, cluster)].bytes += bytes
+    for cluster, value in clusters.items():
+        _commit_cluster(cluster[0], cluster[1], value.hits, value.bytes)
+
+    logger.success(f'Committing {total_hits} hits and {total_bytes} bytes to database. {total_storages} storages updated')
+    
+    session.commit()
+    for key, value in cache.items():
+        FILE_CACHE[key].hits -= value.hits
+        FILE_CACHE[key].bytes -= value.bytes
+        if FILE_CACHE[key].hits == FILE_CACHE[key].bytes == 0:
+            del FILE_CACHE[key]
+    ...
+
+async def init():
+    Base.metadata.create_all(engine)
+    scheduler.run_repeat_later(commit, 5, 10)
+
+async def unload():
+    commit()