diff --git a/Bilibili.py b/Bilibili.py index 26ad3b7..e16e1b7 100644 --- a/Bilibili.py +++ b/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) diff --git a/common/Daemon.py b/common/Daemon.py index e4c3b57..3fcecd9 100644 --- a/common/Daemon.py +++ b/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()) diff --git a/common/decorators.py b/common/decorators.py index 3deb93e..19800a3 100644 --- a/common/decorators.py +++ b/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 diff --git a/common/event.py b/common/event.py index 2fe9029..6596cf2 100644 --- a/common/event.py +++ b/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() diff --git a/config(demo).yaml b/config(demo).yaml index e5d851e..6d72235 100644 --- a/config(demo).yaml +++ b/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吕布卫星人族天梯第一视角: diff --git a/engine/__init__.py b/engine/__init__.py index 64d6625..fd502b1 100644 --- a/engine/__init__.py +++ b/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'] diff --git a/engine/downloader.py b/engine/downloader.py index 1c18080..58d8292 100644 --- a/engine/downloader.py +++ b/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): diff --git a/engine/handler.py b/engine/handler.py index 331eb40..cdfbe12 100644 --- a/engine/handler.py +++ b/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())) diff --git a/engine/plugins/base_adapter.py b/engine/plugins/base_adapter.py index 36b76b1..a84e330 100644 --- a/engine/plugins/base_adapter.py +++ b/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: diff --git a/engine/plugins/upload/bili_chrome.py b/engine/plugins/upload/bili_chrome.py index 48c7898..c715474 100644 --- a/engine/plugins/upload/bili_chrome.py +++ b/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") diff --git a/engine/plugins/upload/bili_web.py b/engine/plugins/upload/bili_web.py index 49f0ab6..9015212 100644 --- a/engine/plugins/upload/bili_web.py +++ b/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: