Bumping version

Update download.py

biliup/__init__.py

@@ -2,7 +2,7 @@ import logging
 import platform
 import sys
 
-__version__ = "0.4.54"
+__version__ = "0.4.55"
 
 LOG_CONF = {
     'version': 1,

biliup/app.py

@@ -67,10 +67,10 @@ class PluginInfo:
         else:
             temp.url_list = [url]
             self.checker[key] = temp
-            from .plugins.twitch import Twitch
-            if temp == Twitch:
-                # 如果支持批量检测，目前只有一个支持，第一版先写死按照特例处理
-                self.batch_check_task()
+            from .engine.download import BatchCheck
+            if issubclass(temp, BatchCheck):
+                # 如果支持批量检测
+                self.batch_check_task(temp)
             else:
                 self.coroutines[key] = asyncio.create_task(shot(temp))
         self.inverted_index[url] = name
@@ -93,23 +93,21 @@ class PluginInfo:
             del self.coroutines[exec_del]
 
     def init_tasks(self):
-        from .engine.download import DownloadBase
-        from .plugins.twitch import Twitch
+        from .engine.download import BatchCheck
 
         for key, plugin in self.checker.items():
-            if plugin == Twitch:
-                # 如果支持批量检测，目前只有一个支持，第一版先写死按照特例处理
-                self.batch_check_task()
+            if issubclass(plugin, BatchCheck):
+                # 如果支持批量检测
+                self.batch_check_task(plugin)
                 continue
             self.coroutines[key] = asyncio.create_task(shot(plugin))
 
-    def batch_check_task(self):
+    def batch_check_task(self, plugin):
         from biliup.engine.event import Event
         from biliup.handler import CHECK
-        from .plugins.twitch import Twitch
 
         async def check_timer():
-            event_manager.send_event(Event(CHECK, (Twitch, None, None)))
+            event_manager.send_event(Event(CHECK, (plugin, None, None)))
 
         timer = Timer(func=check_timer, interval=30)
-        self.coroutines[Twitch.__name__] = asyncio.create_task(timer.astart())
+        self.coroutines[plugin.__name__] = asyncio.create_task(timer.astart())

biliup/engine/decorators.py

@@ -23,19 +23,21 @@ class Plugin:
     @staticmethod
     def download(regexp):
         def decorator(cls):
-            @functools.wraps(cls)
-            def wrapper(*args, **kw):
-                return cls(*args, **kw)
-            wrapper.VALID_URL_BASE = regexp
-            Plugin.download_plugins.append(wrapper)
-            return wrapper
+            cls.VALID_URL_BASE = regexp
+            Plugin.download_plugins.append(cls)
+            return cls
         return decorator
 
     @staticmethod
     def upload(platform):
         def decorator(cls):
-            Plugin.upload_plugins[platform] = cls
-            return cls
+            @functools.wraps(cls)
+            def wrapper(*args, **kw):
+                print(f"args {args}")
+                print(f"kw {kw}")
+                return cls(*args, **kw)
+            Plugin.upload_plugins[platform] = wrapper
+            return wrapper
         return decorator
 
     @classmethod

biliup/engine/download.py

@@ -68,12 +68,6 @@ class DownloadBase(ABC):
         # is_check 是否是检测模式 检测模式可以忽略只有下载时需要的耗时操作
         raise NotImplementedError()
 
-    @staticmethod
-    def batch_check(check_urls: List[str]) -> Generator[str, None, None]:
-        # 批量检测直播或下载状态
-        # 返回的是url_list
-        raise NotImplementedError()
-
     def download(self):
         self.danmaku_download_start(self.gen_download_filename())
         if self.is_download:
@@ -493,4 +487,14 @@ def get_valid_filename(name):
     s = re.sub(r"(?u)[^-\w.%{}\[\]【】「」（）\s]", "", str(name))
     if s in {"", ".", ".."}:
         raise RuntimeError("Could not derive file name from '%s'" % name)
-    return s
+    return s
+
+
+class BatchCheck(ABC):
+    @staticmethod
+    @abstractmethod
+    def batch_check(check_urls: List[str]) -> Generator[str, None, None]:
+        """
+        批量检测直播或下载状态
+        返回的是url_list
+        """

biliup/handler.py

@@ -29,9 +29,8 @@ logger = logging.getLogger('biliup')
 @event_manager.register(CHECK, block='Asynchronous3')
 def singleton_check(platform, name, url):
     if name is None and url is None:
-        # 如果支持批量检测，目前只有一个支持，第一版先写死按照特例处理
-        from .plugins.twitch import Twitch
-        for turl in Twitch.batch_check.__func__(Twitch.url_list):
+        # 如果支持批量检测
+        for turl in platform.batch_check(platform.url_list):
             context['url_upload_count'].setdefault(turl, 0)
             for k, v in config['streamers'].items():
                 if v.get("url", "") == turl:

biliup/plugins/twitch.py

@@ -11,7 +11,7 @@ import yt_dlp
 
 from . import logger
 from ..engine.decorators import Plugin
-from ..engine.download import DownloadBase
+from ..engine.download import DownloadBase, BatchCheck
 from biliup.config import config
 from biliup.plugins.Danmaku import DanmakuClient
 
@@ -66,7 +66,7 @@ class TwitchVideos(DownloadBase):
 
 
 @Plugin.download(regexp=VALID_URL_BASE)
-class Twitch(DownloadBase):
+class Twitch(DownloadBase, BatchCheck):
     def __init__(self, fname, url, suffix='flv'):
         DownloadBase.__init__(self, fname, url, suffix=suffix)
         self.twitch_danmaku = config.get('twitch_danmaku', False)