From 86cbc0f15b1faab533c3bed9ff30ea101b83c863 Mon Sep 17 00:00:00 2001 From: XiaoMiku01 Date: Thu, 6 Mar 2025 17:58:18 +0800 Subject: [PATCH] =?UTF-8?q?Dev=20=E4=BF=AE=E5=A4=8D=E8=BE=B9=E5=BD=95?= =?UTF-8?q?=E8=BE=B9=E4=BC=A0=E7=9A=84=E4=B8=80=E4=BA=9Bbug=20(#1167)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit * fix * fix: 解决边录边传中默认值错误导致的异常 * fix bug test * fix: 修复边录边传功能的异常分pbug * fix bug * fix bug * fix bug * fix: bug * fixbug * fix 修复保存配置时key丢失的问题 --- biliup/app.py | 3 ++ biliup/config.py | 3 +- biliup/engine/download.py | 16 ++++++--- biliup/engine/sync_downloader.py | 10 +++--- biliup/plugins/bili_webup_sync.py | 56 +++++++++++++++++++------------ 5 files changed, 55 insertions(+), 33 deletions(-) diff --git a/biliup/app.py b/biliup/app.py index 47c3037..63f4143 100644 --- a/biliup/app.py +++ b/biliup/app.py @@ -25,6 +25,9 @@ def create_event_manager(): app.context['url_upload_count'] = {} # 正在上传的文件 用于同时上传一个url的时候过滤掉正在上传的 app.context['upload_filename'] = [] + + # 边录边传的下载器使用的map + app.context['sync_downloader_map'] = {} return app diff --git a/biliup/config.py b/biliup/config.py index ac02e1e..17eec58 100644 --- a/biliup/config.py +++ b/biliup/config.py @@ -33,7 +33,8 @@ class Config(UserDict): context = { 'url_upload_count': self.data.get('url_upload_count', {}), 'upload_filename': self.data.get('upload_filename', []), - 'PluginInfo': self.data.get('PluginInfo') + 'PluginInfo': self.data.get('PluginInfo'), + 'sync_downloader_map': self.data.get('sync_downloader_map', {}), } for con in db.execute(select(Configuration.value).where(Configuration.key == 'config')): diff --git a/biliup/engine/download.py b/biliup/engine/download.py index 11e6e16..194c209 100644 --- a/biliup/engine/download.py +++ b/biliup/engine/download.py @@ -29,7 +29,7 @@ from biliup.plugins.bili_webup_sync import BiliWebAsync from biliup.uploader import fmt_title_and_desc from .sync_downloader import SyncDownloader - +from biliup.app import context logger = logging.getLogger('biliup') @@ -123,7 +123,7 @@ class DownloadBase(ABC): max_file_size=int(self.file_size / 1024 / 1024), output_prefix=self.gen_download_filename(True), stream_info=stream_info, - file_name_callback=lambda file_name: self.__download_segment_callback(file_name)) + file_name_callback=lambda file_name: self.__download_segment_callback(file_name), database_row_id=self.database_row_id) return True # streamlink无法处理flv,所以回退到ffmpeg if self.downloader == 'streamlink' and '.flv' not in parsed_url_path: @@ -346,6 +346,7 @@ class DownloadBase(ABC): # TODO: 重试等待时间 time.sleep(3) finally: + self.close() # 下载模式跳过下播延迟检测 @@ -370,6 +371,11 @@ class DownloadBase(ABC): self.download_success_callback() # self.segment_processor_thread logger.info(f'{self.plugin_msg}: 退出下载') + + if str(self.database_row_id) in context["sync_downloader_map"]: + context["sync_downloader_map"].pop(str(self.database_row_id)) + logger.info(f"{self.plugin_msg} {self.database_row_id}: 从同步下载器列表中移除") + stream_info = { 'name': self.fname, 'url': self.url, @@ -524,16 +530,16 @@ def stream_gears_download(url, headers, file_name, segment_time=None, file_size= ) -def sync_download(stream_url, headers, segment_duration=60, max_file_size=100, output_prefix="segment", stream_info=None, file_name_callback: Callable[[str], None] = None): +def sync_download(stream_url, headers, segment_duration=60, max_file_size=100, output_prefix="segment", stream_info=None, file_name_callback: Callable[[str], None] = None, database_row_id=0): logger.info(f"启动同步下载器 max_file_size {max_file_size}MB") video_queue = queue.SimpleQueue() - # video_queue.put(b'\x00') def upload(video_queue, stream_info, stop_event: threading.Event): with SessionLocal() as db: data = get_stream_info(db, f"{stream_info['name']}") data, _ = fmt_title_and_desc({**data, "name": stream_info['name']}) stream_info.update(data) + logger.info(f"stream_info: {stream_info}") # 获取 BiliWebAsync.__init__ 的参数名 init_params = inspect.signature(BiliWebAsync.__init__).parameters # 过滤 info 中的无关键 @@ -547,7 +553,7 @@ def sync_download(stream_url, headers, segment_duration=60, max_file_size=100, o uploader = BiliWebAsync(**filtered_info, video_queue=video_queue) uploader.upload(total_size=max_file_size * 1024 * 1024, stop_event=stop_event, output_prefix=output_prefix, - file_name_callback=file_name_callback) + file_name_callback=file_name_callback, database_row_id=database_row_id) # print("上传器结束") logger.info(f"{stream_info['name']} 上传器结束") # video_queue = queue.SimpleQueue() diff --git a/biliup/engine/sync_downloader.py b/biliup/engine/sync_downloader.py index 395c6f6..e6bd51c 100644 --- a/biliup/engine/sync_downloader.py +++ b/biliup/engine/sync_downloader.py @@ -58,7 +58,7 @@ class SyncDownloader: self.quality = "best" self.headers = headers self.segment_duration = segment_duration - self.read_block_size = 4096 + self.read_block_size = 500 self.max_file_size = max_file_size self.output_prefix = output_prefix @@ -70,13 +70,13 @@ class SyncDownloader: # print("[run] 启动 ffmpeg...") logging.info("[run] 启动 ffmpeg...") if output_filename == "-": - data = ffmpeg_proc.stdout.read(1024) # 读取第一个 data + data = ffmpeg_proc.stdout.read(self.read_block_size) # 读取第一个 data if not data: # 如果第一个 data 为空 logging.info("[run] ffmpeg 没有输出数据，返回 False") return False self.video_queue.put(data) # 将第一个数据放入队列 while True: - data = ffmpeg_proc.stdout.read(1024) + data = ffmpeg_proc.stdout.read(self.read_block_size) if not data: logging.info("[run] ffmpeg stdout 已到达 EOF。结束本段写入。") break @@ -92,7 +92,7 @@ class SyncDownloader: if output_filename == "-": logging.info("[run] 读取 ffmpeg stdout...") # 读取第一个数据 - data = ffmpeg_proc.stdout.read(1024) + data = ffmpeg_proc.stdout.read(self.read_block_size) if not data: # 如果第一个数据为空 logging.info("[run] ffmpeg 没有输出数据，返回 False") streamlink_proc.kill() # 终止 streamlink 进程 @@ -102,7 +102,7 @@ class SyncDownloader: self.video_queue.put(data) # 将第一个数据放入队列 # 继续读取剩余数据 while True: - data = ffmpeg_proc.stdout.read(1024) + data = ffmpeg_proc.stdout.read(self.read_block_size) if not data: logging.info("[run] ffmpeg stdout 已到达 EOF。结束本段写入。") break diff --git a/biliup/plugins/bili_webup_sync.py b/biliup/plugins/bili_webup_sync.py index 1f047b9..d1e149f 100644 --- a/biliup/plugins/bili_webup_sync.py +++ b/biliup/plugins/bili_webup_sync.py @@ -19,7 +19,8 @@ from typing import Callable, Dict, Union, Any, List from urllib import parse from urllib.parse import quote -import aiohttp +# import aiohttp +from biliup.app import context from concurrent.futures.thread import ThreadPoolExecutor import concurrent import requests.utils @@ -74,14 +75,15 @@ class BiliWebAsync(UploadBase): self.user_cookie = user_cookie self.video_queue: queue.SimpleQueue = video_queue - def upload(self, total_size: int, stop_event: threading.Event, output_prefix: str, file_name_callback: Callable[[str], None] = None) -> List[UploadBase.FileInfo]: + def upload(self, total_size: int, stop_event: threading.Event, output_prefix: str, file_name_callback: Callable[[str], None] = None, database_row_id=0) -> List[UploadBase.FileInfo]: # print("开始同步上传") - logger.info("开始同步上传") + logger.info(f"开始同步上传 {database_row_id}") file_index = 1 videos = Data() bili = BiliBili(videos) - bili.login(self.persistence_path, self.user_cookie) + bili.database_row_id = database_row_id + bili.login(self.persistence_path, self.user_cookie) videos.title = self.data["format_title"][:80] # 稿件标题限制80字 if self.credits: videos.desc_v2 = self.creditsToDesc_v2() @@ -225,6 +227,8 @@ class BiliBili: if self.save_dir and not os.path.exists(self.save_dir): os.makedirs(self.save_dir) + self.database_row_id = 0 + def myinfo(self, cookies: dict = None): if cookies: requests.utils.add_dict_to_cookiejar(self.__session.cookies, cookies) @@ -399,6 +403,9 @@ class BiliBili: f"https://member.bilibili.com/preupload?{self._auto_os['query']}", params=query, timeout=5) ret = resp.json() + if "chunk_size" not in ret: + stop_event.set() + return logger.debug(f"preupload: {ret}") if preferred_upos_cdn: # 如果返回的endpoint不在probe_url中，则尝试在endpoints中校验probe_url是否可用 @@ -411,15 +418,20 @@ class BiliBili: else: logger.warning(f"选择的线路 {self._auto_os['os']} 没有返回对应 endpoint，不做修改") video_part = asyncio.run(upload(stream_queue, file_name, total_size, ret)) - if not video_part: + if video_part is None: stop_event.set() # print("异常流 直接退出") return video_part['title'] = video_part['title'][:80] - videos.append(video_part) # 添加已经上传的视频 + if str(self.database_row_id) in context["sync_downloader_map"]: + context_data = context["sync_downloader_map"][str(self.database_row_id)].copy() + context_data.pop('subtitle', None) + videos = Data(**context_data) + + videos.append(video_part) # 添加已经上传的视频 edit = False if videos.aid is None else True - ret = self.submit(submit_api=submit_api, edit=edit) + ret = self.submit(submit_api=submit_api, edit=edit, videos=videos) # logger.info(f"上传成功: {ret}") if edit: logger.info(f"编辑添加成功: {ret}") @@ -427,7 +439,8 @@ class BiliBili: logger.info(f"上传成功: {ret}") aid = ret['data']['aid'] videos.aid = aid - + context['sync_downloader_map'][str(self.database_row_id)] = videos.__dict__ + logger.info(f"上传完成 {file_name} {context['sync_downloader_map'][str(self.database_row_id)] }") if file_name_callback: file_name_callback(self.save_path) @@ -500,7 +513,7 @@ class BiliBili: } for i in range(chunks)] parts.extend(results) - if len(parts) == 0: + if n == 0: return None logger.info(f"{file_name} - total_size: {total_size}, n: {n}") cost = time.perf_counter() - start @@ -511,8 +524,8 @@ class BiliBili: 'output': 'json', 'profile': 'ugcupos/bup' } - attempt = 0 - while attempt <= 5: # 一旦放弃就会丢失前面所有的进度，多试几次吧 + attempt = 1 + while attempt <= 3: # 一旦放弃就会丢失前面所有的进度，多试几次吧 try: r = self.__session.post(url, params=p, json={"parts": parts}, headers=headers, timeout=15).json() if r.get('OK') == 1: @@ -520,9 +533,9 @@ class BiliBili: return {"title": splitext(file_name)[0], "filename": splitext(basename(upos_uri))[0], "desc": ""} raise IOError(r) except IOError: + logger.info(f"请求合并分片 {file_name} 时出现问题，尝试重连，次数：" + str(attempt)) attempt += 1 - logger.info(f"请求合并分片时出现问题，尝试重连，次数：" + str(attempt)) - time.sleep(15) + time.sleep(10) pass def upload_chunk_thread(self, url, chunk, params_clone, headers, file_name, max_retries=3, backoff_factor=1): @@ -631,12 +644,10 @@ class BiliBili: save_file and save_file.close() yield None - def submit(self, submit_api=None, edit=False): - if not self.video.title: - self.video.title = self.video.videos[0]["title"] + def submit(self, submit_api=None, edit=False, videos=None): # 不能提交 extra_fields 字段，提前处理 - post_data = asdict(self.video) + post_data = asdict(videos) if post_data.get('extra_fields'): for key, value in json.loads(post_data.pop('extra_fields')).items(): post_data.setdefault(key, value) @@ -660,6 +671,7 @@ class BiliBili: if submit_api == 'web': ret = self.submit_web(post_data, edit=edit) if ret["code"] == 21138: + time.sleep(5) logger.info(f'改用客户端接口提交{ret}') submit_api = 'client' if submit_api == 'client': @@ -744,9 +756,9 @@ class BiliBili: :param upvideo: :return: 返回官方推荐的tag """ - url = f'https://member.bilibili.com/x/web/archive/tags?' \ - f'typeid={typeid}&title={quote(upvideo["title"])}&filename=filename&desc={desc}&cover={cover}' \ - f'&groupid={groupid}&vfea={vfea}' + url = f'https://member.bilibili.com/x/web/archive/tags?' + f'typeid={typeid}&title={quote(upvideo["title"])}&filename=filename&desc={desc}&cover={cover}' + f'&groupid={groupid}&vfea={vfea}' return self.__session.get(url=url, timeout=5).json() def __enter__(self): @@ -781,8 +793,8 @@ class Data: open_subtitle: InitVar[bool] = False dolby: int = 0 hires: int = 0 - no_reprint = 0 - open_elec = 0 + no_reprint: int = 0 + open_elec: int = 0 extra_fields: str = "" aid: int = None