Bumping version

add event_loop_interval

app/ui/plugins/global.tsx

@@ -139,8 +139,7 @@ const Global: React.FC = () => {
                     field="event_loop_interval"
                     extraText={
                         <div style={{ fontSize: "14px" }}>
-                            平台检测间隔时间，如虎牙所有主播检测完后会等待 30
-                            秒，再去从新检测
+                            单个主播检测间隔时间，单位：秒。比如虎牙有10个主播，每个主播会间隔10秒检测
                             <br />
                             单位：秒
                         </div>

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,

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())
 

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()

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):

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,)))

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 = []
 

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)

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'