Dev 修复边录边传的一些bug (#1167)
* fix * fix: 解决边录边传中默认值错误导致的异常 * fix bug test * fix: 修复边录边传功能的异常分pbug * fix bug * fix bug * fix bug * fix: bug * fixbug * fix 修复保存配置时key丢失的问题
This commit is contained in:
1 parent
2e28d7744d
commit
86cbc0f15b
5 files changed
+55
-33
No files matched your search
@@ -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
|
||||
|
||||
|
||||
|
||||
+2
-1
@@ -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')):
|
||||
|
||||
@@ -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()
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -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
|
||||
|
||||
Reference in new issue
Block a user