diff --git a/Bilibili.py b/Bilibili.py index 7878783..26ad3b7 100644 --- a/Bilibili.py +++ b/Bilibili.py @@ -1,5 +1,6 @@ #!/usr/bin/python3 # coding:utf8 +import asyncio import sys import common from engine.handler import event_manager @@ -23,7 +24,7 @@ if __name__ == '__main__': sys.exit(2) sys.exit(0) elif len(sys.argv) == 1: - main(event_manager) + asyncio.run(main(event_manager)) else: print('usage: %s start|stop|restart' % sys.argv[0]) sys.exit(2) diff --git a/common/Daemon.py b/common/Daemon.py index 245ba57..e4c3b57 100644 --- a/common/Daemon.py +++ b/common/Daemon.py @@ -1,3 +1,4 @@ +import asyncio import logging from signal import SIGTERM import sys @@ -122,22 +123,4 @@ class Daemon(object): @staticmethod def _run(): """ run your fun""" - main(event_manager) - - -# if __name__ == '__main__': -# daemon = Daemon('/home/chromeuser/bilibiliupload/watch_process.pid') -# if len(sys.argv) == 2: -# if 'start' == sys.argv[1]: -# daemon.start() -# elif 'stop' == sys.argv[1]: -# daemon.stop() -# elif 'restart' == sys.argv[1]: -# daemon.restart() -# else: -# print('unknown command') -# sys.exit(2) -# sys.exit(0) -# else: -# print('usage: %s start|stop|restart' % sys.argv[0]) -# sys.exit(2) + asyncio.run(main(event_manager)) diff --git a/common/reload.py b/common/reload.py index 3bef4f9..0667c2e 100644 --- a/common/reload.py +++ b/common/reload.py @@ -1,10 +1,12 @@ -from threading import Thread +import asyncio import sys import os import time import subprocess import logging -import psutil + +from common.timer import Timer + logger = logging.getLogger('log01') @@ -13,33 +15,20 @@ def has_extension(fname_list, *extension): for fname in fname_list: result = list(map(fname.endswith, extension)) if True in result: - array.append(True) - else: - array.append(False) - if True in array: - return True + return True + # array.append(True) + # else: + # array.append(False) + # if True in array: + # return True return False -def get_p_children(pid, _recursive=True): - try: - parent = psutil.Process(pid) - except psutil.NoSuchProcess: - return None - children = parent.children(recursive=_recursive) - return children - - -class Autoreload: - def __init__(self, manager, target, interval=10): - # self.p = _process # 被监控子进程 - self.__thread = Thread(target=self.start_change_detector, args=(interval,)) - self.manager = manager - self._target = target - - def __call__(self, *args, **kwargs): - self.__thread.setDaemon(True) - self.__thread.start() +class AutoReload(Timer): + def __init__(self, *watched, interval=10): + super().__init__(interval) + self.watched = watched + self.mtimes = {} @staticmethod def _iter_module_files(): @@ -53,56 +42,42 @@ class Autoreload: filename = filename[:-1] yield filename - def _is_any_file_changed(self, mtimes): + def _is_any_file_changed(self): """Return 1 if there is any source file of sys.modules changed, otherwise 0. mtimes is dict to store the last modify time for comparing.""" - for filename in self._iter_module_files(): try: mtime = os.stat(filename).st_mtime except IOError: continue - old_time = mtimes.get(filename, None) + old_time = self.mtimes.get(filename, None) if old_time is None: - mtimes[filename] = mtime + self.mtimes[filename] = mtime elif mtime > old_time: logger.info('模块已更新') - return 1 - return 0 + return True + return False @staticmethod def _work_free(): - # wp = psutil.Process(self.p.pid) - # more_children = wp.children(recursive=True) - # children = wp.children() - # if len(more_children) == len(children): - # logger.info('进程空闲') - # return True - # return False - fname_list = os.listdir('.') if has_extension(fname_list, '.mp4', '.part', '.flv'): return False logger.info('进程空闲') return True - def _restart_subp(self, interval=10): + async def atimer(self): + """Check file state ervry interval. If any change is detected, exit this + process with a special code, so that deamon will to restart a new process. + """ + if not self._is_any_file_changed(): + return while True: - time.sleep(interval) + await asyncio.sleep(self.interval) if self._work_free(): - # logger.info('重启进程') - # pid = self.p.pid - # children = get_p_children(pid) - # - # os.kill(pid, SIGTERM) - # - # for process in children: - # # print(process) - # process.terminate() - self._target.stop() - # 关闭事件管理器 - self.manager.stop() + for watched in self.watched: + watched.stop() parent_path = os.path.abspath(os.path.dirname(os.path.dirname(__file__))) # 获得所在的目录的父级目 path = os.path.join(parent_path, 'Bilibili.py') if sys.platform == 'win32': @@ -110,23 +85,4 @@ class Autoreload: else: args = [path, 'start'] subprocess.Popen(args) - logger.info('重启') - # 同属于一个进程组所以不能用killpg - # os.killpg(os.getpgid(pid), SIGTERM) - return - - def start_change_detector(self, interval): - """Check file state ervry interval. If any change is detected, exit this - process with a special code, so that deamon will to restart a new process. - """ - mtimes = {} - while 1: - if self._is_any_file_changed(mtimes): - self._restart_subp(interval) - return - time.sleep(interval) - - -def autoreload(manager, timer, interval=10): - detector = Autoreload(manager, timer, interval) - detector() + return logger.info('重启') diff --git a/common/timer.py b/common/timer.py index cbfbc28..6845757 100644 --- a/common/timer.py +++ b/common/timer.py @@ -1,9 +1,10 @@ +import asyncio import threading class Timer(threading.Thread): - def __init__(self, func=None, args=(), kwargs=None, interval=15): - threading.Thread.__init__(self, daemon=True) + def __init__(self, func=None, args=(), kwargs=None, interval=15, daemon=True): + threading.Thread.__init__(self, daemon=daemon) if kwargs is None: kwargs = {} self._args = args @@ -11,6 +12,20 @@ class Timer(threading.Thread): self._flag = threading.Event() self._func = func self.interval = interval + self.task = None + self.asynchronous = False + + async def astart(self): + self.asynchronous = True + await self.arun() + + async def arun(self): + while True: + await self.atimer() + await asyncio.sleep(self.interval) + + async def atimer(self): + await self._func(*self._args, **self._kwargs) def timer(self): self._func(*self._args, **self._kwargs) @@ -21,4 +36,6 @@ class Timer(threading.Thread): self._flag.wait(self.interval) def stop(self): - self._flag.set() + if not self.asynchronous: + return self._flag.set() + self.task.cancel() diff --git a/engine/__init__.py b/engine/__init__.py index 837b866..d1c50f0 100644 --- a/engine/__init__.py +++ b/engine/__init__.py @@ -1,6 +1,8 @@ +import asyncio + import yaml from common.event import Event -from common.reload import autoreload +from common.reload import AutoReload from common.timer import Timer from engine.downloader import sorted_checker @@ -29,21 +31,20 @@ platforms = checker.keys() context = {**config, "urls": urls, "url_status": url_status} -def check_timer(event_manager): +async def check_timer(event_manager): event_manager.send_event(Event(CHECK_UPLOAD)) for k in platforms: event_manager.send_event(Event(CHECK, (k,))) -def main(event_manager): +async def main(event_manager): + event_manager.start() # 初始化定时器 timer = Timer(func=check_timer, args=(event_manager,), interval=40) # 模块更新自动重启 - autoreload(event_manager, timer, interval=15) - - event_manager.start() - timer.start() + detector = AutoReload(event_manager, timer, interval=15) + await asyncio.gather(detector.astart(), timer.astart()) CHECK = 'check' diff --git a/engine/plugins/upload/bili_web.py b/engine/plugins/upload/bili_web.py index 72c3d2f..9bb4111 100644 --- a/engine/plugins/upload/bili_web.py +++ b/engine/plugins/upload/bili_web.py @@ -266,7 +266,6 @@ class BiliBili: 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": ""}