Update Bilibili.py

Bilibili.py

@@ -3,7 +3,6 @@
 import asyncio
 import sys
 import common
-from engine.handler import event_manager
 from engine import main
 from common.Daemon import Daemon
 
@@ -24,7 +23,7 @@ if __name__ == '__main__':
             sys.exit(2)
         sys.exit(0)
     elif len(sys.argv) == 1:
-        asyncio.run(main(event_manager))
+        asyncio.run(main())
     else:
         print('usage: %s start|stop|restart' % sys.argv[0])
         sys.exit(2)

common/Daemon.py

@@ -6,7 +6,6 @@ import os
 import time
 import atexit
 from engine import main
-from engine.handler import event_manager
 
 logger = logging.getLogger('log01')
 
@@ -123,4 +122,4 @@ class Daemon(object):
     @staticmethod
     def _run():
         """ run your fun"""
-        asyncio.run(main(event_manager))
+        asyncio.run(main())

common/decorators.py

@@ -1,10 +1,25 @@
 import functools
+import importlib
+import pkgutil
+import re
+
+
+def suit_url(pattern, urls):
+    sorted_url = []
+    for i in range(len(urls) - 1, -1, -1):
+        if re.match(pattern, urls[i]):
+            sorted_url.append(urls[i])
+            urls.remove(urls[i])
+    return sorted_url
 
 
 class Plugin:
     download_plugins = []
     upload_plugins = {}
 
+    def __init__(self, pkg):
+        self.load_plugins(pkg)
+
     @staticmethod
     def download(regexp):
         def decorator(cls):
@@ -22,3 +37,42 @@ class Plugin:
             Plugin.upload_plugins[platform] = cls
             return cls
         return decorator
+
+    @classmethod
+    def sorted_checker(cls, urls):
+        from engine.plugins import general
+        curls = urls.copy()
+        checker_plugins = {}
+        for plugin in cls.download_plugins:
+            url_list = suit_url(plugin.VALID_URL_BASE, curls)
+            if not url_list:
+                continue
+            elif hasattr(plugin, "BatchCheck"):
+                checker_plugins[plugin.__name__] = plugin.BatchCheck(url_list)
+            else:
+                plugin.url_list = url_list
+                checker_plugins[plugin.__name__] = plugin
+            if not curls:
+                return checker_plugins
+        general.__plugin__.url_list = curls
+        checker_plugins[general.__plugin__.__name__] = general.__plugin__
+        return checker_plugins
+
+    def load_plugins(self, pkg):
+        """Attempt to load plugins from the path specified.
+        engine.plugins.__path__[0]: full path to a directory where to look for plugins
+        """
+
+        plugins = []
+
+        for loader, name, ispkg in pkgutil.iter_modules([pkg.__path__[0]]):
+            # set the full plugin module name
+            module_name = f"{pkg.__name__}.{name}"
+            module = importlib.import_module(module_name)
+            if ispkg:
+                self.load_plugins(module)
+                continue
+            if module in plugins:
+                continue
+            plugins.append(module)
+        return plugins

common/event.py

@@ -10,10 +10,12 @@ import functools
 
 
 class EventManager(Thread):
-    def __init__(self, kwargs):
+    def __init__(self, context=None):
         """初始化事件管理器"""
         super().__init__(name='Synchronous')
-        self.__kwargs = kwargs
+        if context is None:
+            context = {}
+        self.context = context
         # 事件对象列表
         self.__eventQueue = Queue()
         # 事件管理器开关
@@ -143,13 +145,14 @@ class EventManager(Thread):
             sig = inspect.signature(cls)
             kwargs = {}
             for k in sig.parameters:
-                kwargs[k] = self.__kwargs[k]
+                kwargs[k] = self.context[k]
             instance = cls(**kwargs)
             for type_ in self.__method:
                 for handler in self.__method[type_]:
                     self.add_event_listener(type_, getattr(instance, handler))
             self.__method.clear()
             return cls
+
         return decorator
 
 
@@ -158,4 +161,7 @@ class Event:
     """事件对象"""
     type_: str  # 事件类型
     args: tuple = ()
-    dict: dict = field(default_factory=dict) # 字典用于保存具体的事件数据
+    dict: dict = field(default_factory=dict)  # 字典用于保存具体的事件数据
+
+
+event_manager = EventManager()

config(demo).yaml

@@ -7,7 +7,14 @@ user: # 在填了cookies的情况下优先使用cookies上传，如需使用用
         username: your usrname
         password: your password
 # 默认通过网页接口上传,可选通过操作chrome上传,此时需要填写chromedriver路径
-chromedriver_path: /usr/local/bin/chromedriver
+#chromedriver_path: /usr/local/bin/chromedriver
+
+# b站上传线路选择，默认为自动模式，目前可手动切换为bda2, kodo, ws, qn。
+lines: AUTO
+# 单文件并发上传数，未达到带宽上限时增大此值可提高上传速度
+threads: 3
+# 录像单文件大小限制，单位Byte，超过此大小分段下载
+file_size: 2621440000
 
 streamers:
     星际2INnoVation吕布卫星人族天梯第一视角:

engine/__init__.py

@@ -2,17 +2,14 @@ import asyncio
 
 import yaml
 
-from common.event import Event
+from common.decorators import Plugin
+from common.event import Event, event_manager
 from common.reload import AutoReload
 from common.timer import Timer
-from engine.downloader import sorted_checker
 
 with open(r'config.yaml', encoding='utf-8') as stream:
     config = yaml.load(stream, Loader=yaml.FullLoader)
 
-streamers = config['streamers']
-chromedriver_path = config.get('chromedriver_path')
-
 
 def invert_dict(d: dict):
     inverse_dict = {}
@@ -22,26 +19,30 @@ def invert_dict(d: dict):
     return inverse_dict
 
 
-streamer_url = {k: v['url'] for k, v in streamers.items()}
-inverted_index = invert_dict(streamer_url)
-urls = list(inverted_index.keys())
-url_status = dict.fromkeys(inverted_index, 0)
-checker = sorted_checker(urls)
-platforms = checker.keys()
-
-context = {**config, "urls": urls, "url_status": url_status}
-
-
-async def check_timer(event_manager):
-    event_manager.send_event(Event(CHECK_UPLOAD))
-    for k in platforms:
-        event_manager.send_event(Event(CHECK, (k,)))
-
-
-async def main(event_manager):
+async def main():
+    import engine.plugins
+    streamers = config['streamers']
+    streamer_url = {k: v['url'] for k, v in streamers.items()}
+    inverted_index = invert_dict(streamer_url)
+    urls = list(inverted_index.keys())
+    url_status = dict.fromkeys(inverted_index, 0)
+    checker = Plugin(engine.plugins).sorted_checker(urls)
+    # 初始化事件管理器
+    context = {**config, 'urls': urls, 'url_status': url_status,
+               'checker': checker, 'inverted_index': inverted_index, 'streamer_url': streamer_url}
+    event_manager.context = context
+    import engine.handler
     event_manager.start()
+
+    platforms = checker.keys()
+
+    def check_timer():
+        event_manager.send_event(Event(CHECK_UPLOAD))
+        for k in platforms:
+            event_manager.send_event(Event(CHECK, (k,)))
+
     # 初始化定时器
-    timer = Timer(func=check_timer, args=(event_manager,), interval=40)
+    timer = Timer(func=check_timer, interval=40)
 
     # 模块更新自动重启
     detector = AutoReload(event_manager, timer, interval=15)
@@ -55,5 +56,4 @@ DOWNLOAD = 'download'
 BE_MODIFIED = 'be_modified'
 UPLOAD = 'upload'
 __all__ = ['downloader', 'uploader', 'plugins', 'main',
-           'context', 'inverted_index', 'checker',
            'CHECK', 'BE_MODIFIED', 'DOWNLOAD', 'TO_MODIFY', 'UPLOAD', 'CHECK_UPLOAD']

engine/downloader.py

@@ -1,71 +1,14 @@
-import importlib
 import logging
-import pkgutil
 import re
 import time
 from urllib.error import HTTPError
 
-import engine.plugins
 from common.decorators import Plugin
 from engine.plugins import general, BatchCheckBase
 
 logger = logging.getLogger('log01')
 
 
