From a2e42d255f0d07fcdc44f7056592bfb3e8f5b37e Mon Sep 17 00:00:00 2001 From: ForgQi <34411314+ForgQi@users.noreply.github.com> Date: Mon, 16 Nov 2020 21:32:00 +0800 Subject: [PATCH] Fix upload --- common/timer.py | 12 ++--- engine/plugins/__init__.py | 57 +++++--------------- engine/plugins/base_adapter.py | 90 +++++++++++++++++++++++++++---- engine/plugins/upload/bili_web.py | 13 +++-- 4 files changed, 106 insertions(+), 66 deletions(-) diff --git a/common/timer.py b/common/timer.py index d58343a..cbfbc28 100644 --- a/common/timer.py +++ b/common/timer.py @@ -2,7 +2,7 @@ import threading class Timer(threading.Thread): - def __init__(self, func, args=(), kwargs=None, interval=15): + def __init__(self, func=None, args=(), kwargs=None, interval=15): threading.Thread.__init__(self, daemon=True) if kwargs is None: kwargs = {} @@ -12,13 +12,13 @@ class Timer(threading.Thread): self._func = func self.interval = interval - def __timer(self): - while not self._flag.is_set(): - self._func(*self._args, **self._kwargs) - self._flag.wait(self.interval) + def timer(self): + self._func(*self._args, **self._kwargs) def run(self): - self.__timer() + while not self._flag.is_set(): + self.timer() + self._flag.wait(self.interval) def stop(self): self._flag.set() diff --git a/engine/plugins/__init__.py b/engine/plugins/__init__.py index 9a1e6fe..af37285 100644 --- a/engine/plugins/__init__.py +++ b/engine/plugins/__init__.py @@ -28,44 +28,20 @@ class BatchCheckBase: pass -class Monitoring(Timer): - def __init__(self, parent_pid, file_name): - super().__init__(func=self.kill_child_processes, interval=20) - self.parent = self.children = self.numc = None - self.parent_pid = parent_pid +class Companion(Timer): + def __init__(self, pid, file_name, size=2.5): + super().__init__(interval=20) + self._pid = pid + self.proc = psutil.Process(self._pid) self.file_name = file_name + '.part' + self._size = size self.last_file_size = 0.0 - self.flag = Event() - - def terminate(self): - if self.numc == 0: - logger.error("ChildrenProcess doesn't exist") - else: - for process in self.children: - process.terminate() - # logger.info('下载卡死' + self.file_name) - - def get_process(self, parent_pid): - try: - parent = psutil.Process(parent_pid) - except psutil.NoSuchProcess: - self.stop() - return logger.error("Process doesn't exist") - children = parent.children(recursive=True) - numc = len(children) - return parent, children, numc def kill_child_processes(self): - if self.flag.is_set(): - self.stop() - return file_size = os.path.getsize(self.file_name) / 1024 / 1024 / 1024 if file_size <= self.last_file_size: logger.error('下载卡死' + self.file_name) - if self.numc == 0: - self.parent.terminate() - else: - self.terminate() + self.proc.terminate() time.sleep(1) if os.path.isfile(self.file_name): return logger.info('卡死下载进程可能未成功退出') @@ -73,32 +49,23 @@ class Monitoring(Timer): self.stop() return logger.info('卡死下载进程成功退出') self.last_file_size = file_size - if file_size >= 2.5: - self.flag.set() - self.terminate() + if file_size >= self._size: + self.proc.terminate() logger.info('分段下载' + self.file_name) - def __timer(self): - logger.debug('获取到{0}，{1}'.format(self.parent_pid, self.file_name)) + def run(self): retry = 0 while not self._flag.wait(self.interval): - self.parent, self.children, self.numc = self.get_process(self.parent_pid) if os.path.isfile(self.file_name): - self._func(*self._args, **self._kwargs) + self.kill_child_processes() else: logger.info('%s不存在' % self.file_name) if retry >= 2: - self.terminate() + self.proc.terminate() return logger.info('结束进程，找不到%s' % self.file_name) retry += 1 # logger.info('监控<%s>线程退出' % self.file_name) - def run(self): - try: - self.__timer() - finally: - logger.debug('退出监控<%s>线程' % self.file_name) - def match1(text, *patterns): if len(patterns) == 1: diff --git a/engine/plugins/base_adapter.py b/engine/plugins/base_adapter.py index 1cae2eb..34d6267 100644 --- a/engine/plugins/base_adapter.py +++ b/engine/plugins/base_adapter.py @@ -6,8 +6,8 @@ import time import streamlink import youtube_dl -from engine.plugins import logger, Monitoring - +from engine.plugins import logger, Companion +from threading import Event fake_headers = { 'Accept': 'text/html,application/xhtml+xml,application/xml;q=0.9,*/*;q=0.8', 'Accept-Encoding': 'gzip, deflate', @@ -17,13 +17,10 @@ fake_headers = { class DownloadBase: - url_list = None - def __init__(self, fname, url, suffix=None): self.fname = fname self.url = url self.suffix = suffix - self.flag = None def check_stream(self): logger.debug(self.fname) @@ -45,15 +42,10 @@ class DownloadBase: file_name = self.file_name if self.check_stream(): file_name += "." + self.suffix - pid = os.getpid() - monitor = Monitoring(pid, file_name) - self.flag = monitor.flag - monitor.start() retval = self.download(file_name) self.rename(file_name) - monitor.stop() if retval != 0: - logger.debug('准备递归下载') + logger.debug('继续下载') self.start() else: logger.info('下载完成' + self.fname) @@ -122,6 +114,7 @@ class SDownload(DownloadBase): def __init__(self, fname, url, suffix='mp4'): super().__init__(fname, url, suffix) self.stream = None + self.flag = Event() def check_stream(self): logger.debug(self.fname) @@ -151,6 +144,77 @@ class SDownload(DownloadBase): except OSError: self.rename(filename) raise +# class Monitoring(Timer): +# def __init__(self, parent_pid, file_name): +# super().__init__(func=self.kill_child_processes, interval=20) +# self.parent = self.children = self.numc = None +# self.parent_pid = parent_pid +# self.file_name = file_name + '.part' +# self.last_file_size = 0.0 +# self.flag = Event() +# +# def terminate(self): +# if self.numc == 0: +# logger.error("ChildrenProcess doesn't exist") +# else: +# for process in self.children: +# process.terminate() +# # logger.info('下载卡死' + self.file_name) +# +# def get_process(self, parent_pid): +# try: +# parent = psutil.Process(parent_pid) +# except psutil.NoSuchProcess: +# self.stop() +# return logger.error("Process doesn't exist") +# children = parent.children(recursive=True) +# numc = len(children) +# logger.info(f'{parent}, {children}') +# return parent, children, numc +# +# def kill_child_processes(self): +# if self.flag.is_set(): +# self.stop() +# return +# file_size = os.path.getsize(self.file_name) / 1024 / 1024 / 1024 +# if file_size <= self.last_file_size: +# logger.error('下载卡死' + self.file_name) +# if self.numc == 0: +# self.parent.terminate() +# else: +# self.terminate() +# time.sleep(1) +# if os.path.isfile(self.file_name): +# return logger.info('卡死下载进程可能未成功退出') +# else: +# self.stop() +# return logger.info('卡死下载进程成功退出') +# self.last_file_size = file_size +# if file_size >= 2.5: +# self.flag.set() +# self.terminate() +# logger.info('分段下载' + self.file_name) +# +# def __timer(self): +# logger.debug('获取到{0}，{1}'.format(self.parent_pid, self.file_name)) +# retry = 0 +# while not self._flag.wait(self.interval): +# self.parent, self.children, self.numc = self.get_process(self.parent_pid) +# if os.path.isfile(self.file_name): +# self._func(*self._args, **self._kwargs) +# else: +# logger.info('%s不存在' % self.file_name) +# if retry >= 2: +# self.terminate() +# return logger.info('结束进程，找不到%s' % self.file_name) +# retry += 1 +# # logger.info('监控<%s>线程退出' % self.file_name) +# +# def run(self): +# try: +# self.__timer() +# finally: +# logger.debug('退出监控<%s>线程' % self.file_name) # ffmpeg.exe -i http://vfile1.grtn.cn/2018/1542/0254/3368/154202543368.ssm/154202543368.m3u8 @@ -166,10 +230,14 @@ class FFmpegdl(DownloadBase): '-y', '-i', self.raw_stream_url, '-bsf:a', 'aac_adtstoasc', *self.opt_args, '-c', 'copy', '-f', self.suffix, filename + '.part'] proc = subprocess.Popen(args, stdin=subprocess.PIPE) + monitor = Companion(proc.pid, filename, size=2.5) + monitor.start() try: retval = proc.wait() except KeyboardInterrupt: if sys.platform != 'win32': proc.communicate(b'q') raise + finally: + monitor.stop() return retval diff --git a/engine/plugins/upload/bili_web.py b/engine/plugins/upload/bili_web.py index b876e4b..7d57e0b 100644 --- a/engine/plugins/upload/bili_web.py +++ b/engine/plugins/upload/bili_web.py @@ -262,9 +262,13 @@ class BiliBili: cost = time.perf_counter() - start logger.info(f'{name} uploaded >> {total / 1000 / 1000 / cost:.2f}MB/s') + logger.info(parts) + parts.sort(key=lambda x: x['index']) + logger.info(parts) self.__session.post(f"{endpoint}/mkfile/{total}/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() + logger.info(f'{name}: {r}') if r["OK"] != 1: raise Exception(r) return {"title": splitext(name)[0], "filename": bili_filename, "desc": ""} @@ -310,11 +314,12 @@ class BiliBili: p = { 'name': name, 'uploadId': upload_id, - 'biz_id': biz_id + 'biz_id': biz_id, + 'output': 'json', + 'profile': 'ugcupos/bup' } - r = self.__session.post( - f'{url}?output=json&profile=ugcupos%2Fbup', params=p, - json={"parts": parts}, headers=headers, timeout=15).json() + r = self.__session.post(url, params=p, json={"parts": parts}, headers=headers, timeout=15).json() + logger.info(f'{name}: {r}') if r["OK"] != 1: raise Exception(r) return {"title": splitext(name)[0], "filename": splitext(basename(upos_uri))[0], "desc": ""}