添加自选画质、调整下载策略、修复抖音弹幕分段 (#536)

* 虎牙增加画质选择

* 修复抖音弹幕分段
修复弹幕录制Stop

* 调整直播流获取失败及下播延迟检测功能

* 添加斗鱼自选画质

* 添加哔哩哔哩自选画质

* 添加抖音自选画质

* 优化下载流程

* 优化下载流程

* 虎牙画质修复

* 调整youtube下载逻辑

* 调整下载日志逻辑

* 调整下播检测延迟阈值

* 优化youtube下载策略
优化twitch下载策略
调整检测逻辑
调整封面下载逻辑

* 添加twitch视频下载Cookie#344

* 调整Twitch下载逻辑
调整批量检测功能

* 处理录制弹幕出现的未关闭情况
抖音弹幕录制调整

biliup/downloader.py

@@ -3,8 +3,10 @@ import re
 import time
 from urllib.error import HTTPError
 
+from .config import config
 from .engine.decorators import Plugin
-from .plugins import general, BatchCheckBase
+from .engine.download import DownloadBase
+from .plugins import general
 
 logger = logging.getLogger('biliup')
 
@@ -23,20 +25,31 @@ def download(fname, url, **kwargs):
 
 def check_url(plugin, url_status, secs=15):
     try:
-        if isinstance(plugin, BatchCheckBase):
-            return (yield from plugin.check())
+        # 待检测url
+        check_urls = []
+        # 过滤url
         for url in plugin.url_list:
-            # print(f"URL：{url}")
-            # print(f"URL_STATUS：{url_status[url]}")
             if url_status[url] == 1:
                 logger.debug(f'{url}正在下载中，已跳过检测')
-                # print(f"正在下载中，已跳过检测")
                 continue
-            if plugin(f'检测{url}', url).check_stream():
-                yield url
-            if url != plugin.url_list[-1]:
-                logger.debug('歇息会')
-                time.sleep(secs)
+            if url_status[url] == 2 and not config.get('uploading_record'):
+                logger.debug(f'{url}正在上传中，已跳过检测')
+                continue
+            check_urls.append(url)
+
+        if DownloadBase.batch_check != getattr(plugin.static_class, DownloadBase.batch_check.__name__):
+            # 如果支持批量检测
+            yield from plugin.static_class.batch_check(check_urls)
+        else:
+            # 不支持批量检测
+            for url in check_urls:
+                from .handler import event_manager
+                if plugin(event_manager.context['inverted_index'][url], url).check_stream(True):
+                    yield url
+                if url != check_urls[-1]:
+                    logger.debug('歇息会')
+                    time.sleep(secs)
+
     except HTTPError as e:
         logger.error(f'{plugin.__module__} {e.url} => {e}')
     except IOError:

biliup/engine/decorators.py

@@ -27,6 +27,7 @@ class Plugin:
             def wrapper(*args, **kw):
                 return cls(*args, **kw)
             wrapper.VALID_URL_BASE = regexp
+            wrapper.static_class = cls
             Plugin.download_plugins.append(wrapper)
             return wrapper
         return decorator
@@ -47,8 +48,6 @@ class Plugin:
             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

biliup/engine/download.py

@@ -6,8 +6,10 @@ import subprocess
 import sys
 import threading
 import time
+from typing import Generator
 from urllib.parse import urlparse
 
+import requests
 import stream_gears
 
 from biliup.config import config
@@ -35,6 +37,9 @@ class DownloadBase:
         self.filename_prefix = config.get('filename_prefix')
         self.use_live_cover = config.get('use_live_cover', False)
         self.opt_args = opt_args
+        # 是否是下载模式 跳过下播检测
+        self.is_download = False
+        self.live_cover_url = None
         self.fake_headers = {
             'Accept': 'text/html,application/xhtml+xml,application/xml;q=0.9,*/*;q=0.8',
             'Accept-Encoding': 'gzip, deflate',
@@ -52,54 +57,55 @@ class DownloadBase:
             self.default_output_args += \
                 ['-fs', f"{config.get('file_size', '2621440000')}"]
 
-    def check_stream(self):
-        logger.debug(self.fname)
+    def check_stream(self, is_check=False):
+        # is_check 是否是检测可以避免在检测是否可以录制的时候忽略一些耗时的操作
+        logger.debug(self.fname, is_check)
         raise NotImplementedError()
 
-    def download(self, filename):
+    @staticmethod
+    def batch_check(check_urls: list[str]) -> Generator[str, None, None]:
+        # 批量检测直播或下载状态
+        # 返回的是url_list
+        raise NotImplementedError()
+
+    def get_filename(self, is_fmt=False):
         if self.filename_prefix:  # 判断是否存在自定义录播命名设置
             filename = (self.filename_prefix.format(streamer=self.fname, title=self.room_title).encode(
                 'unicode-escape').decode()).encode().decode("unicode-escape")
         else:
             filename = f'{self.fname}%Y-%m-%dT%H_%M_%S'
         filename = get_valid_filename(filename)
+        if is_fmt:
+            return time.strftime(filename.encode("unicode-escape").decode()).encode().decode("unicode-escape")
+        else:
+            return filename
+
+    def download(self, filename):
+        filename = self.get_filename()
         fmtname = time.strftime(filename.encode("unicode-escape").decode()).encode().decode("unicode-escape")
+
         threading.Thread(target=asyncio.run, args=(self.danmaku_download_start(fmtname),)).start()
-        if self.downloader == 'stream-gears':
-            stream_gears_download(self.raw_stream_url, self.fake_headers, filename, config.get('segment_time'),
-                                  config.get('file_size'))
-        elif self.downloader == 'streamlink':
+
+        if self.downloader == 'streamlink':
             parsed_url = urlparse(self.raw_stream_url)
             path = parsed_url.path
-            if '.flv' in path: #streamlink无法处理flv,所以回退到ffmpeg
-                self.ffmpeg_download(fmtname)
+            if '.flv' in path:  # streamlink无法处理flv,所以回退到ffmpeg
+                return self.ffmpeg_download(fmtname)
             else:
-                self.streamlink_download(fmtname)
-        else:
-            self.ffmpeg_download(fmtname)
+                return self.streamlink_download(fmtname)
+        elif self.downloader == 'ffmpeg':
+            return self.ffmpeg_download(fmtname)
 
-    def get_live_cover(self, uname, room_id, filename, timestamp, cover_url):
-        import requests
-        headers = self.fake_headers.copy()
-        response = requests.get(cover_url, headers=headers, timeout=5)
-        save_dir = f'cover/{uname}_{room_id}/'
-        local_time = time.strftime('%Y-%m-%d_%H-%M-%S', time.localtime(timestamp))
-        if not os.path.exists(save_dir):
-            os.makedirs(save_dir)
-        cover_file_name = get_valid_filename(f'{filename}_{local_time}.png')
-        live_cover_path = f'{save_dir}{cover_file_name}'
-        if os.path.exists(live_cover_path):
-            return live_cover_path
-        else:
-            with open(live_cover_path, 'wb') as f:
-                f.write(response.content)
-                return live_cover_path
+        stream_gears_download(self.raw_stream_url, self.fake_headers, filename, config.get('segment_time'),
+                              config.get('file_size'))
+        return True
 
-    def streamlink_download(self, filename): #streamlink+ffmpeg混合下载模式，适用于下载hls流
+    def streamlink_download(self, filename):  # streamlink+ffmpeg混合下载模式，适用于下载hls流
         streamlink_input_args = ['--stream-segment-threads', '3', '--hls-playlist-reload-attempts', '1']
         streamlink_cmd = ['streamlink', *streamlink_input_args, self.raw_stream_url, 'best', '-O']
-        ffmpeg_input_args = ['-reconnect_streamed', '1', '-reconnect_delay_max', '20', '-rw_timeout', '20000000']
-        ffmpeg_cmd = ['ffmpeg', '-re', '-i', 'pipe:0', '-y',*ffmpeg_input_args, *self.default_output_args, *self.opt_args, '-c', 'copy', '-f', self.suffix]
+        ffmpeg_input_args = ['-rw_timeout', '20000000']
+        ffmpeg_cmd = ['ffmpeg', '-re', '-i', 'pipe:0', '-y', *ffmpeg_input_args, *self.default_output_args,
+                      *self.opt_args, '-c', 'copy', '-f', self.suffix]
         # if config.get('segment_time'):
         #     ffmpeg_cmd += ['-f', 'segment',
         #              f'{filename} part-%03d.{self.suffix}']
@@ -108,7 +114,8 @@ class DownloadBase:
         #         f'{filename}.{self.suffix}.part']
         ffmpeg_cmd += [f'{filename}.{self.suffix}.part']
         streamlink_proc = subprocess.Popen(streamlink_cmd, stdout=subprocess.PIPE)
-        ffmpeg_proc = subprocess.Popen(ffmpeg_cmd, stdin=streamlink_proc.stdout, stdout=subprocess.PIPE, stderr=subprocess.STDOUT)
+        ffmpeg_proc = subprocess.Popen(ffmpeg_cmd, stdin=streamlink_proc.stdout, stdout=subprocess.PIPE,
+                                       stderr=subprocess.STDOUT)
         try:
             with ffmpeg_proc.stdout as stdout:
                 for line in iter(stdout.readline, b''):
@@ -120,15 +127,17 @@ class DownloadBase:
             if sys.platform != 'win32':
                 ffmpeg_proc.communicate(b'q')
             raise
-        return retval
+        if retval != 0:
+            return False
+        return True
 
     def ffmpeg_download(self, filename):
-        default_input_args = ['-headers', ''.join('%s: %s\r\n' % x for x in self.fake_headers.items()),
-                              '-reconnect_streamed', '1', '-reconnect_delay_max', '20', '-rw_timeout', '20000000']
+        default_input_args = ['-headers', ''.join('%s: %s\r\n' % x for x in self.fake_headers.items()), '-rw_timeout',
+                              '20000000']
         parsed_url = urlparse(self.raw_stream_url)
         path = parsed_url.path
         if '.m3u8' in path:
-            default_input_args += ['-max_reload', '3']
+            default_input_args += ['-max_reload', '1000']
         args = ['ffmpeg', '-y', *default_input_args,
                 '-i', self.raw_stream_url, *self.default_output_args, *self.opt_args,
                 '-c', 'copy', '-f', self.suffix]
@@ -152,7 +161,9 @@ class DownloadBase:
             if sys.platform != 'win32':
                 proc.communicate(b'q')
             raise
-        return retval
+        if retval != 0:
+            return False
+        return True
 
     async def danmaku_download_start(self, filename):
         pass
@@ -162,76 +173,114 @@ class DownloadBase:
             return False
         file_name = self.file_name
         retval = self.download(file_name)
-        logger.info(f'{retval}part: {file_name}.{self.suffix}')
+        logger.info(f'part: {file_name}.{self.suffix}')
         self.rename(f'{file_name}.{self.suffix}')
         return retval
 
     def start(self):
-        i = 0
-        logger.info(f'开始下载 {self.__class__.__name__}：{self.fname}')
+        logger.info(f'开始下载：{self.__class__.__name__}:{self.fname}')
         date = time.localtime()
-        lasttime_download = 0
-        delay = min(config.get('delay', 0), 1800)
-        sleep_time = 5 if delay > 30 else 30
-        check_delay = delay
-        ret = None
-        while i < 60:
-            # 限制每次拉流的时间间隔必须大于delay，防止主播还没推流的时候疯狂重复请求
-            thistime_download = time.time()
-            interval = thistime_download - lasttime_download
-            if not ret is False and interval < sleep_time:
-                logger.info(f'频繁请求：等待 {sleep_time - interval:.2f} 秒后再次请求直播流')
-                time.sleep(sleep_time - interval)
+        delay = int(config.get('delay', 0))
+        # 重试次数
+        retry_count = 0
+        # delay 重试次数
+        retry_count_delay = 0
+        # delay 重试次数 向上取整
+        delay_all_retry_count = -(-delay // 60)
 
+        while True:
+            ret = False
             try:
-                thistime_download = time.time()
                 ret = self.run()
             except:
                 logger.exception('Uncaught exception:')
-                continue
             finally:
-                lasttime_download = thistime_download
                 self.close()
+            if ret:
+                if self.is_download:
+                    # 成功下载后也不检测下一个需要下载的视频而是先上传等待下次检测保证上传时使用下载视频的标题
+                    # 开启边录边传会快些
+                    break
+                # 成功下载重置重试次数
+                retry_count = 0
+                retry_count_delay = 0
+            else:
+                if retry_count < 3:
+                    retry_count += 1
+                    if retry_count == 1:
+                        # 只有第一次显示
+                        logger.info(f'获取流失败：{self.__class__.__name__}:{self.fname}，每隔 10 秒重新获取，共重试3次')
+                        time.sleep(10)
+                    continue
+                if self.is_download:
+                    # 下载模式如果下载失败重试三次后直接跳出
+                    break
 
-            if ret is False:
                 if delay:
-                    if i < 5:
-                        i += 1
-                        logger.info(f'获取失败：无法获取直播流，将在 5 秒后重试，重试次数 {i}/{5}')
-                        time.sleep(5)
-                        continue
+                    retry_count_delay += 1
+                    if retry_count_delay > delay_all_retry_count:
+                        # logger.info(f'下播延迟检测结束：{self.__class__.__name__}:{self.fname}')
+                        break
                     else:
-                        if i == 5:
-                            logger.info(f'检测到delay设置 {delay} 秒，将分为 {-int(-(delay / 60) // 1)} 次检测，每隔 60 秒检测一次开播状态')
-                        if check_delay > 60:
-                            i += 1
-                            check_delay -= 60
+                        if delay < 60:
+                            logger.info(
+                                f'下播延迟检测：{self.__class__.__name__}:{self.fname}，将在 {delay} 秒后检测开播状态')
+                            time.sleep(delay)
+                        else:
+                            if retry_count_delay == 1:
+                                # 只有第一次显示
+                                logger.info(
+                                    f'下播延迟检测：{self.__class__.__name__}:{self.fname}，每隔 60 秒检测开播状态，共检测{delay_all_retry_count}次')
                             time.sleep(60)
-                        else:
-                            time.sleep(check_delay)
-                        if self.check_stream():
-                            i = 0
-                            check_delay = delay
-                            continue
-                        else:
-                            if check_delay > 60:
-                                continue
-                            else:
-                                logger.info(f'delay: {delay} 结束，即将退出下载')
-                break
-            elif ret == 1 or self.downloader == 'stream-gears':
-                time.sleep(45)
+                        continue
+                else:
+                    break
 
-            i += 1
-        logger.info(f'退出下载 {i} : {self.fname}')
+        # 获取封面
+        if self.use_live_cover and self.live_cover_url is not None:
+            try:
+                save_dir = f'cover/{self.__class__.__name__}/{self.fname}/'
+                if not os.path.exists(save_dir):
+                    os.makedirs(save_dir)
+                fmtname = time.strftime(self.get_filename().encode("unicode-escape").decode(), date).encode().decode(
+                    "unicode-escape")
+
+                url_path = urlparse(self.live_cover_url).path
+                suffix = None
+                if '.jpg' in url_path:
+                    suffix = 'jpg'
+                elif '.png' in url_path:
+                    suffix = 'png'
+
+                if suffix:
+                    live_cover_path = f'{save_dir}{fmtname}.{suffix}'
+                    if os.path.exists(live_cover_path):
+                        self.live_cover_path = live_cover_path
+                    else:
+                        response = requests.get(self.live_cover_url, headers=self.fake_headers, timeout=30)
+                        with open(live_cover_path, 'wb') as f:
+                            f.write(response.content)
+                            f.close()
+                            self.live_cover_path = live_cover_path
+                    logger.info(
+                        f'封面下载成功：{self.__class__.__name__}：{self.fname}：{os.path.abspath(self.live_cover_path)}')
+                else:
+                    logger.warning('封面下载失败：{self.__class__.__name__}:{self.fname}：封面格式不支持')
+
+            except:
+                logger.exception('封面下载失败：{self.__class__.__name__}:{self.fname}')
+
+        logger.info(f'退出下载：{self.__class__.__name__}:{self.fname}')
         if config['streamers'].get(self.fname, {}).get('downloaded_processor'):
-            downloaded_processor(config['streamers'].get(self.fname, {}).get('downloaded_processor'), f'{{"name": "{self.fname}", "url": "{self.url}", "room_title": "{self.room_title}", "start_time": "{time.strftime("%Y-%m-%d %H:%M:%S", date)}"}}')
+            downloaded_processor(config['streamers'].get(self.fname, {}).get('downloaded_processor'),
+                                 f'{{"name": "{self.fname}", "url": "{self.url}", "room_title": "{self.room_title}", "start_time": "{time.strftime("%Y-%m-%d %H:%M:%S", date)}"}}')
         return {
             'name': self.fname,
             'url': self.url,
             'title': self.room_title,
             'date': date,
             'live_cover_path': self.live_cover_path,
+            'is_download': self.is_download,
         }
 
     @staticmethod
@@ -240,7 +289,7 @@ class DownloadBase:
             os.rename(file_name + '.part', file_name)
             logger.debug('更名{0}为{1}'.format(file_name + '.part', file_name))
         except FileNotFoundError:
-            logger.info('FileNotFoundError:' + file_name)
+            logger.debug('FileNotFoundError:' + file_name)
         except FileExistsError:
             os.rename(file_name + '.part', file_name)
             logger.info('FileExistsError:更名{0}为{1}'.format(file_name + '.part', file_name))
@@ -302,10 +351,10 @@ def downloaded_processor(processors, data):
         if processor.get('run'):
             try:
                 process_output = subprocess.check_output(
-                    processor['run'], shell = True,
-                    input = data,
-                    stderr = subprocess.STDOUT, text = True)
+                    processor['run'], shell=True,
+                    input=data,
+                    stderr=subprocess.STDOUT, text=True)
                 logger.info(process_output.rstrip())
             except subprocess.CalledProcessError as e:
                 logger.exception(e.output)
-                continue
+                continue

biliup/handler.py

@@ -1,6 +1,5 @@
 import logging
 import subprocess
-from pathlib import Path
 
 from . import plugins
 from .downloader import download, check_url
@@ -8,7 +7,6 @@ from .engine import invert_dict, Plugin
 from biliup.config import config
 from .engine.event import Event, EventManager
 from .uploader import upload
-from functools import reduce
 
 CHECK = 'check'
 CHECK_UPLOAD = 'check_upload'
@@ -86,10 +84,6 @@ class KernelFunc:
     def modify(self, url):
         if not url:
             return logger.debug('无人直播')
-        if self.url_status[url] == 1:
-            return logger.debug('已开播正在下载')
-        if self.url_status[url] == 2 and not config.get('uploading_record'):
-            return logger.debug('正在上传稍后下载')
         name = self.inverted_index[url]
 
         if config['streamers'].get(name, {}).get('preprocessor'):

biliup/plugins/Danmaku/__init__.py

@@ -9,10 +9,9 @@ import re
 import ssl
 import time
 import logging
-from functools import partial
+from typing import Optional
 
 import lxml.etree as etree
-import aiofiles
 import aiohttp
 
 from biliup.plugins.Danmaku.douyu import Douyu
@@ -22,165 +21,192 @@ from biliup.plugins.Danmaku.twitch import Twitch
 from biliup.plugins.Danmaku.douyin import Douyin
 
 logger = logging.getLogger('biliup')
-__all__ = ['DanmakuClient']
 
 
 class DanmakuClient:
+    class WebsocketErrorException(Exception):
+        pass
+
     def __init__(self, url, filename):
         self.__starttime = time.time()
         self.__filename = os.path.splitext(filename)[0] + '.xml'
         self.__filename_video_suffix = filename
         self.__url = ''
         self.__site = None
-        self.__site_douyin = None
         self.__hs = None
         self.__ws = None
-        self.__stop = False
         self.__dm_queue = None
-        self.__link_status = True
+        # 正在运行中的任务
+        self.__danmaku_tasks: Optional[asyncio.Task] = None
+        # 弹幕启动时的事件循环 为了保证被正确的关闭
+        self.__loop = None
 
         if 'http://' == url[:7] or 'https://' == url[:8]:
             self.__url = url
         else:
             self.__url = 'http://' + url
-        if re.match(r'^(?:http[s]?://)?.*?live.douyin.com/(.+?)$', url):
-            self.__site_douyin = Douyin(url, filename)
         for u, s in {'douyu.com': Douyu,
                      'huya.com': Huya,
                      'live.bilibili.com': Bilibili,
-                     'twitch.tv': Twitch
+                     'twitch.tv': Twitch,
+                     'live.douyin.com': Douyin
                      }.items():
             if re.match(r'^(?:http[s]?://)?.*?%s/(.+?)$' % u, url):
                 self.__site = s
                 self.__u = u
                 break
 
-        if self.__site is None and self.__site_douyin is None:
-            print('Invalid link!')
-            exit()
+        if self.__site is None:
+            logger.warning(f"{DanmakuClient.__name__}:{self.__url}:不支持录制弹幕")
+            # 抛出异常由外部处理 exit()会导致进程退出
+            raise Exception(f"{DanmakuClient.__name__}:{self.__url}:不支持录制弹幕")
 
     async def init_ws(self):
-        ws_url, reg_datas = await self.__site.get_ws_info(self.__url)
-        ctx = ssl.create_default_context()
-        ctx.set_ciphers('DEFAULT')
         try:
-            self.__ws = await self.__hs.ws_connect(ws_url, ssl_context=ctx)
+            ws_url, reg_datas = await self.__site.get_ws_info(self.__url)
+            ctx = ssl.create_default_context()
+            ctx.set_ciphers('DEFAULT')
+            self.__ws = await self.__hs.ws_connect(ws_url, ssl_context=ctx, headers=getattr(self.__site, 'headers',{}))
             for reg_data in reg_datas:
                 if type(reg_data) == str:
                     await self.__ws.send_str(reg_data)
                 else:
                     await self.__ws.send_bytes(reg_data)
-        except Exception as Error:
-            logger.debug(f"init_ws：{self.__filename}：捕获到异常：{Error}")
-        if not os.path.exists(self.__filename):
-            async with aiofiles.open(self.__filename, mode='w') as f:
-                await f.write("<?xml version='1.0' encoding='UTF-8'?>\n"
-                              "<i xmlns:ns0='http://www.w3.org/1999/XSL/Transform'>\n"
-                              "</i>"
-                              )
+        except:
+            raise self.WebsocketErrorException()
 
     async def heartbeats(self):
-        while not self.__stop and self.__site.heartbeat:
+        while self.__site.heartbeat:
+            # 每隔这么长时间发送一次心跳包
             await asyncio.sleep(self.__site.heartbeatInterval)
-            try:
-                if type(self.__site.heartbeat) == str:
-                    await self.__ws.send_str(self.__site.heartbeat)
-                else:
-                    await self.__ws.send_bytes(self.__site.heartbeat)
-            except Exception as Error:
-                logger.debug(f"heartbeats：{self.__filename}：捕获到异常：{Error}")
-                if not self.__stop:
-                    logger.info(f'触发弹幕重连')
-                    await self.init_ws()
-                    await asyncio.sleep(3)
-
-        await self.__hs.close()
+            # 发送心跳包
+            if type(self.__site.heartbeat) == str:
+                await self.__ws.send_str(self.__site.heartbeat)
+            else:
+                await self.__ws.send_bytes(self.__site.heartbeat)
 
     async def fetch_danmaku(self):
-        # while not self.__stop:
-        await asyncio.sleep(1)
-        await self.init_ws()
-        await asyncio.sleep(1)
-        async for msg in self.__ws:
+        while True:
+            # 使用 async for msg in self.__ws
+            # 会导致在连接断开时 需要很长时间15min或者更多才能检测到
+            msg = await self.__ws.receive()
+
+            if msg.type in [aiohttp.WSMsgType.CLOSED, aiohttp.WSMsgType.ERROR]:
+                # 连接关闭的异常 会到外层统一处理
+                raise self.WebsocketErrorException()
+
             try:
-                if self.__stop:
-                    await self.__dm_queue.put(None)
-                    return
-                # self.__link_status = True
-                ms = self.__site.decode_msg(msg.data)
+                result = self.__site.decode_msg(msg.data)
+
+                if isinstance(result, tuple):
+                    ms, ack = result
+                    if ack is not None:
+                        # 发送ack包
+                        if type(ack) == str:
+                            await self.__ws.send_str(ack)
+                        else:
+                            await self.__ws.send_bytes(ack)
+                else:
+                    ms = result
+
                 for m in ms:
                     await self.__dm_queue.put(m)
             except Exception as Error:
-                logger.debug(f"fetch_danmaku：{self.__filename}：弹幕处理异常：{Error}")
-                await asyncio.sleep(10)
+                logger.warning(f"{DanmakuClient.__name__}:{self.__url}:弹幕接收异常：{Error}")
+                continue
+                # await asyncio.sleep(10) 无需等待 直接获取下一条websocket消息
+                # 这里出现异常只会是 decode_msg 的问题
 
     async def print_danmaku(self):
         parser = etree.XMLParser(recover=True)
-        tree = etree.parse(self.__filename, parser=parser)
-        root = tree.getroot()
+        root = etree.Element("root")
+        tree = etree.ElementTree(root, parser=parser)
         msg_i = 0
-        msg_col = {'0': '16777215', '1': '16717077', '2': '2000880', '3': '8046667', '4': '16744192', '5': '10172916',
+        msg_col = {'0': '16777215', '1': '16717077', '2': '2000880', '3': '8046667', '4': '16744192',
+                   '5': '10172916',
                    '6': '16738740'}
 
         def write_file(filename):
-            with open(filename, "wb") as f:
-                etree.indent(root, "\t")
-                tree.write(f, encoding="UTF-8", xml_declaration=True, pretty_print=True)
-
-        while not self.__stop:
             try:
+                with open(filename, "wb") as f:
+                    etree.indent(root, "\t")
+                    tree.write(f, encoding="UTF-8", xml_declaration=True, pretty_print=True)
+            except Exception as Error:
+                logger.warning(f"{DanmakuClient.__name__}:{self.__url}:弹幕写入异常：{Error}")
+
+        try:
+            while True:
                 m = await self.__dm_queue.get()
-                if m is None:
-                    return
-                if m['msg_type'] == 'danmaku':
-                    d = etree.SubElement(root, 'd')
-                    if 'col' in m:
-                        color = msg_col[m["col"]]
-                    elif 'color' in m:
-                        color = m["color"]
+                if m.get('msg_type') == 'danmaku':
+                    try:
+                        d = etree.SubElement(root, 'd')
+                        if 'col' in m:
+                            color = msg_col[m["col"]]
+                        elif 'color' in m:
+                            color = m["color"]
+                        else:
+                            color = '16777215'
+                        msg_time = format(time.time() - self.__starttime, '.3f')
+                        d.set('p', f"{msg_time},1,25,{color},0,0,0,0")
+                        d.text = m["content"]
+                    except Exception as Error:
+                        logger.warning(f"{DanmakuClient.__name__}:{self.__url}:弹幕处理异常：{Error}")
+                        # 异常后略过本次弹幕
+                        continue
+
+                    if msg_i >= 5:
+                        # 收到五条弹幕后写入 减少io
+                        # 可能会写入失败 会在下次五条或者任务被取消时重新尝试写入
+                        write_file(self.__filename)
+                        msg_i = 0
                     else:
-                        color = '16777215'
-                    msg_time = format(time.time() - self.__starttime, '.3f')
-                    d.set('p', f"{msg_time},1,25,{color},0,0,0,0")
-                    d.text = m["content"]
-            except Exception as Error:
-                logger.debug(f"print_danmaku：{self.__filename}：弹幕处理异常：{Error}")
-
-            try:
-                if m['msg_type'] == 'danmaku' and msg_i >= 5:
-                    # loop = asyncio.get_running_loop()
-                    # await loop.run_in_executor(None, partial(write_file, self.__filename))
-                    write_file(self.__filename)
-                    msg_i = 0
-                else:
-                    msg_i = msg_i + 1
-            except Exception as Error:
-                logger.debug(f"print_danmaku：{self.__filename}：弹幕写入异常：{Error}")
-
-    async def start(self):
-        if self.__site_douyin is not None:
-            await self.__site_douyin.start()
-        else:
-            self.__dm_queue = asyncio.Queue()
-            self.__hs = aiohttp.ClientSession()
-            await self.init_ws()
-            await asyncio.gather(
-                self.heartbeats(),
-                self.fetch_danmaku(),
-                self.print_danmaku(),
-            )
-
-    def stop(self):
-        if self.__site_douyin is not None:
-            self.__site_douyin.stop()
-        else:
-            self.__stop = True
-            self.__hs.close()
+                        msg_i = msg_i + 1
+        except asyncio.CancelledError:
+            # 在这段代码中只有执行m = await self.__dm_queue.get()的时候可能会被取消 所以无需担心其他问题
+            # 任务被取消之前需要写入一下 不然会丢失没写入的
+            write_file(self.__filename)
+            # 如果不存在对应的视频则删除弹幕
             if not (os.path.exists(f"{self.__filename_video_suffix}.part") or
                     os.path.exists(f"{self.__filename_video_suffix}")):
                 os.remove(self.__filename)
 
+    async def start(self):
+        self.__loop = asyncio.get_event_loop()
+        self.__dm_queue = asyncio.Queue()
+        is_first_error = True
+        while True:
+            try:
+                self.__hs = aiohttp.ClientSession()
+                await self.init_ws()
+                # 保存以便取消
+                self.__danmaku_tasks = asyncio.gather(self.heartbeats(), self.fetch_danmaku(), self.print_danmaku())
+                await self.__danmaku_tasks
+            except self.WebsocketErrorException:
+                self.stop()
+                if is_first_error:
+                    # 减少日志输出量
+                    logger.warning(f"{DanmakuClient.__name__}:{self.__url}:弹幕连接异常，将在 30 秒后重试")
+                    is_first_error = False
 
+                # 连接断开等30秒重连
+                # 在关闭之前一直重试
+                await asyncio.sleep(30)
+                continue
+            except asyncio.CancelledError:
+                pass
+            break
+
+    def stop(self):
+        # 关闭任务后loop就会被自动关闭 这里让他先保留一下 等待其他close完成后手动关闭
+        sleep_task = asyncio.run_coroutine_threadsafe(asyncio.sleep(9999), self.__loop)
+        # 关闭任务
+        if self.__danmaku_tasks is not None:
+            self.__danmaku_tasks.cancel()
+        if self.__ws is not None:
+            asyncio.run_coroutine_threadsafe(self.__ws.close(), self.__loop)
+        if self.__hs is not None:
+            asyncio.run_coroutine_threadsafe(self.__hs.close(), self.__loop)
+        sleep_task.cancel()
 
 # 虎牙直播：https://www.huya.com/lpl
 # 斗鱼直播：https://www.douyu.com/9999

biliup/plugins/Danmaku/douyin.py

@@ -1,20 +1,14 @@
 # 抖音的弹幕录制参考了 https://github.com/LyzenX/DouyinLiveRecorder 和 https://github.com/YunzhiYike/live-tool
 # 2023.07.14：KNaiFen：这部分代码参考了https://github.com/SmallPeaches/DanmakuRender
 
-from datetime import datetime
-import threading
-import asyncio
 import gzip
 import time
 import re
-import requests
-import urllib
-import json
-import os
-import queue
+from urllib.parse import unquote
+
+import requests
+import json
 
-import websocket
-import lxml.etree as etree
 from google.protobuf import json_format
 
 from .douyin_util.dy_pb2 import PushFrame, Response, ChatMessage  # 反序列化后导入
@@ -25,156 +19,57 @@ class Douyin:
         'authority': 'live.douyin.com',
         'user-agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/104.0.5112.81 Safari/537.36 Edg/104.0.1293.54',
     }
-    heartbeatInterval = 30
+    heartbeat = b':\x02hb'
+    heartbeatInterval = 10
 
-    def __init__(self, url, filename):
-        self.__ws = None
-        self.__stop = False
-        self.__web_rid = url.split('/')[-1]
-        self.__dm_queue = queue.Queue()
-        self.ttwid = None
-        self.__starttime = time.time()
-        self.__filename = os.path.splitext(filename)[0] + '.xml'
-        self.__filename_video_suffix = filename
-        self.__url = ''
-
-        if not os.path.exists(self.__filename):
-            with open(self.__filename, mode='w') as f:
-                f.write("<?xml version='1.0' encoding='UTF-8'?>\n"
-                        "<i xmlns:ns0='http://www.w3.org/1999/XSL/Transform'>\n"
-                        "</i>"
-                        )
-
-        if not self.headers.get('cookie'):
-            try:
-                response = requests.get(f'https://live.douyin.com/{self.__web_rid}', headers=self.headers, timeout=5)
-                self.headers.update({'cookie': '__ac_nonce=' + response.cookies.get('__ac_nonce')})
-
-                response = requests.get(f'https://live.douyin.com/{self.__web_rid}', headers=self.headers, timeout=5)
-                self.headers['cookie'] += '; ttwid=' + response.cookies.get('ttwid')
-            except Exception as e:
-                # logging.exception(e)
-                raise Exception('获取抖音cookies错误.')
-
-        if len(self.__web_rid) == 19:
-            self.real_rid = self.__web_rid
+    @staticmethod
+    async def get_ws_info(url):
+        web_rid = url.split('/')[-1]
+        if not Douyin.headers.get('cookie'):
+            response = requests.get(f'https://live.douyin.com/{web_rid}', headers=Douyin.headers, timeout=5)
+            Douyin.headers['cookie'] = '__ac_nonce=' + response.cookies.get('__ac_nonce')
+            response = requests.get(f'https://live.douyin.com/{web_rid}', headers=Douyin.headers, timeout=5)
+            Douyin.headers['cookie'] += '; ttwid=' + response.cookies.get('ttwid')
         else:
-            try:
-                resp = self._get_response_douyin()
-                self.real_rid = resp['app']['initialState']['roomStore']['roomInfo']['roomId']
-            except:
-                raise Exception('房间号错误.')
+            response = requests.get(f'https://live.douyin.com/{web_rid}', headers=Douyin.headers, timeout=5)
 
-    def _get_response_douyin(self):
-        text = requests.get(f'https://live.douyin.com/{self.__web_rid}', headers=self.headers, timeout=5).text
-        render_data = re.findall(r"<script id=\"RENDER_DATA\" type=\"application/json\">.*?</script>", text)[0]
-        data = urllib.parse.unquote(render_data)
-        data = re.sub(r'(<script.*?>|</script>)', '', data)
-        data = json.loads(data)
+        render_data = re.findall(r"<script id=\"RENDER_DATA\" type=\"application/json\">.*?</script>", response.text)[0]
+        render_data = unquote(render_data)
+        render_data = re.sub(r'(<script.*?>|</script>)', '', render_data)
+        data = json.loads(render_data)
+        real_rid = data['app']['initialState']['roomStore']['roomInfo']['roomId']
+        user_unique_id = data['app']['odin']['user_unique_id']
 
-        return data
+        url = f"wss://webcast3-ws-web-lf.douyin.com/webcast/im/push/v2/?app_name=douyin_web&version_code=180800&webcast_sdk_version=1.3.0&update_version_code=1.3.0&compress=gzip&internal_ext=internal_src:dim|wss_push_room_id:{real_rid}|wss_push_did:{user_unique_id}|dim_log_id:2023011316221327ACACF0E44A2C0E8200|fetch_time:${int(time.time())}123|seq:1|wss_info:0-1673598133900-0-0|wrds_kvs:WebcastRoomRankMessage-1673597852921055645_WebcastRoomStatsMessage-1673598128993068211&cursor=u-1_h-1_t-1672732684536_r-1_d-1&host=https://live.douyin.com&aid=6383&live_id=1&did_rule=3&debug=false&endpoint=live_pc&support_wrds=1&im_path=/webcast/im/fetch/&device_platform=web&cookie_enabled=true&screen_width=1228&screen_height=691&browser_language=zh-CN&browser_platform=Mozilla&browser_name=Mozilla&browser_version=5.0%20(Windows%20NT%2010.0;%20Win64;%20x64)%20AppleWebKit/537.36%20(KHTML,%20like%20Gecko)%20Chrome/100.0.4896.75%20Safari/537.36&browser_online=true&tz_name=Asia/Shanghai&identity=audience&room_id={real_rid}&heartbeatDuration=0&signature=00000000"
+        return url, []
 
-    def get_danmu_ws_url(self):
-        resp = self._get_response_douyin()
-        user_unique_id = resp['app']['odin']['user_unique_id']
-        return f"wss://webcast3-ws-web-lf.douyin.com/webcast/im/push/v2/?app_name=douyin_web&version_code=180800&webcast_sdk_version=1.3.0&update_version_code=1.3.0&compress=gzip&internal_ext=internal_src:dim|wss_push_room_id:{self.real_rid}|wss_push_did:{user_unique_id}|dim_log_id:2023011316221327ACACF0E44A2C0E8200|fetch_time:${int(time.time())}123|seq:1|wss_info:0-1673598133900-0-0|wrds_kvs:WebcastRoomRankMessage-1673597852921055645_WebcastRoomStatsMessage-1673598128993068211&cursor=u-1_h-1_t-1672732684536_r-1_d-1&host=https://live.douyin.com&aid=6383&live_id=1&did_rule=3&debug=false&endpoint=live_pc&support_wrds=1&im_path=/webcast/im/fetch/&device_platform=web&cookie_enabled=true&screen_width=1228&screen_height=691&browser_language=zh-CN&browser_platform=Mozilla&browser_name=Mozilla&browser_version=5.0%20(Windows%20NT%2010.0;%20Win64;%20x64)%20AppleWebKit/537.36%20(KHTML,%20like%20Gecko)%20Chrome/100.0.4896.75%20Safari/537.36&browser_online=true&tz_name=Asia/Shanghai&identity=audience&room_id={self.real_rid}&heartbeatDuration=0&signature=00000000"
-
-    async def start(self):
-        self.__ws = websocket.WebSocketApp(
-            url=self.get_danmu_ws_url(),
-            header=self.headers,
-            cookie=self.headers.get('cookie'),
-            on_message=self._onMessage,
-            on_error=self._onError,
-            on_open=self._onOpen,
-        )
-        t = threading.Thread(target=self.__ws.run_forever, daemon=True)
-        t.start()
-
-        parser = etree.XMLParser(recover=True)
-        tree = etree.parse(self.__filename, parser=parser)
-        root = tree.getroot()
-
-        def write_file(filename):
-            with open(filename, "wb") as f:
-                etree.indent(root, "\t")
-                tree.write(f, encoding="UTF-8", xml_declaration=True, pretty_print=True)
-
-        msg_i = 0
-        while not self.__stop:
-            try:
-                m = self.__dm_queue.get_nowait()
-            except Exception as Error:
-                await asyncio.sleep(1)
-                continue
-
-            try:
-                if m['msg_type'] == 'danmaku':
-                    d = etree.SubElement(root, 'd')
-                    msg_time = format(time.time() - self.__starttime, '.3f')
-                    d.set('p', f"{msg_time},1,25,16777215,0,0,0,0")
-                    d.text = m["content"]
-
-                    if msg_i >= 1:
-                        write_file(self.__filename)
-                        msg_i = 0
-                    else:
-                        msg_i = msg_i + 1
-            except Exception as Error:
-                pass
-
-
-
-    async def stop(self):
-        self.__stop = True
-        self.__ws.close()
-        if not (os.path.exists(f"{self.__filename_video_suffix}.part") or
-                os.path.exists(f"{self.__filename_video_suffix}")):
-            os.remove(self.__filename)
-
-    def _onOpen(self, ws):
-        t = threading.Thread(target=self._heartbeat, args=(ws,), daemon=True)
-        t.start()
-
-    def _onMessage(self, ws: websocket.WebSocketApp, message: bytes):
+    @staticmethod
+    def decode_msg(data):
         wssPackage = PushFrame()
-        wssPackage.ParseFromString(message)
+        wssPackage.ParseFromString(data)
         logid = wssPackage.logId
         decompressed = gzip.decompress(wssPackage.payload)
         payloadPackage = Response()
         payloadPackage.ParseFromString(decompressed)
 
-        # 发送ack包
+        ack = None
         if payloadPackage.needAck:
             obj = PushFrame()
             obj.payloadType = 'ack'
             obj.logId = logid
             obj.payloadType = payloadPackage.internalExt
-            data = obj.SerializeToString()
-            ws.send(data, websocket.ABNF.OPCODE_BINARY)
-        # 处理消息
+            ack = obj.SerializeToString()
+
+        msgs = []
         for msg in payloadPackage.messagesList:
-            # now = datetime.now()
             if msg.method == 'WebcastChatMessage':
                 chatMessage = ChatMessage()
                 chatMessage.ParseFromString(msg.payload)
                 data = json_format.MessageToDict(chatMessage, preserving_proto_field_name=True)
-                name = data['user']['nickName']
+                # name = data['user']['nickName']
                 content = data['content']
                 # msg_dict = {"time": now, "name": name, "content": content, "msg_type": "danmaku", "color": "ffffff"}
                 msg_dict = {"content": content, "msg_type": "danmaku"}
-                # print(msg_dict)
-                self.__dm_queue.put_nowait(msg_dict)
-
-
-    def _heartbeat(self, ws: websocket.WebSocketApp):
-        while not self.__stop:
-            obj = PushFrame()
-            obj.payloadType = 'hb'
-            data = obj.SerializeToString()
-            ws.send(data, websocket.ABNF.OPCODE_BINARY)
-            time.sleep(10)
-
-    def _onError(self, ws, error):
-        raise error
+                msgs.append(msg_dict)
 
+        return msgs, ack

biliup/plugins/__init__.py

@@ -3,26 +3,6 @@ import re
 
 logger = logging.getLogger('biliup')
 
-
-class BatchCheckBase:
-    def __init__(self, pattern_id, urls):
-        self.usr_dict = {}
-        self.usr_list = []
-        self.pattern_id = pattern_id
-        for url in urls:
-            self.get_id(url)
-
-    def get_id(self, url):
-        m = re.match(self.pattern_id, url)
-        if m:
-            usr_id = m.group('id')
-            self.usr_dict[usr_id.lower()] = url
-            self.usr_list.append(usr_id)
-
-    def check(self):
-        pass
-
-
 def match1(text, *patterns):
     if len(patterns) == 1:
         pattern = patterns[0]

biliup/plugins/acfun.py

@@ -12,7 +12,7 @@ class Acfun(DownloadBase):
     def __init__(self, fname, url, suffix='flv'):
         super().__init__(fname, url, suffix)
 
-    def check_stream(self):
+    def check_stream(self, is_check=False):
         if len(self.url.split("acfun.cn/live/")) < 2:
             logger.debug("直播间地址错误")
             return False

biliup/plugins/afreecaTV.py

@@ -17,7 +17,7 @@ class AfreecaTV(DownloadBase):
     def __init__(self, fname, url, suffix='flv'):
         super().__init__(fname, url, suffix)
 
-    def check_stream(self):
+    def check_stream(self, is_check=False):
         logger.debug(self.fname)
         username = match1(self.url, VALID_URL_BASE)
         res_bno = requests.post(CHANNEL_API_URL + "?bjid=" + username,

biliup/plugins/bilibili.py

@@ -18,7 +18,7 @@ class Bilibili(DownloadBase):
         if config.get('user', {}).get('bili_cookie') is not None:
             self.fake_headers['cookie'] = config.get('user', {}).get('bili_cookie')
 
-    def check_stream(self):
+    def check_stream(self, is_check=False):
 
         # 预读配置
         params = {
@@ -26,7 +26,7 @@ class Bilibili(DownloadBase):
             'protocol': '0,1',# 0: http_stream, 1: http_hls
             'format': '0,1,2',# 0: flv, 1: ts, 2: fmp4
             'codec': '0', # 0: avc, 1: hevc
-            'qn': '10000',
+            'qn': config.get('bili_qn', '10000'),
             'platform': 'html5', # 使用 html5 时默认屏蔽 p2p
             'dolby': '5',
             'panorama': '1'
@@ -52,17 +52,12 @@ class Bilibili(DownloadBase):
             if room_info['code'] != 0 or room_info['data']['room_info']['live_status'] != 1:
                 logger.debug(room_info['message'])
                 return False
-            cover_url = room_info['data']['room_info']['cover']
+            self.live_cover_url = room_info['data']['room_info']['cover']
             live_start_time = room_info['data']['room_info']['live_start_time']
             uname = room_info['data']['anchor_info']['base_info']['uname']
             if self.room_title is None:
                 self.room_title = room_info['data']['room_info']['title']
-            if self.use_live_cover is True:  # 获取直播封面并保存到cover目录下
-                try:
-                    self.live_cover_path = \
-                    super().get_live_cover(uname, params['room_id'], self.room_title, live_start_time, cover_url)
-                except:
-                    logger.error("获取直播封面失败")
+
             # 当 Cookie 存在，并且自定义APi使用Cookie开关关闭时，仅使用官方 Api
             isallow = True if s.headers.get('cookie') is None else config.get('user', {}).get('customAPI_use_cookie', False)
             try:

biliup/plugins/cc.py

@@ -13,7 +13,7 @@ class CC(DownloadBase):
     def __init__(self, fname, url, suffix='flv'):
         super().__init__(fname, url, suffix)
 
-    def check_stream(self):
+    def check_stream(self, is_check=False):
         logger.debug(self.fname)
         rid = re.search(r"[0-9]{4,}", self.url).group(0)
         res = requests.get(

biliup/plugins/douyin.py

@@ -16,8 +16,8 @@ class Douyin(DownloadBase):
         super().__init__(fname, url, suffix)
         self.douyin_danmaku = config.get('douyin_danmaku', False)
 
-    def check_stream(self):
-        douyin_url="https://live.douyin.com/"
+    def check_stream(self, is_check=False):
+        douyin_url = "https://live.douyin.com/"
         headers = {
             "user-agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) "
                           "Chrome/94.0.4606.71 Safari/537.36 Edg/94.0.992.38",
@@ -29,13 +29,13 @@ class Douyin(DownloadBase):
                 logger.debug("直播间地址错误")
                 return False
             else:
-                mainPage=requests.get(self.url, headers=headers).text\
-                .split('<script id="RENDER_DATA" type="application/json">')[1].split('</script>')[0]
+                mainPage = requests.get(self.url, headers=headers).text \
+                    .split('<script id="RENDER_DATA" type="application/json">')[1].split('</script>')[0]
                 txt = urllib.request.unquote(mainPage)
                 rex = re.compile(r'(?<=\"web_rid\":\")[0-9]*(?=\")')
                 rid = rex.findall(txt)[0]
         else:
-            #判断是否为纯数字房间号
+            # 判断是否为纯数字房间号
             rid = self.url.split(douyin_url)[1]
             rid = '+{}'.format(rid) if rid.isdigit() else rid
         try:
@@ -51,7 +51,42 @@ class Douyin(DownloadBase):
             return False
         if room_info.get('stream_url'):
             r5 = room_info['stream_url']['live_core_sdk_data']['pull_data']['stream_data']
-            self.raw_stream_url = json.loads(r5)['data']['origin']['main']['flv']
+            stream_data = json.loads(r5)['data']
+
+            # 原画origin 蓝光uhd 超清hd 高清sd 标清ld 流畅md 仅音频ao
+            quality_items = ['origin', 'uhd', 'hd', 'sd', 'ld', 'md']
+            quality = config.get('douyin_quality', 'origin')
+            if quality not in quality_items:
+                quality = quality_items[0]
+
+            # 如果没有这个画质则取相近的 优先低清晰度
+            if quality not in stream_data:
+                # 可选的清晰度 含自身
+                optional_quality_items = [x for x in quality_items if x in stream_data.keys() or x == quality]
+                # 自身在可选清晰度的位置
+                optional_quality_index = optional_quality_items.index(quality)
+                # 自身在所有清晰度的位置
+                quality_index = quality_items.index(quality)
+                # 高清晰度偏移
+                quality_left_offset = None
+                # 低清晰度偏移
+                quality_right_offset = None
+
+                if optional_quality_index + 1 < len(optional_quality_items):
+                    quality_right_offset = quality_items.index(optional_quality_items[optional_quality_index + 1]) - quality_index
+
+                if optional_quality_index - 1 >= 0:
+                    quality_left_offset = quality_index - quality_items.index(optional_quality_items[optional_quality_index - 1])
+
+                # 取相邻的清晰度
+                if quality_right_offset <= quality_left_offset:
+                    quality = optional_quality_items[optional_quality_index + 1]
+                else:
+                    quality = optional_quality_items[optional_quality_index - 1]
+
+
+
+            self.raw_stream_url = json.loads(r5)['data'][quality]['main']['flv']
             self.room_title = room_info['title']
             return True
 
@@ -60,3 +95,8 @@ class Douyin(DownloadBase):
             logger.info("开始弹幕录制")
             self.danmaku = DanmakuClient(self.url, filename + "." + self.suffix)
             await self.danmaku.start()
+
+    def close(self):
+        if self.douyin_danmaku:
+            self.danmaku.stop()
+            logger.info("结束弹幕录制")

biliup/plugins/douyu.py

@@ -14,7 +14,7 @@ class Douyu(DownloadBase):
         super().__init__(fname, url, suffix)
         self.douyu_danmaku = config.get('douyu_danmaku', False)
 
-    def check_stream(self):
+    def check_stream(self, is_check=False):
         if len(self.url.split("douyu.com/")) < 2:
             logger.error("斗鱼：" + self.url + "：地址错误")
             return False
@@ -44,7 +44,7 @@ class Douyu(DownloadBase):
                 'cdn': config.get('douyucdn', 'tct-h5'),
                 'iar': 0,
                 'ive': 0,
-                'rate': 0,
+                'rate': config.get('douyu_rate', 0),
             }
         except:
             logger.warning("斗鱼：" + vid + "：获取roominfo错误")

biliup/plugins/egame.py

@@ -9,7 +9,7 @@ class egame(DownloadBase):
     def __init__(self, fname, url, suffix='flv'):
         super().__init__(fname, url, suffix)
 
-    def check_stream(self):
+    def check_stream(self, is_check=False):
         if len(self.url.split("egame.qq.com/")) < 2:
             logger.debug("直播间格式错误")
             return False

biliup/plugins/general.py

@@ -11,7 +11,7 @@ class YDownload(DownloadBase):
         super().__init__(fname, url, suffix)
         self.ydl_opts = {}
 
-    def check_stream(self):
+    def check_stream(self, is_check=False):
         try:
             self.get_sinfo()
             return True
@@ -48,7 +48,7 @@ class SDownload(DownloadBase):
         self.stream = None
         self.flag = Event()
 
-    def check_stream(self):
+    def check_stream(self, is_check=False):
         logger.debug(self.fname)
         import streamlink
         try:
@@ -84,7 +84,7 @@ class Generic(DownloadBase):
         super().__init__(fname, url, suffix)
         self.handler = self
 
-    def check_stream(self):
+    def check_stream(self, is_check=False):
         logger.debug(self.fname)
         try:
             site, url = url_to_module(self.url)

biliup/plugins/huya.py

@@ -18,13 +18,14 @@ class Huya(DownloadBase):
         self.huya_danmaku = config.get('huya_danmaku', False)
         self.fake_headers['Referer'] = 'https://www.huya.com/'
 
-    def check_stream(self):
+    def check_stream(self, is_check=False):
         try:
             res = requests.get(self.url, timeout=5, headers=self.fake_headers)
             res.close()
         except:
             logger.warning("虎牙 " + self.url.split("huya.com/")[1] + "：获取错误，本次跳过")
             return False
+
         huya = None
         if match1(res.text, '"stream": "([a-zA-Z0-9+=/]+)"'):
             huya = base64.b64decode(match1(res.text, '"stream": "([a-zA-Z0-9+=/]+)"')).decode()
@@ -36,24 +37,43 @@ class Huya(DownloadBase):
                 huya = None
         if huya:
             try:
+                # 自选cdn
                 huyacdn = config.get('huyacdn', 'AL')
-                huyajson1 = json.loads(huya)['data'][0]['gameStreamInfoList']
-                huyajson2 = json.loads(huya)['vMultiStreamInfo']
-                ratio = huyajson2[0]['iBitRate']
-                ibitrate_list = []
-                sdisplayname_list = []
-                for key in huyajson2:
-                    ibitrate_list.append(key['iBitRate'])
-                    sdisplayname_list.append(key['sDisplayName'])
-                    if len(sdisplayname_list) > len(set(sdisplayname_list)):
-                        ratio = max(ibitrate_list)
-                huyajson = huyajson1[0]
-                for cdn in huyajson1:
+                # 最大录制码率
+                huya_max_ratio = config.get('huya_max_ratio', 0)
+
+                # 流信息
+                stream_items = json.loads(huya)['data'][0]['gameStreamInfoList']
+                # 码率信息
+                ratio_items = json.loads(huya)['vMultiStreamInfo']
+                # 最大码率(不含hdr)
+                max_ratio = json.loads(huya)['data'][0]['gameLiveInfo']['bitRate']
+
+                # 录制码率
+                record_ratio = 0
+                # 如果限制了最大码率
+                if huya_max_ratio != 0:
+                    # 挑选合适的码率
+                    for ratio_item in ratio_items:
+                        # iBitRate = 0的就是最大码率
+                        if ratio_item['iBitRate'] == 0:
+                            ratio_item['iBitRate'] = max_ratio
+                        # 不录制大于最大码率的
+                        if huya_max_ratio >= ratio_item['iBitRate'] > record_ratio:
+                            record_ratio = ratio_item['iBitRate']
+
+                    # 原画
+                    if record_ratio == max_ratio:
+                        record_ratio = 0
+
+                huyajson = stream_items[0]
+                for cdn in stream_items:
                     if cdn['sCdnType'] == huyacdn:
                         huyajson = cdn
-                absurl = f'{huyajson["sFlvUrl"]}/{huyajson["sStreamName"]}.{huyajson["sFlvUrlSuffix"]}?' \
-                        f'{huyajson["sFlvAntiCode"]}'
-                self.raw_stream_url = html.unescape(absurl) + "&ratio=" + str(ratio)
+                        break
+
+                absurl = f'{huyajson["sFlvUrl"]}/{huyajson["sStreamName"]}.{huyajson["sFlvUrlSuffix"]}?{huyajson["sFlvAntiCode"]}'
+                self.raw_stream_url = html.unescape(absurl) + "&ratio=" + str(record_ratio)
                 self.room_title = json.loads(huya)['data'][0]['gameLiveInfo']['introduction']
                 return True
             except:
@@ -69,4 +89,4 @@ class Huya(DownloadBase):
     def close(self):
         if self.huya_danmaku:
             self.danmaku.stop()
-            logger.info("结束弹幕录制")
+            logger.info("结束弹幕录制")

biliup/plugins/inke.py

@@ -15,7 +15,7 @@ class inke(DownloadBase):
     def __init__(self, fname, url, suffix='flv'):
         super().__init__(fname, url, suffix)
 
-    def check_stream(self):
+    def check_stream(self, is_check=False):
         logger.debug(self.fname)
         rid = re.search(r'uid=([a-zA-Z0-9]+)', self.url).group(1)
         r1 = requests.get(

biliup/plugins/kuaishou.py

@@ -13,7 +13,7 @@ class Kuaishou(DownloadBase):
     def __init__(self, fname, url, suffix='flv'):
         super().__init__(fname, url, suffix)
 
-    def check_stream(self):
+    def check_stream(self, is_check=False):
         murl = ""
         nurl = f"https://live.kuaishou.com/u/{get_kwaiId(self.url)}"
 

biliup/plugins/missevan.py

@@ -16,7 +16,7 @@ class Missevan(DownloadBase):
             'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64; rv:106.0) Gecko/20100101 Firefox/106.0'
         }
 
-    def check_stream(self):
+    def check_stream(self, is_check=False):
         headers = {
             'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64; rv:106.0) Gecko/20100101 Firefox/106.0'
         }

biliup/plugins/nico.py

@@ -20,7 +20,7 @@ class Nico(DownloadBase):
     def __init__(self, fname, url, suffix='flv'):
         super().__init__(fname, url, suffix)
 
-    def check_stream(self):
+    def check_stream(self, is_check=False):
         try:
             response = requests.get(self.url)
             # 正则表达式

biliup/plugins/now.py

@@ -14,7 +14,7 @@ class now(DownloadBase):
     def __init__(self, fname, url, suffix='flv'):
         super().__init__(fname, url, suffix)
 
-    def check_stream(self):
+    def check_stream(self, is_check=False):
         logger.debug(self.fname)
         rid = re.search(r'roomid=([a-zA-Z0-9]+)', self.url).group(1)
         r1 = requests.get(

biliup/plugins/twitch.py

@@ -1,6 +1,9 @@
 import random
+import re
 import subprocess
 import time
+from typing import Generator
+from urllib.parse import urlencode
 
 import requests
 import yt_dlp
@@ -8,79 +11,45 @@ import yt_dlp
 from . import logger
 from ..engine.decorators import Plugin
 from ..engine.download import DownloadBase
-from ..plugins import BatchCheckBase
 from biliup.config import config
 from biliup.plugins.Danmaku import DanmakuClient
 
 VALID_URL_BASE = r'(?:https?://)?(?:(?:www|go|m)\.)?twitch\.tv/(?P<id>[0-9_a-zA-Z]+)'
-_OPERATION_HASHES = {
-    'CollectionSideBar': '27111f1b382effad0b6def325caef1909c733fe6a4fbabf54f8d491ef2cf2f14',
-    'FilterableVideoTower_Videos': 'a937f1d22e269e39a03b509f65a7490f9fc247d7f83d6ac1421523e3b68042cb',
-    'ClipsCards__User': 'b73ad2bfaecfd30a9e6c28fada15bd97032c83ec77a0440766a56fe0bd632777',
-    'ChannelCollectionsContent': '07e3691a1bad77a36aba590c351180439a40baefc1c275356f40fc7082419a84',
-    'StreamMetadata': '1c719a40e481453e5c48d9bb585d971b8b372f8ebb105b17076722264dfa5b3e',
-    'ComscoreStreamingQuery': 'e1edae8122517d013405f237ffcc124515dc6ded82480a88daef69c83b53ac01',
-    'VideoPreviewOverlay': '3006e77e51b128d838fa4e835723ca4dc9a05c5efd4466c1085215c6e437e65c',
-}
+VALID_URL_VIDEOS = r'https?://(?:(?:www|go|m)\.)?twitch\.tv/(?P<id>[^/]+)/(?:videos|profile|clips)'
 _CLIENT_ID = 'kimne78kx3ncx6brgo4mv6wki5h1ko'
 
+# Twitch 授权信息是否到期
+AUTH_EXPIRE_STATUS = False
 
-@Plugin.download(regexp=r'https?://(?:(?:www|go|m)\.)?twitch\.tv/(?P<id>[^/]+)/(?:videos|profile|clips)')
+@Plugin.download(regexp=VALID_URL_VIDEOS)
 class TwitchVideos(DownloadBase):
     def __init__(self, fname, url, suffix='mp4'):
         DownloadBase.__init__(self, fname, url, suffix=suffix)
+        self.is_download = True
+        self.cookiejarFile = config.get('user', {}).get('twitch_cookie_file')
 
-    def check_stream(self):
+    def check_stream(self, is_check=False):
+        # TODO 这里原本的批量检测是有问题的 先用yt_dlp实现 等待后续新增新的批量检测方式 后续这里的auth信息和直播一样采用twitch_cookie
         with yt_dlp.YoutubeDL({'download_archive': 'archive.txt'}) as ydl:
-            info = ydl.extract_info(self.url, download=False)
+            try:
+                info = ydl.extract_info(self.url, download=False, process=False)
+            except:
+                logger.warning(f"{self.url}：获取错误，本次跳过")
+                return False
             for entry in info['entries']:
                 if ydl.in_download_archive(entry):
                     continue
-                self.raw_stream_url = entry['url']
-                self.room_title = entry.get('title')
-                ydl.record_download_archive(entry)
+                if not is_check:
+                    download_info = ydl.extract_info(entry['url'], download=False)
+                    self.raw_stream_url = download_info['url']
+                    for thumbnail in download_info['thumbnails']:
+                        if 'preference' in thumbnail and thumbnail['preference'] == 1:
+                            self.live_cover_url = thumbnail['url']
+                            break
+                    self.room_title = entry['title']
+                    ydl.record_download_archive(entry)
                 return True
-
-    class BatchCheck(BatchCheckBase):
-        def __init__(self, urls):
-            BatchCheckBase.__init__(self, pattern_id=VALID_URL_BASE, urls=urls)
-
-        def check(self):
-            with yt_dlp.YoutubeDL({'download_archive': 'archive.txt'}) as ydl:
-                for channel_name, url in self.not_live():
-                    info = ydl.extract_info(url, download=False, process=False)
-                    for entry in info['entries']:
-                        if ydl.in_download_archive(entry):
-                            continue
-                        yield url
-                    time.sleep(10)
-                    # ydl.record_download_archive(entry)
-
-        def not_live(self):
-            gql = self.get_streamer()
-            i = -1
-            for data in gql:
-                i += 1
-                user = data['data'].get('user')
-                if not user:
-                    continue
-                stream = user['stream']
-                if not stream:
-                    yield self.usr_list[i], self.usr_dict.get(self.usr_list[i])
-
-        def get_streamer(self):
-            for channel_name in self.usr_list:
-                op = {
-                    'operationName': 'StreamMetadata',
-                    'variables': {'channelLogin': channel_name.lower()}
-                }
-                op['extensions'] = {
-                    'persistedQuery': {
-                        'version': 1,
-                        'sha256Hash': _OPERATION_HASHES[op['operationName']],
-                    }
-                }
-                yield get_streamer(op)
+        return False
 
 
 @Plugin.download(regexp=VALID_URL_BASE)
@@ -91,12 +60,44 @@ class Twitch(DownloadBase):
         self.twitch_disable_ads = config.get('twitch_disable_ads', True)
         self.proc = None
 
-    def check_stream(self):
-        if not list(Twitch.BatchCheck([self.url]).check()):
-            return
-        gql = Twitch.BatchCheck([self.url]).get_streamer()
-        for data in gql:
-            self.room_title = data.get('data').get('user').get('lastBroadcast').get('title')
+    def check_stream(self, is_check=False):
+        channel_name = re.match(VALID_URL_BASE, self.url).group('id').lower()
+        user = post_gql({
+            "query": '''
+                query query($channel_name:String!) {
+                    user(login: $channel_name){
+                        stream {
+                            id
+                            title
+                            type
+                            previewImageURL(width: 0,height: 0)
+                            playbackAccessToken(
+                                params: {
+                                    platform: "web",
+                                    playerBackend: "mediaplayer",
+                                    playerType: "site"
+                                }
+                            ) {
+                                signature
+                                value             
+                            }
+                        }
+                    }
+                }
+            ''',
+            'variables': {'channel_name': channel_name}
+        }).get('data', {}).get('user')
+        if not user:
+            logger.warning(f"{Twitch.__name__}: {self.url}: 获取错误，本次跳过")
+            return False
+        elif not user['stream'] or user['stream']['type'] != 'live':
+            return False
+
+        self.room_title = user['stream']['title']
+        self.live_cover_url = user['stream']['previewImageURL']
+        if is_check:
+            return True
+
         if self.downloader == 'ffmpeg':
             port = random.randint(1025, 65535)
             stream_shell = [
@@ -110,28 +111,63 @@ class Twitch(DownloadBase):
             ]
             if self.twitch_disable_ads:  # 去广告，去掉、跳过嵌入的广告流
                 stream_shell.insert(1, "--twitch-disable-ads")
+
             twitch_cookie = config.get('user', {}).get('twitch_cookie')
-            if twitch_cookie:
-                twitch_cookie = "--twitch-api-header=Authorization=OAuth " + twitch_cookie
-                stream_shell.insert(1, twitch_cookie)
+            # 在设置且有效的情况下使用
+            if twitch_cookie and not AUTH_EXPIRE_STATUS:
+                stream_shell.insert(1, "--twitch-api-header=Authorization=OAuth " + twitch_cookie)
+
             self.proc = subprocess.Popen(stream_shell)
             self.raw_stream_url = f"http://localhost:{port}"
             i = 0
             while i < 5:
                 if not (self.proc.poll() is None):
-                    return
+                    return False
                 time.sleep(1)
                 i += 1
             return True
-        with yt_dlp.YoutubeDL() as ydl:
-            try:
-                info = ydl.extract_info(self.url, download=False)
-            except yt_dlp.utils.DownloadError as e:
-                logger.warning(self.url, exc_info=e)
-                return
-            self.raw_stream_url = info['formats'][-1]['url']
+        else:
+            query = {
+                "player": "twitchweb",
+                "p": random.randint(1000000, 10000000),
+                "allow_source": "true",
+                "allow_audio_only": "true",
+                "allow_spectre": "false",
+                'fast_bread': "true",
+                'sig': user.get('stream').get('playbackAccessToken').get('signature'),
+                'token': user.get('stream').get('playbackAccessToken').get('value'),
+            }
+            self.raw_stream_url = f'https://usher.ttvnw.net/api/channel/hls/{channel_name}.m3u8?{urlencode(query)}'
             return True
 
+    @staticmethod
+    def batch_check(check_urls: list[str]) -> Generator[str, None, None]:
+        ops = []
+        for url in check_urls:
+            channel_name = re.match(VALID_URL_BASE, url).group('id')
+            op = {
+                "query": '''
+                    query query($login:String!) {
+                        user(login: $login){
+                            stream {
+                              type
+                            }
+                        }
+                    }
+                ''',
+                'variables': {'login': channel_name.lower()}
+            }
+            ops.append(op)
+        gql = post_gql(ops)
+        for index, data in enumerate(gql):
+            user = data.get('data', {}).get('user')
+            if not user:
+                logger.warning(f"{Twitch.__name__}: {check_urls[index]}: 获取错误，本次跳过")
+                continue
+            elif not user['stream'] or user['stream']['type'] != 'live':
+                continue
+            yield check_urls[index]
+
     async def danmaku_download_start(self, filename):
         if self.twitch_danmaku:
             logger.info("开始弹幕录制")
@@ -148,47 +184,28 @@ class Twitch(DownloadBase):
         except:
             logger.exception(f'terminate {self.fname} failed')
 
-    class BatchCheck(BatchCheckBase):
-        def __init__(self, urls):
-            BatchCheckBase.__init__(self, pattern_id=VALID_URL_BASE, urls=urls)
 
-        def check(self):
-            gql = self.get_streamer()
-            i = -1
-            for data in gql:
-                i += 1
-                user = data['data'].get('user')
-                if not user:
-                    continue
-                stream = user['stream']
-                if not stream:
-                    continue
-                yield self.usr_dict.get(self.usr_list[i])
+def post_gql(ops):
+    global AUTH_EXPIRE_STATUS
+    headers = {
+        'Content-Type': 'text/plain;charset=UTF-8',
+        'Client-ID': _CLIENT_ID,
+    }
+    twitch_cookie = config.get('user', {}).get('twitch_cookie')
+    if not AUTH_EXPIRE_STATUS and twitch_cookie:
+        headers['Authorization'] = f'OAuth {twitch_cookie}'
 
-        def get_streamer(self):
-            ops = []
-            for channel_name in self.usr_list:
-                op = {
-                    'operationName': 'StreamMetadata',
-                    'variables': {'channelLogin': channel_name.lower()}
-                }
-                op['extensions'] = {
-                    'persistedQuery': {
-                        'version': 1,
-                        'sha256Hash': _OPERATION_HASHES[op['operationName']],
-                    }
-                }
-                ops.append(op)
-            return get_streamer(ops)
-
-
-def get_streamer(ops):
     gql = requests.post(
         'https://gql.twitch.tv/gql',
         json=ops,
-        headers={
-            'Content-Type': 'text/plain;charset=UTF-8',
-            'Client-ID': _CLIENT_ID,
-        }, timeout=15)
+        headers=headers,
+        timeout=15)
     gql.close()
-    return gql.json()
+    data = gql.json()
+
+    if isinstance(data, dict) and data.get('error') == 'Unauthorized':
+        AUTH_EXPIRE_STATUS = True
+        logger.warning("Twitch Cookie已失效请及时更换，之后操作将忽略Twitch Cookie")
+        return post_gql(ops)
+
+    return data

biliup/plugins/youtube.py

@@ -2,24 +2,23 @@ import asyncio
 import random
 import subprocess
 import threading
-from urllib.parse import urlparse
 
+import yaml
 import yt_dlp
 import streamlink
 import os
 import time
 from PIL import Image
-from yt_dlp.utils import MaxDownloadsReached
 
-from yt_dlp.utils import DateRange
+from yt_dlp.utils import DateRange, UserNotLive
 from biliup.config import config
-from biliup.plugins.Danmaku import DanmakuClient
 from ..engine.decorators import Plugin
-from ..engine.download import DownloadBase, get_valid_filename, stream_gears_download
+from ..engine.download import DownloadBase, get_valid_filename
 from . import logger
 
 VALID_URL_BASE = r'https?://(?:(?:www|m)\.)?youtube\.com/(?P<id>.*?)\??(.*?)'
 
+
 @Plugin.download(regexp=VALID_URL_BASE)
 class Youtube(DownloadBase):
     def __init__(self, fname, url, suffix='webm'):
@@ -27,15 +26,18 @@ class Youtube(DownloadBase):
         self.use_new_ytb_downloader = config.get('use_new_ytb_downloader', False)
         self.ytb_danmaku = config.get('ytb_danmaku', False)
         self.cookiejarFile = config.get('user', {}).get('youtube_cookie')
-        self.vcodec = config.get('youtube_prefer_vcodec','av01|vp9|avc')
-        self.acodec = config.get('youtube_prefer_acodec','opus|mp4a')
-        self.resolution = config.get('youtube_max_resolution','4320')
-        self.filesize = config.get('youtube_max_videosize','100G')
-        self.beforedate = config.get('youtube_before_date','20770707')
-        self.afterdate = config.get('youtube_after_date','19700101')
+        self.vcodec = config.get('youtube_prefer_vcodec', 'av01|vp9|avc')
+        self.acodec = config.get('youtube_prefer_acodec', 'opus|mp4a')
+        self.resolution = config.get('youtube_max_resolution', '4320')
+        self.filesize = config.get('youtube_max_videosize', '100G')
+        self.beforedate = config.get('youtube_before_date', '20770707')
+        self.afterdate = config.get('youtube_after_date', '19700101')
         self.use_youtube_cover = config.get('use_youtube_cover', False)
+        # 需要下载的url
+        self.download_url = None
 
-    def check_stream(self):
+    def check_stream(self, is_check=False):
+        self.download_url = None
         if self.use_new_ytb_downloader and self.downloader == 'ffmpeg':
             _fname, self.room_title = self.get_stream_info(self.url)
             port = random.randint(1025, 65535)
@@ -50,40 +52,74 @@ class Youtube(DownloadBase):
             i = 0
             while i < 5:
                 if not (self.proc.poll() is None):
-                    return
+                    return False
                 time.sleep(1)
                 i += 1
             return True
         else:
-            with yt_dlp.YoutubeDL({'download_archive': 'archive.txt', 'ignoreerrors': True, 'extract_flat': True,
-                                   'cookiefile': self.cookiejarFile}) as ydl:
-                info = ydl.extract_info(self.url, download=False, process=False)
-                if info is None:
-                    logger.warning(self.cookiejarFile)
-                    logger.warning(self.url)
+            with yt_dlp.YoutubeDL({'download_archive': 'archive.txt', 'cookiefile': self.cookiejarFile,
+                                   'format': f"bestvideo[vcodec~='^({self.vcodec})'][height<={self.resolution}][filesize<{self.filesize}]+bestaudio[acodec~='^({self.acodec})']/best[height<={self.resolution}]/best",
+                                   }) as ydl:
+                try:
+                    info = ydl.extract_info(self.url, download=False, process=False)
+                except UserNotLive:
                     return False
-                if '_type' in info:
-                    # /live 形式链接取标题
-                    if info['_type'] in 'url' and info['webpage_url_basename'] in 'live':
-                        live = ydl.extract_info(info['url'], download=False, process=False)
-                        self.room_title = live['title']
-                    # Playlist 暂时返回列表名
-                    if info['_type'] in 'playlist':
-                        self.room_title = info['title']
-                else:
-                    # 视频取标题
-                    self.room_title = info['title']
+                except:
+                    logger.warning(f"{self.url}：获取错误，本次跳过")
+                    return False
+
+                # 视频取标题
+                self.room_title = info['title']
+
                 if info.get('entries') is None:
                     if ydl.in_download_archive(info):
                         return False
+                    self.download_url = self.url
                     return True
+
+                # 时间范围缓存避免每次都要读取
+                cache_save_dir = './cache'
+                cache_filename = f'{cache_save_dir}/yt_dlp_cache.yaml'
+                if not os.path.exists(cache_save_dir):
+                    os.makedirs(cache_save_dir)
+                if not os.path.exists(cache_filename):
+                    with open(cache_filename, 'w') as f:
+                        f.close()
+
+                cache = yaml.load(open(cache_filename, 'r'), Loader=yaml.FullLoader)
+                if cache is None:
+                    cache = {}
+
+                self.is_download = True
+
                 for entry in info['entries']:
-                    # 取 Playlist 内视频标题
-                    self.room_title = entry['title']
+                    # 检测是否已下载
                     if ydl.in_download_archive(entry):
                         continue
-                    # ydl.record_download_archive()
-                    return True
+
+                    try:
+                        # 检测时间范围
+                        if entry['id'] not in cache:
+                            cache[entry['id']] = {}
+                        upload_date = cache.get(entry['id']).get('upload_date')
+                        if upload_date is None:
+                            upload_date = ydl.extract_info(entry['url'], download=False).get('upload_date')
+
+                        cache[entry['id']]['upload_date'] = upload_date
+                        if upload_date not in DateRange(self.afterdate, self.beforedate):
+                            continue
+                    except:
+                        continue
+
+                    # 取 Playlist 内视频标题
+                    self.room_title = entry['title']
+                    self.download_url = entry['url']
+                    break
+
+                with open(cache_filename, "w") as f:
+                    yaml.dump(cache, f, encoding='utf-8', allow_unicode=True)
+
+                return self.download_url is not None
 
     def download(self, filename):
         if self.use_new_ytb_downloader and self.downloader == 'ffmpeg':
@@ -99,44 +135,39 @@ class Youtube(DownloadBase):
         else:
             try:
                 ydl_opts = {
-                    'outtmpl': filename+ '.%(ext)s',
+                    'outtmpl': filename + '.%(ext)s',
                     'format': f"bestvideo[vcodec~='^({self.vcodec})'][height<={self.resolution}][filesize<{self.filesize}]+bestaudio[acodec~='^({self.acodec})']/best[height<={self.resolution}]/best",
                     'cookiefile': self.cookiejarFile,
                     # 'proxy': proxyUrl,
-                    'ignoreerrors': True,
-                    'max_downloads': 1,
-                    'daterange' : DateRange(self.afterdate, self.beforedate),
                     'break_on_reject': True,
                     'download_archive': 'archive.txt',
-                    'lazy_playlist': config.get('youtube_lazy_playlist', False)
                 }
-                if self.use_youtube_cover is True:
+                if self.use_youtube_cover:
                     ydl_opts['writethumbnail'] = True
                 with yt_dlp.YoutubeDL(ydl_opts) as ydl:
-                    ydl.download([self.url])
-            except MaxDownloadsReached:
-                save_dir = f'cover/youtube/'
-                webp_cover_path = f'{filename}.webp'
-                jpg_cover_path = f'{filename}.jpg'
-                if os.path.exists(webp_cover_path):
-                    with Image.open(webp_cover_path) as img:
-                        img = img.convert('RGB')
-                        if not os.path.exists(save_dir):
-                            os.makedirs(save_dir)
-                        img.save(f'{save_dir}{filename}.jpg', format='JPEG')
-                    os.remove(webp_cover_path)
+                    ydl.download([self.download_url])
+
+                if self.use_youtube_cover:
+                    save_dir = f'cover/{self.__class__.__name__}/{self.fname}/'
+                    if not os.path.exists(save_dir):
+                        os.makedirs(save_dir)
+                    webp_cover_path = f'{filename}.webp'
+                    jpg_cover_path = f'{filename}.jpg'
+                    if os.path.exists(webp_cover_path):
+                        with Image.open(webp_cover_path) as img:
+                            img = img.convert('RGB')
+                            if not os.path.exists(save_dir):
+                                os.makedirs(save_dir)
+                            img.save(f'{save_dir}{filename}.jpg', format='JPEG')
+                        os.remove(webp_cover_path)
+                    elif os.path.exists(jpg_cover_path):
+                        os.rename(jpg_cover_path, f'{save_dir}{filename}.jpg')
+
                     self.live_cover_path = f'{save_dir}{filename}.jpg'
-                elif os.path.exists(jpg_cover_path):
-                    os.rename(jpg_cover_path, f'{save_dir}{filename}.jpg')
-                    self.live_cover_path = f'{save_dir}{filename}.jpg'
-                return False
-            except yt_dlp.utils.RejectedVideoReached:
-                logger.info("已下载完毕指定日期内的视频，结束本次任务")
-                return False
-            except yt_dlp.utils.DownloadError:
+            except:
                 logger.exception(self.fname)
-                return 1
-            return 0
+                return False
+            return True
 
     def get_stream_info(self, url):
         session = streamlink.Streamlink()
@@ -146,4 +177,4 @@ class Youtube(DownloadBase):
         streams._get_streams()
         author = streams.get_author()
         title = streams.get_title()
-        return author, title
+        return author, title

biliup/plugins/yy.py

@@ -10,7 +10,7 @@ class YY(DownloadBase):
     def __init__(self, fname, url, suffix='flv'):
         super().__init__(fname, url, suffix)
 
-    def check_stream(self):
+    def check_stream(self, is_check=False):
         headers = {
             'content-type': 'text/plain;charset=UTF-8',
             'referer': 'https://www.yy.com/',

public/config.toml

@@ -32,8 +32,8 @@ threads = 3
 
 #------杂项------#
 ### 检测到主播下播后延迟再次检测，单位：秒，避免特殊情况提早启动上传导致漏录
-### 当delay不存在时，默认延迟时间为0秒，最长延迟时间为30分钟(1800秒)。若设置的延迟时间超过60秒，则会启用分段检测机制，每隔60秒进行一次开播状态的检测。
-delay = 10
+### 当delay不存在时，默认延迟时间为0秒，没有快速上传的需求推荐设置5分钟(300秒)或按需设置。若设置的延迟时间超过60秒，则会启用分段检测机制，每隔60秒进行一次开播状态的检测。
+delay = 300
 ### 检测间隔时间，单位：秒
 ### event_loop_interval的值应 = 最多的相同平台直播间数 * checker_sleep + 5~10秒的余量
 ### 比如：当有5个哔哩哔哩、3个斗鱼、8个虎牙，event_loop_interval为8 * 15(checker_sleep) + 5~10秒的余量 = 125~130
@@ -50,7 +50,7 @@ check_sourcecode = 15
 
 
 ######### 各平台录播设置 #########
-### 使用直播间封面作为投稿封面。此封面优先级低于单个主播指定的自定义封面。（目前仅支持bilibili。直播封面将会保存于cover文件夹下，请自行定期删除）
+### 使用直播间封面作为投稿封面。此封面优先级低于单个主播指定的自定义封面。（目前支持bilibili,TWITCH。直播封面将会保存于cover文件夹下，请自行定期删除）
 #use_live_cover = true
 
 #------斗鱼------#
@@ -59,6 +59,11 @@ check_sourcecode = 15
 #douyucdn = "tct-h5"
 ### 录制斗鱼弹幕，默认关闭【目前暂时不支持视频按时长分段下的弹幕文件自动分段，只有使用ffmpeg（包括streamlink混合模式）作为下载器才支持】
 #douyu_danmaku = false
+### 斗鱼自选画质
+### 刚开播可能没有除了原画之外的画质 会先录制原画 后续视频分段(仅ffmpeg streamlink)时录制设置的画质
+### 0 原画,4 蓝光4m,3 超清,2 高清
+#douyu_rate = 0
+
 
 #------虎牙------#
 ### 如遇到虎牙录制卡顿可以尝试切换线路。可选以下线路
@@ -66,10 +71,20 @@ check_sourcecode = 15
 #huyacdn = "AL"
 ### 录制虎牙弹幕，默认关闭【目前暂时不支持视频按时长分段下的弹幕文件自动分段，只有使用ffmpeg（包括streamlink混合模式）作为下载器才支持】
 #huya_danmaku = false
+### 虎牙自选录制码率
+### 可以避免录制如20M的码率，每小时8G左右大小，上传及转码耗时过长。
+### 20000（蓝光20M）, 10000（蓝光10M）, 8000（蓝光8M）, 2000（超清）, 500（流畅）
+### 设置为10000则录制小于等于蓝光10M的画质
+#huya_max_ratio = 10000
 
 #------抖音------#
 ### 录制抖音弹幕，默认关闭【目前暂时不支持视频按时长分段下的弹幕文件自动分段，只有使用ffmpeg（包括streamlink混合模式）作为下载器才支持】
 #douyin_danmaku = false
+### 抖音自选画质
+### 刚开播可能没有除了原画之外的画质 会先录制原画 后续视频分段(仅ffmpeg streamlink)时录制设置的画质
+### origin 原画,uhd 蓝光,hd 超清,sd 高清,ld 标清,md 流畅
+### 没有选中的画质则会自动选择相近的画质优先低清晰度
+#douyin_quality = origin
 
 #------快手直播（中国大陆）------#
 ### 如遇到快手录制卡顿可以尝试切换线路。默认随机。可选以下线路
@@ -113,6 +128,12 @@ check_sourcecode = 15
 ### 每次会随机返回填入的其中一个线路，并且会自动判断所填入的节点是否可用
 #bili_force_cn01 = true
 #bili_force_cn01_domains = "cn-jxnc-cm-01-16.bilivideo.com,cn-gddg-cm-01-06.bilivideo.com,cn-fjqz-cm-01-06.bilivideo.com,cn-jssz-cm-01-02.bilivideo.com,cn-gddg-ct-01-12.bilivideo.com"
+### 哔哩哔哩自选画质
+### 刚开播可能没有除了原画之外的画质 会先录制原画 后续视频分段(仅ffmpeg streamlink)时录制设置的画质
+### 30000 杜比,20000 4K,10000 原画,401 蓝光(杜比),400 蓝光,250 超清,150 高清,80 流畅,0 B站默认(多数情况下是蓝光 400)
+### 没有选中的画质则会自动选择相近的画质
+#bili_qn = 10000
+
 
 #------YouTube------#
 ### 设置偏好的YouTube下载封装格式
@@ -136,9 +157,6 @@ check_sourcecode = 15
 #youtube_after_date = '20220201'
 ### 仅下载该日期之前的视频（可与上面的youtube_after_date配合使用，构成指定下载范围区间）
 #youtube_before_date = '20230501'
-### 默认情况下，会先提取完所有页面后才会启动下载视频
-### 开启后，将在视频的每个播放列表页面提取后，启动下载【默认关闭】
-#youtube_lazy_playlist = true
 ### YouTube的Streamlink下载器模式，开启之后会调用Streamlink获取直播流并且使用FFMPEG下载，默认关闭【仅当下载器是FFMPEG时才有效】
 ### 开启之后上面关于YOUTUBE的选项全部无效【我也不知道这个功能有什么屌用】
 #use_new_ytb_downloader = false
@@ -238,7 +256,7 @@ tags = [ "biliup", "视频标签" ]
 #------Twitch直播录制------#
 ### 只在录制twitch直播时才生效
 ### 如录制Twitch时遇见视频流中广告过多的情况，可尝试在此填入cookie，可以大幅减少视频流中的twitch广告（经测试需要在该Cookie所属账号开了TwitchTurbo会员才有用）
-### 该cookie有过期风险，cookie过期后会导致无法获取到直播流（我个人用了四个月都没过期）
+### 该cookie有过期风险，cookie过期后会在日志输出警告请及时更换cookie，cookie失效的情况下后续录制将忽略cookie（我个人用了四个月都没过期）
 ### twitch_cookie获取方式：在浏览器中打开Twitch.tv，F12调出控制台，在控制台中执行：document.cookie.split("; ").find(item=>item.startsWith("auth-token="))?.split("=")[1]
 ### twitch_cookie需要在downloader= "ffmpeg"时候才会生效
 #twitch_cookie = 'asdiouo2h987r23hf2893fh923y7'

public/config.yaml

@@ -32,8 +32,8 @@ threads: 3
 
 #------杂项------#
 ### 检测到主播下播后延迟再次检测，单位：秒，避免特殊情况提早启动上传导致漏录
-### 当delay不存在时，默认延迟时间为0秒，最长延迟时间为30分钟(1800秒)。若设置的延迟时间超过60秒，则会启用分段检测机制，每隔60秒进行一次开播状态的检测。
-delay: 10
+### 当delay不存在时，默认延迟时间为0秒，没有快速上传的需求推荐设置5分钟(300秒)或按需设置。若设置的延迟时间超过60秒，则会启用分段检测机制，每隔60秒进行一次开播状态的检测。
+delay: 300
 ### 检测间隔时间，单位：秒
 ### event_loop_interval的值应 = 最多的相同平台直播间数 * checker_sleep + 5~10秒的余量
 ### 比如：当有5个哔哩哔哩、3个斗鱼、8个虎牙，event_loop_interval为8 * 15(checker_sleep) + 5~10秒的余量 = 125~130
@@ -50,7 +50,7 @@ check_sourcecode: 15
 
 
 ######### 各平台录播设置 #########
-### 使用直播间封面作为投稿封面。此封面优先级低于单个主播指定的自定义封面。（直播封面将会保存于cover文件夹下，请自行定期删除）
+### 使用直播间封面作为投稿封面。此封面优先级低于单个主播指定的自定义封面。（目前支持bilibili,TWITCH。直播封面将会保存于cover文件夹下，请自行定期删除）
 #use_live_cover: true
 
 #------斗鱼------#
@@ -59,6 +59,10 @@ check_sourcecode: 15
 #douyucdn: tct-h5
 ### 录制斗鱼弹幕，默认关闭【目前暂时不支持视频按时长分段下的弹幕文件自动分段，只有使用ffmpeg（包括streamlink混合模式）作为下载器才支持】
 #douyu_danmaku: false
+### 斗鱼自选画质
+### 刚开播可能没有除了原画之外的画质 会先录制原画 后续视频分段(仅ffmpeg streamlink)时录制设置的画质
+### 0 原画,4 蓝光4m,3 超清,2 高清
+#douyu_rate:  0
 
 #------虎牙------#
 ### 如遇到虎牙录制卡顿可以尝试切换线路。可选以下线路
@@ -66,10 +70,20 @@ check_sourcecode: 15
 #huyacdn: AL
 ### 录制虎牙弹幕，默认关闭【目前暂时不支持视频按时长分段下的弹幕文件自动分段，只有使用ffmpeg（包括streamlink混合模式）作为下载器才支持】
 #huya_danmaku: false
+### 虎牙自选录制码率
+### 可以避免录制如20M的码率，每小时8G左右大小，上传及转码耗时过长。
+### 20000（蓝光20M）, 10000（蓝光10M）, 8000（蓝光8M）, 2000（超清）, 500（流畅）
+### 设置为10000则录制小于等于蓝光10M的画质
+#huya_max_ratio: 10000
 
 #------抖音------#
 ### 录制抖音弹幕，默认关闭【目前暂时不支持视频按时长分段下的弹幕文件自动分段，只有使用ffmpeg（包括streamlink混合模式）作为下载器才支持】
 #douyin_danmaku: false
+### 抖音自选画质
+### 刚开播可能没有除了原画之外的画质 会先录制原画 后续视频分段(仅ffmpeg streamlink)时录制设置的画质
+### origin 原画,uhd 蓝光,hd 超清,sd 高清,ld 标清,md 流畅
+### 没有选中的画质则会自动选择相近的画质优先低清晰度
+#douyin_quality: origin
 
 #------快手直播（中国大陆）------#
 ### 如遇到快手录制卡顿可以尝试切换线路。默认随机。可选以下线路
@@ -113,6 +127,11 @@ check_sourcecode: 15
 ### 每次会随机返回填入的其中一个线路，并且会自动判断所填入的节点是否可用
 #bili_force_cn01: false
 #bili_force_cn01_domains: 'cn-jxnc-cm-01-16.bilivideo.com,cn-gddg-cm-01-06.bilivideo.com,cn-fjqz-cm-01-06.bilivideo.com,cn-jssz-cm-01-02.bilivideo.com,cn-gddg-ct-01-12.bilivideo.com'
+### 哔哩哔哩自选画质
+### 刚开播可能没有除了原画之外的画质 会先录制原画 后续视频分段(仅ffmpeg streamlink)时录制设置的画质
+### 30000 杜比,20000 4K,10000 原画,401 蓝光(杜比),400 蓝光,250 超清,150 高清,80 流畅,0 B站默认(多数情况下是蓝光 400)
+### 没有选中的画质则会自动选择相近的画质
+#bili_qn: 10000
 
 #------YouTube------#
 ### 设置偏好的YouTube下载封装格式
@@ -136,9 +155,6 @@ check_sourcecode: 15
 #youtube_after_date: '20220201'
 ### 仅下载该日期之前的视频（可与上面的youtube_after_date配合使用，构成指定下载范围区间）
 #youtube_before_date: '20230501'
-### 默认情况下，会先提取完所有页面后才会启动下载视频
-### 开启后，将在视频的每个播放列表页面提取后，立刻启动下载【默认关闭】
-#youtube_lazy_playlist: true
 ### YouTube的Streamlink下载器模式，开启之后会调用Streamlink获取直播流并且使用FFMPEG下载，默认关闭【仅当下载器是FFMPEG时才有效】
 ### 开启之后上面关于YOUTUBE的选项全部无效【我也不知道这个功能有什么屌用】
 #use_new_ytb_downloader: false
@@ -232,7 +248,7 @@ streamers:
     #------Twitch直播录制------#
     ### 只在录制twitch直播时才生效
     ### 如录制Twitch时遇见视频流中广告过多的情况，可尝试在此填入cookie，可以大幅减少视频流中的twitch广告（经测试需要在该Cookie所属账号开了TwitchTurbo会员才有用）
-    ### 该cookie有过期风险，cookie过期后会导致无法获取到直播流（我个人用了四个月都没过期）
+    ### 该cookie有过期风险，cookie过期后会在日志输出警告请及时更换cookie，cookie失效的情况下后续录制将忽略cookie（我个人用了四个月都没过期）
     ### twitch_cookie获取方式：在浏览器中打开Twitch.tv，F12调出控制台，在控制台中执行：document.cookie.split("; ").find(item=>item.startsWith("auth-token="))?.split("=")[1]
     ### twitch_cookie需要在downloader= "ffmpeg"时候才会生效
     #twitch_cookie: 'aisjdoiuasdoihansdoh3ooi209'