-def load_plugins(pkg=engine.plugins):
-    """Attempt to load plugins from the path specified.
-    engine.plugins.__path__[0]: full path to a directory where to look for plugins
-    """
-
-    plugins = []
-
-    for loader, name, ispkg in pkgutil.iter_modules([pkg.__path__[0]]):
-        # set the full plugin module name
-        module_name = f"{pkg.__name__}.{name}"
-        module = importlib.import_module(module_name)
-        if ispkg:
-            load_plugins(module)
-            continue
-        if module in plugins:
-            continue
-        plugins.append(module)
-        # self.load_plugin(module_name)
-    # print(self.plugins)
-    return plugins
-
-
-load_plugins()
-
-
-def suit_url(pattern, urls):
-    sorted_url = []
-    for i in range(len(urls) - 1, -1, -1):
-        if re.match(pattern, urls[i]):
-            sorted_url.append(urls[i])
-            urls.remove(urls[i])
-    return sorted_url
-
-
-def sorted_checker(urls):
-    curls = urls.copy()
-    checker_plugins = {}
-    for plugin in Plugin.download_plugins:
-        url_list = suit_url(plugin.VALID_URL_BASE, curls)
-        if not url_list:
-            continue
-        elif hasattr(plugin, "BatchCheck"):
-            checker_plugins[plugin.__name__] = plugin.BatchCheck(url_list)
-        else:
-            plugin.url_list = url_list
-            checker_plugins[plugin.__name__] = plugin
-        if not curls:
-            return checker_plugins
-    general.__plugin__.url_list = curls
-    checker_plugins[general.__plugin__.__name__] = general.__plugin__
-    # onebyone.append(__import__('engine.plugins.general', fromlist=['general',]))
-    return checker_plugins
-
-
 def download(fname, url):
     for plugin in Plugin.download_plugins:
         if re.match(plugin.VALID_URL_BASE, url):

engine/handler.py

@@ -1,23 +1,16 @@
 import common
-import engine
 from common import logger
-from common.event import Event
+from common.event import Event, event_manager
 from engine import *
 from engine.downloader import download, check_url
 from engine.plugins.upload import UploadBase
 from engine.uploader import upload
 
-# 初始化事件管理器
-event_manager = common.event.EventManager(context)
-
 
 @event_manager.register(DOWNLOAD, block=True)
 def process(name, url):
     date = common.time_now()
     try:
-        # p = multiprocessing.Process(target=download, args=(name, url))
-        # p.start()
-        # p.join()
         download(name, url)
     finally:
         return Event(UPLOAD, (name, url, date))
@@ -35,14 +28,17 @@ def process_upload(name, url, date):
 
 @event_manager.server()
 class KernelFunc:
-    def __init__(self, urls, url_status: dict):
+    def __init__(self, urls, url_status: dict, checker, inverted_index, streamer_url):
         self.urls = urls
         self.url_status = url_status
         self.__raw_streamer_status = url_status.copy()
+        self.checker = checker
+        self.inverted_index = inverted_index
+        self.streamer_url = streamer_url
 
     @event_manager.register(CHECK, block=True)
     def singleton_check(self, platform):
-        plugin = checker[platform]
+        plugin = self.checker[platform]
         for url in check_url(plugin):
             yield Event(TO_MODIFY, args=(url,))
 
@@ -54,7 +50,7 @@ class KernelFunc:
             return logger.debug('已开播正在下载')
         if self.url_status[url] == 2:
             return logger.debug('正在上传稍后下载')
-        name = inverted_index[url]
+        name = self.inverted_index[url]
         logger.debug(f'{name}刚刚开播，去下载')
         self.url_status[url] = 1
         return Event(DOWNLOAD, args=(name, url))
@@ -65,7 +61,7 @@ class KernelFunc:
 
     @event_manager.register(CHECK_UPLOAD)
     def free_upload(self):
-        for title, urls in engine.streamer_url.items():
+        for title, urls in self.streamer_url.items():
             if self.free(urls) and UploadBase.filter_file(title):
                 yield Event(UPLOAD, args=(title, urls[0], common.time_now()))
 

engine/plugins/base_adapter.py

@@ -7,6 +7,7 @@ from threading import Event
 import streamlink
 import youtube_dl
 
+import engine
 from engine.plugins import logger
 
 fake_headers = {
@@ -35,7 +36,7 @@ class DownloadBase:
             return False
         file_name = f'{self.file_name}.{self.suffix}'
         retval = self.download(file_name)
-        logger.info(f'完成{retval}: {file_name}')
+        logger.info(f'{retval}part: {file_name}')
         self.rename(file_name)
         return retval
 
@@ -43,12 +44,12 @@ class DownloadBase:
         i = 0
         try:
             logger.info('开始下载%s：%s' % (self.__class__.__name__, self.fname))
-            while i < 20:
+            while i < 30:
                 ret = self.run()
                 if ret is False:
                     return
                 elif ret == 1:
-                    time.sleep(15)
+                    time.sleep(45)
                 i += 1
         except:
             logger.exception("Uncaught exception:")
@@ -88,7 +89,6 @@ class YDownload(DownloadBase):
     def get_sinfo(self):
         info_list = []
         with youtube_dl.YoutubeDL() as ydl:
-            # cu = self.url.get(self.__class__.__name__)
             if self.url:
                 info = ydl.extract_info(self.url, download=False)
             else:
@@ -109,7 +109,6 @@ class YDownload(DownloadBase):
 
     def dl(self):
         with youtube_dl.YoutubeDL(self.ydl_opts) as ydl:
-            # ydl.download([self.url[self.__class__.__name__]])
             ydl.download([self.url])
 
 
@@ -127,7 +126,7 @@ class SDownload(DownloadBase):
                 self.stream = streams["best"]
                 fd = self.stream.open()
                 fd.close()
-                streams.close()
+                # streams.close()
                 return True
         except streamlink.StreamlinkError:
             return
@@ -156,14 +155,16 @@ class FFmpegdl(DownloadBase):
         super().__init__(fname, url, suffix)
         self.raw_stream_url = None
         self.opt_args = []
-        self.default_output_args = ['-bsf:a', 'aac_adtstoasc', '-fs', '2621440000']
+        self.default_output_args = [
+            '-bsf:a', 'aac_adtstoasc',
+            '-fs', f"{engine.config.get('file_size') if engine.config.get('file_size') else '2621440000'}"
+        ]
         self.default_input_args = ['-headers', ''.join('%s: %s\r\n' % x for x in fake_headers.items()),
-                                   '-reconnect', '1', '-reconnect_streamed', '1',
-                                   '-reconnect_delay_max', '15', '-rw_timeout', '15000000']
+                                   '-reconnect_streamed', '1', '-reconnect_delay_max', '20', '-rw_timeout', '20000000']
 
     def download(self, filename):
-        args = ['ffmpeg', *self.default_input_args,
-                '-y', '-i', self.raw_stream_url, *self.default_output_args, *self.opt_args,
+        args = ['ffmpeg', '-y', *self.default_input_args,
+                '-i', self.raw_stream_url, *self.default_output_args, *self.opt_args,
                 '-c', 'copy', '-f', self.suffix, f'{filename}.part']
         proc = subprocess.Popen(args, stdin=subprocess.PIPE)
         try:

engine/plugins/upload/bili_chrome.py

@@ -54,7 +54,7 @@ class BiliChrome(UploadBase):
         options = webdriver.ChromeOptions()
 
         options.add_argument('headless')
-        self.driver = webdriver.Chrome(executable_path=engine.chromedriver_path, chrome_options=options)
+        self.driver = webdriver.Chrome(executable_path=engine.config.get('chromedriver_path'), chrome_options=options)
         # service_log_path=service_log_path)
         try:
             self.driver.get("https://www.bilibili.com")

engine/plugins/upload/bili_web.py

@@ -28,13 +28,17 @@ from engine.plugins.upload import UploadBase, logger
 class BiliWeb(UploadBase):
     def __init__(self, principal, data):
         super().__init__(principal, data, persistence_path='engine/bili.cookie')
-        # cookie = data['cookie']
-        # self.__data: Upload.Data = data['config']
 
     def upload(self, file_list):
         video = Data()
+        lines = engine.config.get('lines')
+        threads = engine.config.get('threads')
         with BiliBili(video) as bili:
             bili.login(self.persistence_path)
+            if lines:
+                bili.lines = lines
+            if threads:
+                bili.tasks = threads
             for file in file_list:
                 video_part = bili.upload_file(file)  # 上传视频
                 video.videos.append(video_part)  # 添加已经上传的视频
@@ -60,6 +64,8 @@ class BiliWeb(UploadBase):
 class BiliBili:
     def __init__(self, video: 'Data'):
         self.app_key = 'bca7e84c2d947ac6'
+        self.lines = 'AUTO'
+        self.tasks = 3
         self.__session = requests.Session()
         self.video = video
         self.__session.mount('https://', HTTPAdapter(max_retries=Retry(total=5, method_whitelist=False)))
@@ -195,29 +201,30 @@ class BiliBili:
     def upload_file(self, filepath: str):
         """上传本地视频文件,返回视频信息dict
         b站目前支持4种上传线路upos, kodo, gcs, bos
-        kodo: {"os":"kodo","query": "bucket=bvcupcdnkodobm&probe_version=20200810",
-        "probe_url":"//up-na0.qbox.me/crossdomain.xml"}
         gcs: {"os":"gcs","query":"bucket=bvcupcdngcsus&probe_version=20200810",
         "probe_url":"//storage.googleapis.com/bvcupcdngcsus/OK"},
         bos: {"os":"bos","query":"bucket=bvcupcdnboshb&probe_version=20200810",
         "probe_url":"??"}
-        upos:
-        {"os":"upos","query":"upcdn=ws&probe_version=20200810","probe_url":"//upos-sz-upcdnws.bilivideo.com/OK"}
-        {"os":"upos","query":"upcdn=qn&probe_version=20200810","probe_url":"//upos-sz-upcdnqn.bilivideo.com/OK"}
-        {"os":"upos","query":"upcdn=bda2&probe_version=20200810","probe_url":"//upos-sz-upcdnbda2.bilivideo.com/OK"}
         """
         if not self._auto_os:
             self._auto_os = self.probe()
-            # self._auto_os = {"os": "kodo", "query": "bucket=bvcupcdnkodobm&probe_version=20200810",
-            #                  "probe_url": "//up-na0.qbox.me/crossdomain.xml"}
-            logger.info(f"自动线路选择{self._auto_os['os']}: {self._auto_os['query']}. time: {self._auto_os.get('cost')}")
-        profile = 'ugcupos/bup' if 'upos' == self._auto_os['os'] else "ugcupos/bupfetch"
-        query = f"r={self._auto_os['os']}&profile={quote(profile, safe='')}" \
-                f"&ssl=0&version=2.8.9&build=2080900&{self._auto_os['query']}"
+            if self.lines == 'kodo':
+                self._auto_os = {"os": "kodo", "query": "bucket=bvcupcdnkodobm&probe_version=20200810",
+                                 "probe_url": "//up-na0.qbox.me/crossdomain.xml"}
+            if self.lines == 'bda2':
+                self._auto_os = {"os": "upos", "query": "upcdn=bda2&probe_version=20200810",
+                                 "probe_url": "//upos-sz-upcdnbda2.bilivideo.com/OK"}
+            if self.lines == 'ws':
+                self._auto_os = {"os": "upos", "query": "upcdn=ws&probe_version=20200810",
+                                 "probe_url": "//upos-sz-upcdnws.bilivideo.com/OK"}
+            if self.lines == 'qn':
+                self._auto_os = {"os": "upos", "query": "upcdn=qn&probe_version=20200810",
+                                 "probe_url": "//upos-sz-upcdnqn.bilivideo.com/OK"}
+            logger.info(f"线路选择{self._auto_os['os']}: {self._auto_os['query']}. time: {self._auto_os.get('cost')}")
         if self._auto_os['os'] == 'upos':
-            return self.upos(filepath, query)
+            upload = self.upos
         elif self._auto_os['os'] == 'kodo':
-            return self.kodo(filepath, query)
+            upload = self.kodo
         elif self._auto_os['os'] == "gcs":
             raise NotImplementedError('gcs')
         elif self._auto_os['os'] == "bos":
@@ -225,103 +232,107 @@ class BiliBili:
         else:
             logger.error(f"NoSearch:{self._auto_os['os']}")
             raise NotImplementedError(self._auto_os['os'])
-
-    def kodo(self, filepath, query):
-        total = os.path.getsize(filepath)
-        chunk_size = 4194304
+        total_size = os.path.getsize(filepath)
         with open(filepath, 'rb') as f:
-            name = f.name
-            ret = self.__session.get(
-                f'https://member.bilibili.com/preupload?name={quote(name)}&size={total}&{query}', timeout=5).json()
-            bili_filename = ret['bili_filename']
-            key = ret['key']
-            endpoint = f"https:{ret['endpoint']}"
-            token = ret['uptoken']
-            fetch_url = ret['fetch_url']
-            fetch_headers = ret['fetch_headers']
-            url = f'{endpoint}/mkblk'
-            headers = {
-                'Authorization': f"UpToken {token}",
+            query = {
+                'r': self._auto_os['os'],
+                'profile': 'ugcupos/bup' if 'upos' == self._auto_os['os'] else "ugcupos/bupfetch",
+                'ssl': 0,
+                'version': '2.8.12',
+                'build': 2081200,
+                'name': f.name,
+                'size': total_size,
             }
-            # 开始上传
-            parts = []  # 分块信息
-            chunks = math.ceil(total / chunk_size)  # 获取分块数量
+            ret = self.__session.get(
+                f"https://member.bilibili.com/preupload?{self._auto_os['query']}", params=query,
+                timeout=5)
+            return asyncio.run(upload(f, total_size, ret.json()))
 
-            async def upload_chunk(session, chunks_data, params, index):
-                async with session.post(f'{url}/{len(chunks_data)}',
-                                        data=chunks_data, headers=headers) as response:
-                    end = time.perf_counter() - start
-                    ctx = await response.json()
-                    parts.append({"index": index, "ctx": ctx['ctx']})
-                    sys.stdout.write(f"\r{params['end'] / 1000 / 1000 / end:.2f}MB/s "
-                                     f"=> {params['partNumber'] / chunks:.1%}")
+    async def kodo(self, file, total_size, ret, chunk_size=4194304):
+        filename = file.name
+        bili_filename = ret['bili_filename']
+        key = ret['key']
+        endpoint = f"https:{ret['endpoint']}"
+        token = ret['uptoken']
+        fetch_url = ret['fetch_url']
+        fetch_headers = ret['fetch_headers']
+        url = f'{endpoint}/mkblk'
+        headers = {
+            'Authorization': f"UpToken {token}",
+        }
+        # 开始上传
+        parts = []  # 分块信息
+        chunks = math.ceil(total_size / chunk_size)  # 获取分块数量
 
-            start = time.perf_counter()
-            asyncio.run(self._upload({}, f, chunk_size, upload_chunk))
-            cost = time.perf_counter() - start
+        async def upload_chunk(session, chunks_data, params):
+            async with session.post(f'{url}/{len(chunks_data)}',
+                                    data=chunks_data, headers=headers) as response:
+                end = time.perf_counter() - start
+                ctx = await response.json()
+                parts.append({"index": params['chunk'], "ctx": ctx['ctx']})
+                sys.stdout.write(f"\r{params['end'] / 1000 / 1000 / end:.2f}MB/s "
+                                 f"=> {params['partNumber'] / chunks:.1%}")
 
-        logger.info(f'{name} uploaded >> {total / 1000 / 1000 / cost:.2f}MB/s')
+        start = time.perf_counter()
+        await self._upload({}, file, chunk_size, upload_chunk)
+        cost = time.perf_counter() - start
+
+        logger.info(f'{filename} uploaded >> {total_size / 1000 / 1000 / cost:.2f}MB/s')
         parts.sort(key=lambda x: x['index'])
-        self.__session.post(f"{endpoint}/mkfile/{total}/key/{base64.urlsafe_b64encode(key.encode()).decode()}",
+        self.__session.post(f"{endpoint}/mkfile/{total_size}/key/{base64.urlsafe_b64encode(key.encode()).decode()}",
                             data=','.join(map(lambda x: x['ctx'], parts)), headers=headers, timeout=10)
         r = self.__session.post(f"https:{fetch_url}", headers=fetch_headers, timeout=5).json()
         if r["OK"] != 1:
             raise Exception(r)
-        return {"title": splitext(name)[0], "filename": bili_filename, "desc": ""}
+        return {"title": splitext(filename)[0], "filename": bili_filename, "desc": ""}
 
-    def upos(self, filepath, query):
-        total = os.path.getsize(filepath)
-        with open(filepath, 'rb') as f:
-            name = f.name
-            ret = self.__session.get(
-                f'https://member.bilibili.com/preupload?name={quote(name)}&size={total}&{query}', timeout=5).json()
-            chunk_size = ret['chunk_size']
-            auth = ret["auth"]
-            endpoint = ret["endpoint"]
-            biz_id = ret["biz_id"]
-            upos_uri = ret["upos_uri"]
-            url = f"https:{endpoint}/{upos_uri.replace('upos://', '')}"  # 视频上传路径
-            headers = {
-                "X-Upos-Auth": auth
-            }
-            # 向上传地址申请上传，得到上传id等信息
-            upload_id = self.__session.post(f'{url}?uploads&output=json', timeout=5,
-                                            headers=headers).json()["upload_id"]
-            # 开始上传
-            parts = []  # 分块信息
-            chunks = math.ceil(total / chunk_size)  # 获取分块数量
+    async def upos(self, file, total_size, ret):
+        filename = file.name
+        chunk_size = ret['chunk_size']
+        auth = ret["auth"]
+        endpoint = ret["endpoint"]
+        biz_id = ret["biz_id"]
+        upos_uri = ret["upos_uri"]
+        url = f"https:{endpoint}/{upos_uri.replace('upos://', '')}"  # 视频上传路径
+        headers = {
+            "X-Upos-Auth": auth
+        }
+        # 向上传地址申请上传，得到上传id等信息
+        upload_id = self.__session.post(f'{url}?uploads&output=json', timeout=5,
+                                        headers=headers).json()["upload_id"]
+        # 开始上传
+        parts = []  # 分块信息
+        chunks = math.ceil(total_size / chunk_size)  # 获取分块数量
 
-            async def upload_chunk(session, chunks_data, params, index):
-                async with session.put(url, params=params,
-                                       data=chunks_data, headers=headers):
-                    end = time.perf_counter() - start
-                    parts.append({"partNumber": index + 1, "eTag": "etag"})
-                    sys.stdout.write(f"\r{params['end'] / 1000 / 1000 / end:.2f}MB/s "
-                                     f"=> {params['partNumber'] / chunks:.1%}")
+        async def upload_chunk(session, chunks_data, params):
+            async with session.put(url, params=params, raise_for_status=True,
+                                   data=chunks_data, headers=headers):
+                end = time.perf_counter() - start
+                parts.append({"partNumber": params['chunk'] + 1, "eTag": "etag"})
+                sys.stdout.write(f"\r{params['end'] / 1000 / 1000 / end:.2f}MB/s "
+                                 f"=> {params['partNumber'] / chunks:.1%}")
 
-            start = time.perf_counter()
-            asyncio.run(self._upload({
-                'uploadId': upload_id,
-                'chunks': chunks,
-                'total': total
-            }, f, chunk_size, upload_chunk))
-            cost = time.perf_counter() - start
-        logger.info(f'{name} uploaded >> {total / 1000 / 1000 / cost:.2f}MB/s')
+        start = time.perf_counter()
+        await self._upload({
+            'uploadId': upload_id,
+            'chunks': chunks,
+            'total': total_size
+        }, file, chunk_size, upload_chunk)
+        cost = time.perf_counter() - start
         p = {
-            'name': name,
+            'name': filename,
             'uploadId': upload_id,
             'biz_id': biz_id,
             'output': 'json',
             'profile': 'ugcupos/bup'
         }
         r = self.__session.post(url, params=p, json={"parts": parts}, headers=headers, timeout=15).json()
-        logger.info(f'{name}: {r}')
-        if r["OK"] != 1:
+        logger.info(f'{filename} uploaded >> {total_size / 1000 / 1000 / cost:.2f}MB/s. {r}')
+        if r.get('OK') != 1:
             raise Exception(r)
-        return {"title": splitext(name)[0], "filename": splitext(basename(upos_uri))[0], "desc": ""}
+        return {"title": splitext(filename)[0], "filename": splitext(basename(upos_uri))[0], "desc": ""}
 
-    @staticmethod
-    async def _upload(params, file, chunk_size, afunc, tasks=3):
+    async def _upload(self, params, file, chunk_size, afunc):
         params['chunk'] = -1
 
         async def upload_chunk():
@@ -334,10 +345,16 @@ class BiliBili:
                 params['partNumber'] = params['chunk'] + 1
                 params['start'] = params['chunk'] * chunk_size
                 params['end'] = params['start'] + params['size']
-                await afunc(session, chunks_data, params, params['chunk'])
+                clone = params.copy()
+                for i in range(10):
+                    try:
+                        await afunc(session, chunks_data, clone)
+                        break
+                    except (asyncio.TimeoutError, aiohttp.ClientError) as e:
+                        logger.error(f"retry chunk{clone['chunk']} >> {i+1}. {e}")
 
         async with aiohttp.ClientSession() as session:
-            await asyncio.gather(*[upload_chunk() for _ in range(tasks)])
+            await asyncio.gather(*[upload_chunk() for _ in range(self.tasks)])
 
     def submit(self):
         if not self.video.title: