diff --git a/app/ui/plugins/global.tsx b/app/ui/plugins/global.tsx index d87cfcb..c2ee0e3 100644 --- a/app/ui/plugins/global.tsx +++ b/app/ui/plugins/global.tsx @@ -139,8 +139,7 @@ const Global: React.FC = () => { field="event_loop_interval" extraText={
- 平台检测间隔时间,如虎牙所有主播检测完后会等待 30 - 秒,再去从新检测 + 单个主播检测间隔时间,单位:秒。比如虎牙有10个主播,每个主播会间隔10秒检测
单位:秒
diff --git a/biliup/__init__.py b/biliup/__init__.py index 34234fe..3b2b053 100644 --- a/biliup/__init__.py +++ b/biliup/__init__.py @@ -2,7 +2,7 @@ import logging import platform import sys -__version__ = "0.4.41" +__version__ = "0.4.42" LOG_CONF = { 'version': 1, diff --git a/biliup/__main__.py b/biliup/__main__.py index bd59ec4..7445a46 100644 --- a/biliup/__main__.py +++ b/biliup/__main__.py @@ -63,11 +63,9 @@ def arg_parser(): async def main(args): - from .app import event_manager, context - from biliup.downloader import check_flag + from .app import event_manager event_manager.start() - wait = max(config.get('event_loop_interval', 40) - 3, 3) # 启动时删除临时文件夹 shutil.rmtree('./cache/temp', ignore_errors=True) @@ -77,12 +75,12 @@ async def main(args): if not args.no_http: import biliup.web runner = await biliup.web.service(args) - detector = AutoReload(event_manager, runner.cleanup, check_flag.set, interval=interval) + detector = AutoReload(event_manager, runner.cleanup, interval=interval) biliup.common.reload.global_reloader = detector await detector.astart() else: import biliup.common.reload - detector = AutoReload(event_manager, check_flag.set, interval=interval) + detector = AutoReload(event_manager, interval=interval) biliup.common.reload.global_reloader = detector await asyncio.gather(detector.astart()) diff --git a/biliup/app.py b/biliup/app.py index 7aa18e0..c2c5802 100644 --- a/biliup/app.py +++ b/biliup/app.py @@ -1,5 +1,6 @@ import asyncio import logging +from concurrent.futures import ThreadPoolExecutor from . import plugins from biliup.config import config @@ -11,8 +12,13 @@ logger = logging.getLogger('biliup') def create_event_manager(): pool1_size = config.get('pool1_size', 3) pool2_size = config.get('pool2_size', 3) + pool = { + 'Asynchronous1': ThreadPoolExecutor(pool1_size, thread_name_prefix='Asynchronous1'), + 'Asynchronous2': ThreadPoolExecutor(pool2_size, thread_name_prefix='Asynchronous2'), + 'Asynchronous3': ThreadPoolExecutor(2, thread_name_prefix='Asynchronous3'), + } # 初始化事件管理器 - app = EventManager(config, pool1_size=pool1_size, pool2_size=pool2_size) + app = EventManager(config, pool) app.context['url_upload_count'] = {} # 正在上传的文件 用于同时上传一个url的时候过滤掉正在上传的 app.context['upload_filename'] = [] @@ -36,7 +42,7 @@ async def shot(event): cur = event.url_list[index] event_manager.send_event(Event(CHECK, (event, context['PluginInfo'].inverted_index[cur], cur))) index += 1 - await asyncio.sleep(30) + await asyncio.sleep(config.get('event_loop_interval', 30)) @event_manager.server() diff --git a/biliup/database/db.py b/biliup/database/db.py index 0234ce7..e40c763 100644 --- a/biliup/database/db.py +++ b/biliup/database/db.py @@ -30,10 +30,6 @@ def datetime_to_struct_time(date: datetime): return time.localtime(date.timestamp()) -# class DB: - """数据库交互类""" - -# @classmethod def init(no_http): """初始化数据库""" run = not Path.cwd().joinpath("data/data.sqlite3").exists() @@ -47,17 +43,7 @@ def init(no_http): migrate_via_alembic() return run or no_http -# @classmethod -def close(db: Session): - """ 关闭 Session, 释放连接到连接池 """ - db.close() -# @classmethod -# def remove(db: Session): -# """ 释放连接, 并删除注册表中的 Session """ -# db.remove() - -# @classmethod def get_stream_info(db: Session, name: str) -> dict: """根据 streamer 获取下载信息, 若不存在则返回空字典""" res = db.execute( @@ -71,7 +57,7 @@ def get_stream_info(db: Session, name: str) -> dict: return res return {} -# @classmethod + def get_stream_info_by_filename(db: Session, filename: str) -> dict: """通过文件名获取下载信息, 若不存在则返回空字典""" try: @@ -87,7 +73,7 @@ def get_stream_info_by_filename(db: Session, filename: str) -> dict: stream_info_dict["date"] = datetime_to_struct_time(stream_info_dict["date"]) # 将开播时间转回 struct_time 类型 return stream_info_dict -# @classmethod + def add_stream_info(db: Session, name: str, url: str, date: time.struct_time) -> int: """添加下载信息, 返回所添加行的 id """ streamerinfo = StreamerInfo( @@ -101,7 +87,7 @@ def add_stream_info(db: Session, name: str, url: str, date: time.struct_time) -> db.commit() return streamerinfo.id -# @classmethod + def delete_stream_info(db: Session, name: str) -> int: """根据 streamer 删除下载信息, 返回删除的行数, 若不存在则返回 0 """ result = db.execute( @@ -110,7 +96,7 @@ def delete_stream_info(db: Session, name: str) -> int: # db.refresh(result) return result.rowcount() -# @classmethod + def delete_stream_info_by_date(db: Session, name: str, date: time.struct_time) -> int: """根据 streamer 和开播时间删除下载信息, 返回删除的行数, 若不存在则返回 0 """ start_datetime = struct_time_to_datetime(date) @@ -124,7 +110,7 @@ def delete_stream_info_by_date(db: Session, name: str, date: time.struct_time) - db.commit() return result.rowcount() -# @classmethod + def update_cover_path(db: Session, database_row_id: int, live_cover_path: str): """更新封面存储路径""" if not live_cover_path: @@ -133,7 +119,7 @@ def update_cover_path(db: Session, database_row_id: int, live_cover_path: str): streamerinfo.live_cover_path = live_cover_path db.commit() -# @classmethod + def update_room_title(db: Session, database_row_id: int, title: str): """更新直播标题""" if not title: @@ -142,7 +128,7 @@ def update_room_title(db: Session, database_row_id: int, title: str): streamerinfo.title = title db.commit() -# @classmethod + def update_file_list(db: Session, database_row_id: int, file_name: str) -> int: """向视频文件列表中添加文件名""" streamer_info = db.get(StreamerInfo, database_row_id) @@ -151,13 +137,13 @@ def update_file_list(db: Session, database_row_id: int, file_name: str) -> int: db.commit() return file_list.id -# @classmethod + def get_file_list(db: Session, database_row_id: int) -> List[str]: """获取视频文件列表""" file_list = db.get(StreamerInfo, database_row_id).filelist return [file.file for file in file_list] -# @classmethod + def migrate_via_alembic(): """ 自动迁移,通过 alembic 实现 """ def process_revision_directives(context, revision, directives): diff --git a/biliup/downloader.py b/biliup/downloader.py index 372d8b5..69d8fd3 100644 --- a/biliup/downloader.py +++ b/biliup/downloader.py @@ -13,7 +13,6 @@ from .engine.event import Event from .plugins import general logger = logging.getLogger('biliup') -check_flag = threading.Event() def download(fname, url, **kwargs): pg = general.__plugin__(fname, url) @@ -25,84 +24,3 @@ def download(fname, url, **kwargs): pg.__dict__[k] = kwargs.get(k) break return pg.start() - - -def check_url(checker): - # 单主播检测延迟 - checker_sleep = config.get('checker_sleep', 10) - # 平台检测延迟 - event_loop_interval = config.get('event_loop_interval', 30) - class_reference = type(checker('', '')) - - while not check_flag.is_set(): - try: - # 待检测url - check_urls = [] - # 过滤url - for url in checker.url_list: - # 同一主播多个url - # 多个url只能同时下载一个 - is_download = False - streamer_urls = context['streamers'][context['inverted_index'][url]]['url'] - for streamer_url in streamer_urls: - if context['url_status'][streamer_url] == 1: - logger.debug(f'{url}-{streamer_url}-正在下载中,跳过检测') - is_download = True - break - if is_download: - continue - - send_upload_event({'name': context['inverted_index'][url], 'url': url}) - check_urls.append(url) - - if DownloadBase.batch_check != getattr(class_reference, DownloadBase.batch_check.__name__): - # 如果支持批量检测 - # 发送下载的事件 - for url in class_reference.batch_check(check_urls): - send_download_event(context['inverted_index'][url], url) - else: - # 不支持批量检测 - for (index, url) in enumerate(check_urls): - # 某个检测异常略过不应影响其他检测 - try: - if index > 0: - logger.debug('歇息会') - time.sleep(checker_sleep) - - if checker(context['inverted_index'][url], url).check_stream(True): - send_download_event(context['inverted_index'][url], url) - except HTTPError as e: - logger.error(f'{checker.__module__} {e.url} => {e}') - except IOError: - logger.exception("IOError") - except: - logger.exception("Uncaught exception:") - - except: - # 除了单个检测异常 其他异常会影响整体 直接略过本次 等待下次整体检测 - logger.exception("Uncaught exception:") - - time.sleep(event_loop_interval) - - -def send_download_event(name, url): - # 永远不可能对同一个url同时发送两次下载事件 - from .handler import event_manager, DOWNLOAD - - # 需要等待上传文件列表检索完成后才可以开始下次下载 - with NamedLock(f'upload_file_list_{name}'): - for streamer_url in context['streamers'][context['inverted_index'][url]]['url']: - if context['url_status'][streamer_url] == 1: - return False - context['url_status'][url] = 1 - event_manager.send_event(Event(DOWNLOAD, args=(name, url,))) - return True - - -def send_upload_event(stream_info): - # 可能对同一个url同时发送两次上传事件 - with NamedLock(f"upload_count_{stream_info['url']}"): - from .handler import event_manager, UPLOAD - # += 不是原子操作 - context['url_upload_count'][stream_info['url']] += 1 - event_manager.send_event(Event(UPLOAD, (stream_info,))) diff --git a/biliup/engine/event.py b/biliup/engine/event.py index 12519d5..bfeaabc 100644 --- a/biliup/engine/event.py +++ b/biliup/engine/event.py @@ -11,9 +11,11 @@ import functools logger = logging.getLogger('biliup') class EventManager(Thread): - def __init__(self, context=None, pool1_size=3, pool2_size=3): + def __init__(self, context=None, pool=None): """初始化事件管理器""" super().__init__(name='Synchronous', daemon=True) + if pool is None: + pool = {} if context is None: context = {} self.context = context @@ -21,11 +23,8 @@ class EventManager(Thread): self.__eventQueue = Queue() # 事件管理器开关 self.__active = True - # 事件处理线程池1 - self._pool = { - 'Asynchronous1': ThreadPoolExecutor(pool1_size, thread_name_prefix='Asynchronous1'), - 'Asynchronous2': ThreadPoolExecutor(pool2_size, thread_name_prefix='Asynchronous2') - } + # 事件处理线程池 + self._pool = pool # 阻塞函数列表 self.__block = [] diff --git a/biliup/handler.py b/biliup/handler.py index 681c6fc..57e2d02 100644 --- a/biliup/handler.py +++ b/biliup/handler.py @@ -1,4 +1,3 @@ -import copy import json import logging import os @@ -13,7 +12,7 @@ from biliup.config import config from .app import event_manager, context from .common.tools import NamedLock from .database.db import get_stream_info_by_filename, SessionLocal -from .downloader import download, send_upload_event +from .downloader import download from .engine.event import Event from .engine.upload import UploadBase from .uploader import upload, fmt_title_and_desc @@ -146,7 +145,7 @@ def process_upload(stream_info): with NamedLock(f'upload_count_{url}'): url_upload_count[url] -= 1 -@event_manager.register(UPLOADED, block='Asynchronous2') +@event_manager.register(UPLOADED, block='Asynchronous3') def uploaded(name, live_cover_path, data: List): # data = file_list post_processor = config['streamers'].get(name, {}).get("postprocessor", None) diff --git a/biliup/web/__init__.py b/biliup/web/__init__.py index 344314b..13fdfd1 100644 --- a/biliup/web/__init__.py +++ b/biliup/web/__init__.py @@ -440,6 +440,21 @@ async def dump_config(request): return web.json_response({'path': file}) +@routes.get('/v1/status') +async def app_status(request): + from biliup.app import context + from biliup.config import Config + from biliup.app import PluginInfo + res = {} + for key, value in context.items(): # 遍历删除不能被 json 序列化的键值对 + if isinstance(value, Config): + continue + if isinstance(value, PluginInfo): + continue + res[key] = value + return web.json_response(res) + + @routes.get('/bili/archive/pre') async def pre_archive(request): # path = 'cookies.json'