okooooooo/biliup
1
0 Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Dev 适配 biliup/biliup-rs#186 测试边录边上传功能 (#1140)

Browse Source 
* fix

* 适配 biliup/biliup-rs#186  extra_fields

* dev 边录边上传功能初版

* 修复下载器错误

* 实现同步编辑投稿功能

* 上传模板标题格式化

* fix

* 默认分p大小2G

* 重写了对列

* 修改输出队列机制

* 多线程上传有bug

* 修复异常阻塞和多线程上传bug

* reset

* fix

* 修复线程池异常

* 添加额外字段支持；增加同步下载器选项；去除无用日志；修复B站FLV流小概率无法获取；去掉B站弹幕重复登录态验证。

* 删除补0测戳

* 调整 ffmpeg header 参数位置

* 添加同步保存到本地的配置项 和 运行分段后自定义命令

* 添加缺失的投稿参数

* 添加边录边传保存目录项

* 格式化

* 修正 webup_sync 上传线路选择

* 修复 webup_sync 的 submit_web

* 修正参数错误

* 删除多余代码

* 删除多余代码

---------

Co-authored-by: xxxxuanran <xxxxuanran@gmail.com>
This commit is contained in:
XiaoMiku01andxxxxuanran authored and GitHub committed 2025-02-03 11:03:44 +00:00
1 parent 694d05fac0
commit 94b404c57e
16 files changed
+1371 -88
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

app/lib/api-streamer.ts
+1
View File
@@ -88,6 +88,7 @@ export interface StudioEntity {
open_elec: number;
credits: Credit[];
uploader: string;
extra_fields?: string;
}
export interface LiveStreamerEntity {
app/ui/TemplateFields.tsx
+118 -2
View File
@@ -1,4 +1,4 @@
import React, { useState, useMemo, useEffect } from 'react'
import React, { useState, useMemo, useEffect, useCallback } from 'react'
import { FormFCChild } from '@douyinfe/semi-ui/lib/es/form'
import {
IconChevronDown,
@@ -18,6 +18,8 @@ import {
Notification,
ScrollList,
ScrollItem,
// TextArea,
useFormState,
} from '@douyinfe/semi-ui'
import useSWR from 'swr'
import { BiliType, fetcher, StudioEntity } from '../lib/api-streamer'
@@ -181,7 +183,36 @@ const TemplateFields: React.FC<FormFCChild<StudioEntity & { isDtime: boolean }>>
setSelectedMinutes(Math.floor(minutes / 5) * 5)
}
}
}, [formApi])
}, [formApi]);
// useEffect(() => {
// const extraFieldsList = formApi.getValue('extra_fields_list') as Array<{key: string; value: string}>;
// if (!extraFieldsList?.length) return;
// const extraFieldsObj = extraFieldsList.reduce((tempObj: Record<string, string | number>, item) => {
// if (item?.key && item?.value !== undefined) {
// tempObj[item.key] = !isNaN(Number(item.value)) ? Number(item.value) : item.value;
// }
// return tempObj;
// }, {});
// formApi.setValue('extra_fields', JSON.stringify(extraFieldsObj));
// }, []);
// useEffect(() => {
// const extraFields = formApi.getValue('extra_fields');
// if (extraFields) {
// try {
// const parsedFields = JSON.parse(extraFields);
// const fieldsList = Object.entries(parsedFields).map(([key, value]) => ({
// key,
// value: String(value)
// }));
// formApi.setValue('extra_fields_list', fieldsList);
// } catch (e) {
// console.error('解析 extra_fields 失败:', e);
// }
// }
// }, []);
return (
<>
@@ -290,6 +321,91 @@ const TemplateFields: React.FC<FormFCChild<StudioEntity & { isDtime: boolean }>>
maxCount={2000}
showClear
/>
<TextArea
style={{ maxWidth: 560 }}
field="extra_fields"
label="额外字段"
placeholder="Json格式，示例：{key: 'value'}"
autosize
maxCount={2000}
showClear
rules={[
{
validator: (rule, value) => {
if (!value) return true;
try {
JSON.parse(value);
return true;
} catch (e) {
return false;
}
},
message: '请输入正确的Json格式文本',
}
]}
/>
{/* <ArrayField
field='extra_fields_list'
initValue={(() => {
const extraFields = formApi.getValue('extra_fields');
if (!extraFields) return [];
try {
const parsedFields = JSON.parse(extraFields);
return Object.entries(parsedFields).map(([key, value]) => ({
key,
value: String(value)
}));
} catch (e) {
console.error('解析 extra_fields 失败:', e);
return [];
}
})()}
>
{({ add, arrayFields }) => (
<React.Fragment>
<Button onClick={add} icon={<IconPlusCircle />} theme='light'>Add new line</Button>
{
arrayFields.map(({ field, key, remove }, i) => (
<div key={key} style={{ width: 1000, display: 'flex' }}>
<Input
field={`${field}.key`}
label={`${field}.key`}
style={{ width: 200, marginRight: 16 }}
rules={[
{
validator: (rule, value) => {
if (!value) return true;
const currentFields = (formApi.getValue('extra_fields_list') || []) as Array<{key: string; value: string}>;
const keyCount = currentFields.filter(
(item: any, index: number) =>
item.key === value &&
`extra_fields_list.${index}.key` !== field + '.key'
).length;
if (keyCount > 1) {
throw new Error('键名不能重复');
}
return true;
}
}
]}
/>
<Input
field={`${field}.value`}
label={`${field}.value`}
style={{ width: 200, marginRight: 16 }}
/>
<Button
type='danger'
theme='borderless'
icon={<IconMinusCircle />}
onClick={remove}
style={{ margin: 12 }}
/>
</div>
))}
</React.Fragment>
)}
</ArrayField> */}
<div style={{ display: 'flex', alignItems: 'center', color: 'var(--semi-color-tertiary)' }}>
<Switch
app/ui/plugins/global.tsx
+54 -6
View File
@@ -1,10 +1,12 @@
'use client'
import React from 'react'
import React, { useEffect } from 'react'
import styles from '../../styles/dashboard.module.scss'
import { Form, Select, Space } from '@douyinfe/semi-ui'
import { Form, Select, Space, useFormApi } from '@douyinfe/semi-ui'
import { IconUpload, IconDownload } from '@douyinfe/semi-icons'
const Global: React.FC = () => {
const formApi = useFormApi()
return (
<>
{/* 全局下载 */}
@@ -28,17 +30,19 @@ const Global: React.FC = () => {
label="下载插件（downloader）"
field="downloader"
placeholder="stream-gears（默认）"
maxTagCount={3}
// initValue="stream-gears"
extraText={
<div style={{ fontSize: '14px' }}>
选择全局默认的下载插件, 可选:
<br />
1. streamlink（仅限 hls 流，不支持的流将回退到 ffmpeg。请手动安装ffmpeg）
1. streamlink（仅限 hls 流，不支持的流将回退到 ffmpeg。需安装 FFmpeg）
<br />
2. ffmpeg（ffmpeg 下载。请手动安装ffmpeg）
2. ffmpeg（需安装 FFmpeg）
<br />
3. stream-gears（默认）
3. stream-gears（默认。防 FLV 流花屏）
<br />
4. sync-downloader（流式边录边传，需先为主播设定上传模板。不受
pool2/threads/segment_time 控制，默认 3 线程上传，请确保上传带宽充足。需安装 FFmpeg）
</div>
}
style={{ width: '100%' }}
@@ -51,7 +55,51 @@ const Global: React.FC = () => {
<Select.Option value="streamlink">streamlink（hls多线程下载）</Select.Option>
<Select.Option value="ffmpeg">ffmpeg</Select.Option>
<Select.Option value="stream-gears">stream-gears（默认）</Select.Option>
<Select.Option value="sync-downloader">sync-downloader（边录边传）</Select.Option>
</Form.Select>
{formApi.getValue('downloader') === 'sync-downloader' ? (
<>
<Form.Input
field="sync_save_dir"
label="边录边传额外保存本地目录（sync_save_dir）"
placeholder=""
style={{ width: '100%' }}
fieldStyle={{
alignSelf: 'stretch',
padding: 0,
}}
showClear={true}
disabled={formApi.getValue('downloader') === 'sync-downloader' ? false : true}
rules={[
{
pattern: /^[^*|?"<>]*$/,
message: '路径中不能包含Windows不允许的字符 * | ? " < >',
},
{
pattern: /^(?![a-zA-Z]：).*$/,
message: '以字母开头时，第二个字符不能是中文冒号',
},
{
pattern: /^[^:]*$|^[a-zA-Z]:[\/\\][^:]*$/,
message: '冒号只能出现在第二个字符位置，且后面必须连接斜杠',
},
{
pattern: /^(?!.*?\.{3,})(?!.*?\.{2}(?![\/\\])).*$/,
message: '点号最多只能连续出现两次，且后面必须连接斜杠',
},
{
pattern: /^(?!.*\/\\)(?!.*\\\/).*$/,
message: '不允许连接正反斜杠',
},
{
pattern: /^(?!.*([\\]{3,}|[\/]{2,})).*$/,
message: '反斜杠最多只能连续出现两次，正斜杠最多只能连续出现一次',
},
]}
stopValidateWithError={true}
/>
</>
) : null}
<Form.InputNumber
label="视频分段大小（file_size）"
extraText={
app/upload-manager/add/page.tsx
+1
View File
@@ -77,6 +77,7 @@ export default function Add() {
dtime: values.isDtime ? values?.dtime : null,
credits: values.credits,
uploader: values.uploader,
extra_fields: values.extra_fields ?? '',
}
const result = await trigger(studioEntity)
app/upload-manager/edit/page.tsx
+1
View File
@@ -92,6 +92,7 @@ const Edit = () => {
dtime: values?.isDtime ? values?.dtime : null,
credits: values?.credits ?? null,
uploader: values?.uploader ?? null,
extra_fields: values?.extra_fields ?? '',
}
const result = await trigger(studioEntity)
await mutate(result)
biliup/Danmaku/bilibili.py
+7 -25
View File
@@ -27,31 +27,13 @@ class Bilibili:
@staticmethod
async def get_ws_info(url, content):
# 判断是否要录制详细弹幕
if content.get('detail', False):
# 获取传入的用户信息
cookie_str = content.get('cookie', "") if content.get('cookie', False) else ""
buid = content.get('bili_uid', 0)
Bilibili.headers['cookie'] = cookie_str
# 如buid不为0但没有cookie传入
if not cookie_str and buid != 0:
build = 0
# 获取B站用户ID
if cookie_str and buid == 0:
async with aiohttp.ClientSession(headers=Bilibili.headers) as session:
try:
async with session.get(f"https://api.bilibili.com/x/web-interface/nav", timeout=5) as resp:
resp_data = await resp.json()
buid = resp_data["data"]["mid"]
logger.info(f"获取B站用户ID {buid}")
except Exception as e:
buid = 0
Bilibili.headers['cookie'] = ""
pass
uid = content['uid']
# 传入内容中，如果 uid 不为 0，则 cookie 必然存在，且必然为详细模式
if uid > 0:
Bilibili.headers['cookie'] = content['cookie']
else:
buid = 0
Bilibili.headers['cookie'] = ""
# 获取弹幕认证信息
danmu_wss_url = 'wss://broadcastlv.chat.bilibili.com/sub'
@@ -74,7 +56,7 @@ class Bilibili:
pass
w_data = {
'uid': buid,
'uid': uid,
'roomid': room_id,
'protover': 3,
'platform': 'web',
@@ -83,7 +65,7 @@ class Bilibili:
}
data = json.dumps(w_data).encode('utf-8')
logger.info(f"danmaku auth info {data}")
# logger.info(f"danmaku auth info {data}")
reg_datas = [(pack('>i', len(data) + 16) + b'\x00\x10\x00\x01' + pack('>i', 7) + pack('>i', 1) + data)]
return danmu_wss_url, reg_datas
biliup/database/db.py
+3 -1
View File
@@ -56,7 +56,8 @@ def get_stream_info(db: Session, name: str) -> dict:
order_by(desc(StreamerInfo.id))
).first()
if res:
res = res._asdict()
res = res._asdict()['StreamerInfo'].as_dict()
# print(res['StreamerInfo'].as_dict())
res["date"] = datetime_to_struct_time(res["date"])
return res
return {}
@@ -188,6 +189,7 @@ def migrate_via_alembic():
command.upgrade(alembic_cfg, 'head')
print("检测到旧版数据库，已完成自动迁移")
def backup(db: Session):
"""备份数据库"""
pass
biliup/database/models.py
+1
View File
@@ -130,6 +130,7 @@ class UploadStreamers(BaseModel):
up_selection_reply: Mapped[int] = mapped_column(nullable=True) # 精选评论
up_close_reply: Mapped[int] = mapped_column(nullable=True) # 关闭评论
up_close_danmu: Mapped[int] = mapped_column(nullable=True) # 精选评论
extra_fields: Mapped[str] = mapped_column(nullable=True) # 额外字段
livestreamers: Mapped[List["LiveStreamers"]] = relationship(back_populates="uploadstreamers")
biliup/engine/download.py
+80 -15
View File
@@ -1,6 +1,8 @@
import asyncio
import inspect
import logging
import os
import queue
import re
import subprocess
import threading
@@ -16,13 +18,17 @@ from requests.utils import DEFAULT_ACCEPT_ENCODING
from httpx import HTTPStatusError
from biliup.common.util import client, loop, check_timerange
from biliup.database.db import add_stream_info, SessionLocal, update_cover_path, update_room_title, update_file_list
from biliup.database.db import add_stream_info, SessionLocal, get_stream_info, update_cover_path, update_room_title, update_file_list
from biliup.plugins import random_user_agent
import stream_gears
from PIL import Image
from biliup.config import config
from biliup.Danmaku import IDanmakuClient
from biliup.plugins.bili_webup_sync import BiliWebAsync
from biliup.uploader import fmt_title_and_desc
from .sync_downloader import SyncDownloader
logger = logging.getLogger('biliup')
@@ -87,6 +93,8 @@ class DownloadBase(ABC):
def download(self):
logger.info(f"{self.plugin_msg}: Start downloading {self.raw_stream_url}")
# 调试使用边录边上传功能
# self.downloader = 'sync-downloader'
if self.is_download:
if not shutil.which("ffmpeg"):
logger.error("未安装 FFMpeg 或不存在于 PATH 内")
@@ -96,17 +104,32 @@ class DownloadBase(ABC):
return self.ffmpeg_segment_download()
parsed_url_path = urlparse(self.raw_stream_url).path
if self.downloader == 'streamlink' or self.downloader == 'ffmpeg':
if shutil.which("ffmpeg"):
# streamlink无法处理flv,所以回退到ffmpeg
if self.downloader == 'streamlink' and '.flv' not in parsed_url_path:
return self.ffmpeg_download(use_streamlink=True)
else:
return self.ffmpeg_download()
else:
if self.downloader != 'stream-gears':
if not shutil.which("ffmpeg"):
logger.error("未安装 FFMpeg 或不存在于 PATH 内，本次下载使用 stream-gears")
logger.debug("Current user's PATH is:" + os.getenv("PATH"))
# 同步下载上传器
if self.downloader == 'sync-downloader':
logger.info(f"{self.plugin_msg}: 使用同步下载器")
stream_info = config.get('streamers', {}).get(self.fname, {})
stream_info.update({'name': self.fname})
if not self.file_size:
self.file_size = 2 * 1024 * 1024 * 1024
else:
min_size = 10 * 1024 * 1024
self.file_size = ((self.file_size + min_size - 1) // min_size) * min_size # 向上取整
sync_download(self.raw_stream_url, self.fake_headers,
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))
return True
# streamlink无法处理flv,所以回退到ffmpeg
if self.downloader == 'streamlink' and '.flv' not in parsed_url_path:
return self.ffmpeg_download(use_streamlink=True)
return self.ffmpeg_download()
if '.flv' in parsed_url_path:
# 假定flv流
self.suffix = 'flv'
@@ -178,7 +201,10 @@ class DownloadBase(ABC):
# ffmpeg 输入参数
input_args = []
# ffmpeg 输出参数
output_args = []
output_args = [
'-c',
'copy',
]
if use_streamlink:
streamlink_cmd = [
'streamlink',
@@ -218,7 +244,7 @@ class DownloadBase(ABC):
else:
output_args += ['-f', self.suffix]
args = ['ffmpeg', '-y', *input_args, *output_args, '-c', 'copy',
args = ['ffmpeg', '-y', *input_args, *output_args,
f'{fmt_file_name}.{self.suffix}.part']
with subprocess.Popen(args, stdin=subprocess.DEVNULL if not streamlink_proc else streamlink_proc.stdout,
stdout=subprocess.PIPE, stderr=subprocess.STDOUT) as proc:
@@ -329,7 +355,7 @@ class DownloadBase(ABC):
self.download_cover(
time.strftime(self.gen_download_filename().encode("unicode-escape").decode(), end_time if end_time else time.localtime()
).encode().decode("unicode-escape"))
).encode().decode("unicode-escape"))
# 更新数据库中封面存储路径
with SessionLocal() as db:
update_cover_path(db, self.database_row_id, self.live_cover_path)
@@ -416,7 +442,7 @@ class DownloadBase(ABC):
url = m3u8_obj.playlists[0].uri
logger.info(f'{self.plugin_msg}: stream url: {url}')
r = await __client_get(url)
else: # 处理 Flv
else: # 处理 Flv
r = await __client_get(url, stream=True)
if r.headers.get('Location'):
url = r.headers['Location']
@@ -425,7 +451,7 @@ class DownloadBase(ABC):
if r.status_code == 200:
return url
except HTTPStatusError as e:
logger.error(f'{self.plugin_msg}: url {url}: status_code-{e.response.status_code}')
logger.debug(f'{self.plugin_msg}: url {url}: status_code-{e.response.status_code}')
except:
logger.debug(f'{self.plugin_msg}: url {url}: ', exc_info=True)
return None
@@ -496,6 +522,44 @@ 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):
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)
# 获取 BiliWebAsync.__init__ 的参数名
init_params = inspect.signature(BiliWebAsync.__init__).parameters
# 过滤 info 中的无关键
filtered_info = {key: value for key, value in stream_info.items() if key in init_params}
filtered_info['submit_api'] = config.get('submit_api')
filtered_info['lines'] = config.get('lines', 'AUTO')
# 映射 'uploader' 到 'principal'
filtered_info['principal'] = ""
filtered_info["data"] = stream_info
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)
# print("上传器结束")
logger.info(f"{stream_info['name']} 上传器结束")
# video_queue = queue.SimpleQueue()
downloader = SyncDownloader(stream_url, headers, segment_duration, max_file_size, output_prefix, video_queue)
# 启动上传器
upload_thread = threading.Thread(target=upload, args=(video_queue, stream_info, downloader.stop_event), daemon=True)
upload_thread.start()
downloader.run()
logger.info(f"{stream_info['name']} 下载器结束")
def get_valid_filename(name):
"""
Return the given string converted to a string that can be used for a clean
@@ -532,7 +596,8 @@ def get_duration(segment_time_str, time_range):
time_diff = end_time_today - now
if segment_time_str:
segment_time_parts = list(map(int, segment_time_str.split(":")))
segment_time = timedelta(hours=segment_time_parts[0], minutes=segment_time_parts[1], seconds=segment_time_parts[2])
segment_time = timedelta(hours=segment_time_parts[0],
minutes=segment_time_parts[1], seconds=segment_time_parts[2])
if time_diff > segment_time:
return segment_time_str
biliup/engine/sync_downloader.py
+249
View File
@@ -0,0 +1,249 @@
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
import logging
import os
import queue
import threading
import subprocess
import time
from urllib.parse import urlparse
logger = logging.getLogger('biliup.engine.sync_downloader')
def pad_file_to_size(filename, min_file_size):
"""
若文件大小不足 min_file_size，则在文件末尾填充 0x00 至目标大小。
"""
if not os.path.exists(filename):
return
current_size = os.path.getsize(filename)
if current_size < min_file_size:
need_pad = min_file_size - current_size
print(f"[pad_file_to_size] 补齐文件 {filename}："
f"填充 {need_pad} 字节 0x00 使其达到 {min_file_size} 字节")
with open(filename, "ab") as f:
f.write(b"\x00" * need_pad)
class SyncDownloader:
"""
同步下载-切片类
说明：
1. 在 run() 方法中，单线程循环执行录制逻辑；
2. 每次启动一段 ffmpeg 录制，并读取 streamlink stdout 作为输入；
3. 当 ffmpeg 录制结束后，杀掉当前的 streamlink 进程并补齐文件大小；
4. 若中途发现 streamlink 无数据可读（EOF），则说明没有更多内容可下载，结束整个程序。
"""
def __init__(self,
stream_url="http://localhost:8888/stream0/index.m3u8",
headers={"a": "1"},
segment_duration=10,
max_file_size=100,
output_prefix="segment_",
video_queue=None):
"""
:param stream_url: 拉流地址
:param segment_duration: 每段录制时长（秒）（暂时不用）
:param max_file_size: 文件最小大小，不足时进行 0x00 填充 单位 MB
:param read_block_size: 从 streamlink stdout 读取数据时的单次块大小
:param output_prefix: 输出文件名前缀
"""
self.stream_url = stream_url
self.quality = "best"
self.headers = headers
self.segment_duration = segment_duration
self.read_block_size = 4096
self.max_file_size = max_file_size
self.output_prefix = output_prefix
self.video_queue: queue.SimpleQueue = video_queue
self.stop_event = threading.Event()
def run_ffmpeg_with_url(self, ffmpeg_cmd, output_filename):
with subprocess.Popen(ffmpeg_cmd, stderr=subprocess.PIPE, stdout=subprocess.PIPE) as ffmpeg_proc:
# print("[run] 启动 ffmpeg...")
logging.info("[run] 启动 ffmpeg...")
if output_filename == "-":
data = ffmpeg_proc.stdout.read(1024) # 读取第一个 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)
if not data:
logging.info("[run] ffmpeg stdout 已到达 EOF。结束本段写入。")
break
self.video_queue.put(data)
ffmpeg_proc.wait()
return True # 如果正常执行，返回 True
def run_streamlink_with_ffmpeg(self, streamlink_cmd, ffmpeg_cmd, output_filename):
with subprocess.Popen(streamlink_cmd, stdout=subprocess.PIPE, stderr=subprocess.PIPE) as streamlink_proc:
logging.info("[run] 启动 streamlink...")
with subprocess.Popen(ffmpeg_cmd, stdin=streamlink_proc.stdout, stdout=subprocess.PIPE) as ffmpeg_proc:
logging.info("[run] 启动 ffmpeg...")
if output_filename == "-":
logging.info("[run] 读取 ffmpeg stdout...")
# 读取第一个数据
data = ffmpeg_proc.stdout.read(1024)
if not data: # 如果第一个数据为空
logging.info("[run] ffmpeg 没有输出数据，返回 False")
streamlink_proc.kill() # 终止 streamlink 进程
ffmpeg_proc.kill() # 终止 ffmpeg 进程
return False
self.video_queue.put(data) # 将第一个数据放入队列
# 继续读取剩余数据
while True:
data = ffmpeg_proc.stdout.read(1024)
if not data:
logging.info("[run] ffmpeg stdout 已到达 EOF。结束本段写入。")
break
self.video_queue.put(data)
ffmpeg_proc.wait()
logging.info("[run] ffmpeg 已到达输出大小并退出。结束本段写入。")
# streamlink_proc.kill() # 确保终止 streamlink 进程
# streamlink_proc.wait()
return True # 如果一切正常，返回 True
def build_ffmpeg_cmd(self, input_source, output_filename, headers, segment_duration):
cmd = [
"ffmpeg",
"-loglevel", "quiet",
# "-" # 覆盖输出文件
]
if headers:
cmd += ["-headers", ''.join(f'{key}: {value}\r\n' for key, value in headers.items())]
for i in ["-i", input_source, # 输入源
# "-t", str(segment_duration),
"-fs", f"{self.max_file_size}M",
"-c:v", "copy",
"-c:a", "copy",
"-movflags", "+frag_keyframe+empty_moov",
"-f", "matroska"]:
cmd.append(i)
cmd.append("-")
return cmd
def run(self):
"""
主逻辑：循环进行分段录制。
- 每段录制：
1) 启动 streamlink；若 EOF 则退出。
2) 启动 ffmpeg (带 -fs 参数限制输出大小)；
3) 从 streamlink stdout 读数据，写给 ffmpeg stdin；
4) ffmpeg 到时长后退出，然后杀掉 streamlink；
5) 进入下一段，如此往复。
- 若中途发现 streamlink 无数据（EOF）则跳出循环。
"""
file_index = 1
retry_count = 0
while True:
if self.stop_event.is_set():
break
if retry_count >= 5:
logger.info("这个直播流已经失效，停止下载器")
return
output_filename = f"{self.output_prefix}{file_index:03d}.mkv"
# print(f"\n[run] ========== 准备录制第 {file_index} 段：{output_filename} ==========")
# logging.info(f"\n[run] == 当前下载流地址：{self.stream_url} ==")
logging.info(f"\n[run] == 准备录制第 {file_index} 段：{output_filename} ==")
output_filename = "-"
is_hls = '.m3u8' in urlparse(self.stream_url).path
if not is_hls:
# print("[run] 输入源不是 HLS 地址，将直接使用 ffmpeg 进行录制。", self.stream_url)
logging.info("[run] 输入源不是 HLS 地址，将直接使用 ffmpeg 进行录制。")
ffmpeg_cmd = self.build_ffmpeg_cmd(self.stream_url, output_filename,
self.headers, self.segment_duration)
if not self.run_ffmpeg_with_url(ffmpeg_cmd, output_filename):
retry_count += 1
time.sleep(1)
continue
else:
# print("[run] 输入源是 HLS 地址，将使用 streamlink + ffmpeg 进行录制。")
logging.info("[run] 输入源是 HLS 地址，将使用 streamlink + ffmpeg 进行录制。")
streamlink_cmd = [
"streamlink",
"--stream-segment-threads", "3",
"--hls-playlist-reload-attempts", "1",
"--http-header", ';'.join([f'{key}={value}' for key, value in self.headers.items()]),
self.stream_url,
self.quality,
"-O" # 输出到 stdout
]
# output_filename = "-"
ffmpeg_cmd = self.build_ffmpeg_cmd("pipe:0", output_filename, None, self.segment_duration)
if not self.run_streamlink_with_ffmpeg(streamlink_cmd, ffmpeg_cmd, output_filename):
retry_count += 1
time.sleep(1)
continue
# 6. 进入下一段
# if file_index != 1:
self.video_queue.put(None) # 通知消费者线程本段录制结束
file_index += 1
def main():
slicer = SyncDownloader(
stream_url="http://127.0.0.1:8888/live/index.m3u8",
segment_duration=10,
read_block_size=4096,
output_prefix="segment_"
)
# ====【可选】消费者线程示例，演示如何拿到 video_queue 的数据====
def consumer():
file_index = 1
while True:
data_count = 0
with open(f"output_{file_index}.mkv", "wb") as f:
while True:
data = slicer.video_queue.get() # 阻塞式获取
if data is None:
break
f.write(data)
data_count += len(data)
# print(f"[consumer] 写入文件 output_{file_index}.mkv，大小：{data_count} 字")
print(f"[consumer] 写入文件 output_{file_index}.mkv，大小：{data_count} 字")
if data_count < 100:
print(f"[consumer] 无效文件，删除 output_{file_index}.mkv")
os.remove(f"output_{file_index}.mkv")
slicer.stop_event.set()
break
pad_file_to_size(f"output_{file_index}.mkv", 100 * 1024 * 1024) # 补齐文件大小
file_index += 1
# if slicer.stop_event.is_set():
# break
# 在这里可以对 data 做进一步处理，比如再推到别的地方
# 启动消费者线程
t = threading.Thread(target=consumer, daemon=True)
t.start()
# 启动下载录制主逻辑
slicer.run()
# 停止消费者（如录制完毕后可执行）
# slicer.video_queue.put(None)
# t.join()
if __name__ == "__main__":
main()
biliup/plugins/bili_webup_sync.py
+810
View File
@@ -0,0 +1,810 @@
import asyncio
import base64
import concurrent.futures
import hashlib
import json
import logging
import math
import os
import re
import sys
import threading
import time
import queue
import urllib.parse
from dataclasses import asdict, dataclass, field, InitVar
from json import JSONDecodeError
from os.path import splitext, basename
from typing import Callable, Dict, Union, Any, List
from urllib import parse
from urllib.parse import quote
import aiohttp
from concurrent.futures.thread import ThreadPoolExecutor
import concurrent
import requests.utils
import rsa
import xml.etree.ElementTree as ET
from requests.adapters import HTTPAdapter, Retry
from biliup.config import config
from ..engine import Plugin
from ..engine.upload import UploadBase, logger
logger = logging.getLogger('biliup.engine.bili_web_sync')
@Plugin.upload(platform="bili_web_sync")
class BiliWebAsync(UploadBase):
def __init__(
self, principal, data, submit_api=None, copyright=2, postprocessor=None, dtime=None,
dynamic='', lines='AUTO', threads=3, tid=122, tags=None, cover_path=None, description='',
dolby=0, hires=0, no_reprint=0, open_elec=0, credits=None,
user_cookie='cookies.json', copyright_source=None, extra_fields="", video_queue=None
):
super().__init__(principal, data, persistence_path='bili.cookie', postprocessor=postprocessor)
if credits is None:
credits = []
if tags is None:
tags = []
self.lines = lines
self.submit_api = submit_api
self.threads = threads
self.tid = tid
self.tags = tags
self.dtime = dtime
if cover_path:
self.cover_path = cover_path
elif "live_cover_path" in self.data:
self.cover_path = self.data["live_cover_path"]
else:
self.cover_path = None
self.desc = description
self.credits = credits
self.dynamic = dynamic
self.copyright = copyright
self.dolby = dolby
self.hires = hires
self.no_reprint = no_reprint
self.open_elec = open_elec
self.copyright_source = copyright_source
self.extra_fields = extra_fields
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]:
# print("开始同步上传")
logger.info("开始同步上传")
file_index = 1
videos = Data()
bili = BiliBili(videos)
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()
else:
videos.desc_v2 = [{
"raw_text": self.desc,
"biz_id": "",
"type": 1
}]
videos.desc = self.desc
videos.copyright = self.copyright
if self.copyright == 2:
videos.source = self.data["url"] # 添加转载地址说明
# 设置视频分区,默认为174 生活，其他分区
videos.tid = self.tid
videos.set_tag(self.tags)
if self.dtime:
videos.delay_time(int(time.time()) + self.dtime)
if self.cover_path:
videos.cover = bili.cover_up(self.cover_path).replace('http:', '')
# 其他参数设置
videos.extra_fields = self.extra_fields
videos.dolby = self.dolby
videos.hires = self.hires
videos.no_reprint = self.no_reprint
videos.open_elec = self.open_elec
thread_list = []
while True:
# 调试使用 分p 强制停止
# if file_index > 10:
# logger.info(f"[consumer debug] 停止下载回调")
# stop_event.set()
# break
file_name = f"{output_prefix}_{file_index}.mkv"
# if file_name_callback:
# file_name_callback(file_name)
data_size = 0
video_upload_queue = queue.SimpleQueue()
t = threading.Thread(target=bili.upload_stream, args=(video_upload_queue,
file_name, total_size, self.lines, videos, stop_event, file_name_callback), daemon=True, name=f"upload_{file_index}")
thread_list.append(t)
t.start()
while True:
try:
data = self.video_queue.get(timeout=10)
except queue.Empty:
break
if data is None:
video_upload_queue.put(None)
break
video_upload_queue.put(data)
# print(video_upload_queue.empty())
data_size += len(data)
# print(f"[consumer] 读取 {file_name} {data_size} 字节")
logger.info(f"[consumer] 读取 {file_name} {data_size} 字节")
file_index += 1
# print("[consumer] bili.video.videos", bili.video.videos)
logger.info(f"[consumer] bili.video.videos {bili.video.videos}")
if data_size < 100:
# print(f"[consumer] 停止下载回调")
# n = video_upload_queue.get()
logger.info(f"[consumer] 停止下载回调")
stop_event.set()
break
logger.info("等待上传线程结束")
for t in thread_list:
t.join()
# ret = bili.submit(self.submit_api) # 提交视频
# logger.info(f"上传成功: {ret}")
file_list = []
# if config.get('sync_save_dir', None):
# file_list = [os for file_name in os.listdir("sync_downloaded")]
# print("上传完成", file_list)
return file_list
def creditsToDesc_v2(self):
desc_v2 = []
desc_v2_tmp = self.desc
for credit in self.credits:
try:
num = desc_v2_tmp.index("@credit")
desc_v2.append({
"raw_text": " " + desc_v2_tmp[:num],
"biz_id": "",
"type": 1
})
desc_v2.append({
"raw_text": credit["username"],
"biz_id": str(credit["uid"]),
"type": 2
})
self.desc = self.desc.replace(
"@credit", "@" + credit["username"] + " ", 1)
desc_v2_tmp = desc_v2_tmp[num + 7:]
except IndexError:
logger.error('简介中的@credit占位符少于credits的数量,替换失败')
desc_v2.append({
"raw_text": " " + desc_v2_tmp,
"biz_id": "",
"type": 1
})
desc_v2[0]["raw_text"] = desc_v2[0]["raw_text"][1:] # 开头空格会导致识别简介过长
return desc_v2
class BiliBili:
def __init__(self, video: 'Data'):
self.app_key = None
self.appsec = None
# if self.app_key is None or self.appsec is None:
self.app_key = 'ae57252b0c09105d'
self.appsec = 'c75875c596a69eb55bd119e74b07cfe3'
self.__session = requests.Session()
self.video = video
self.__session.mount('https://', HTTPAdapter(max_retries=Retry(total=5)))
self.__session.headers.update({
'user-agent': "Mozilla/5.0 (X11; Linux x86_64) AppleWebKit/537.36 Chrome/63.0.3239.108",
'referer': "https://www.bilibili.com/",
'connection': 'keep-alive'
})
self.cookies = None
self.access_token = None
self.refresh_token = None
self.account = None
self.__bili_jct = None
self._auto_os = None
self.persistence_path = 'engine/bili.cookie'
self.save_dir = config.get('sync_save_dir', None)
self.save_path = ''
if self.save_dir and not os.path.exists(self.save_dir):
os.makedirs(self.save_dir)
def myinfo(self, cookies: dict = None):
if cookies:
requests.utils.add_dict_to_cookiejar(self.__session.cookies, cookies)
response = self.__session.get('https://api.bilibili.com/x/space/myinfo', timeout=15)
return response.json()
def login(self, persistence_path, user_cookie):
self.persistence_path = user_cookie
if os.path.isfile(user_cookie):
print('使用持久化内容上传')
self.load()
if self.cookies:
try:
self.login_by_cookies(self.cookies)
except:
logger.exception('login error')
self.login_by_password(**self.account)
else:
self.login_by_password(**self.account)
self.store()
def load(self):
try:
with open(self.persistence_path) as f:
self.cookies = json.load(f)
self.access_token = self.cookies['token_info']['access_token']
self.refresh_token = self.cookies['token_info']['refresh_token']
except (JSONDecodeError, KeyError):
logger.exception('加载cookie出错')
def store(self):
with open(self.persistence_path, "w") as f:
json.dump({**self.cookies,
'access_token': self.access_token,
'refresh_token': self.refresh_token
}, f)
def login_by_password(self, username, password):
print('使用账号上传')
key_hash, pub_key = self.get_key()
encrypt_password = base64.b64encode(rsa.encrypt(f'{key_hash}{password}'.encode(), pub_key))
payload = {
"actionKey": 'appkey',
"appkey": self.app_key,
"build": 6270200,
"captcha": '',
"challenge": '',
"channel": 'bili',
"device": 'phone',
"mobi_app": 'android',
"password": encrypt_password,
"permission": 'ALL',
"platform": 'android',
"seccode": "",
"subid": 1,
"ts": int(time.time()),
"username": username,
"validate": "",
}
response = self.__session.post("https://passport.bilibili.com/x/passport-login/oauth2/login", timeout=5,
data={**payload, 'sign': self.sign(parse.urlencode(payload))})
r = response.json()
if r['code'] != 0 or r.get('data') is None or r['data'].get('cookie_info') is None:
raise RuntimeError(r)
try:
for cookie in r['data']['cookie_info']['cookies']:
self.__session.cookies.set(cookie['name'], cookie['value'])
if 'bili_jct' == cookie['name']:
self.__bili_jct = cookie['value']
self.cookies = self.__session.cookies.get_dict()
self.access_token = r['data']['token_info']['access_token']
self.refresh_token = r['data']['token_info']['refresh_token']
except:
raise RuntimeError(r)
return r
def login_by_cookies(self, cookie):
cookies_dict = {c['name']: c['value'] for c in cookie['cookie_info']['cookies']}
requests.utils.add_dict_to_cookiejar(self.__session.cookies, cookies_dict)
if 'bili_jct' in cookies_dict:
self.__bili_jct = cookies_dict['bili_jct']
data = self.__session.get("https://api.bilibili.com/x/web-interface/nav", timeout=5).json()
if data["code"] != 0:
raise Exception(data)
print('使用cookies上传')
def sign(self, param):
return hashlib.md5(f"{param}{self.appsec}".encode()).hexdigest()
def get_key(self):
url = "https://passport.bilibili.com/x/passport-login/web/key"
payload = {
'appkey': f'{self.app_key}',
'sign': self.sign(f"appkey={self.app_key}"),
}
response = self.__session.get(url, data=payload, timeout=5)
r = response.json()
if r and r["code"] == 0:
return r['data']['hash'], rsa.PublicKey.load_pkcs1_openssl_pem(r['data']['key'].encode())
def probe(self):
ret = self.__session.get('https://member.bilibili.com/preupload?r=probe', timeout=5).json()
logger.info(f"线路:{ret['lines']}")
data, auto_os = None, None
min_cost = 0
if ret['probe'].get('get'):
method = 'get'
else:
method = 'post'
data = bytes(int(1024 * 0.1 * 1024))
for line in ret['lines']:
start = time.perf_counter()
test = self.__session.request(method, f"https:{line['probe_url']}", data=data, timeout=30)
cost = time.perf_counter() - start
print(line['query'], cost)
if test.status_code != 200:
return
if not min_cost or min_cost > cost:
auto_os = line
min_cost = cost
auto_os['cost'] = min_cost
return auto_os
def upload_stream(
self,
stream_queue: queue.SimpleQueue,
file_name,
total_size,
lines='AUTO',
videos: 'Data' = None,
stop_event: threading.Event = None,
file_name_callback: Callable[[str], None] = None,
submit_api: Callable[[str], None] = None
):
logger.info(f"{file_name} 开始上传")
if self.save_dir:
self.save_path = os.path.join(self.save_dir, file_name)
cs_upcdn = ['alia', 'bda', 'bda2', 'bldsa', 'qn', 'tx', 'txa']
jd_upcdn = ['jd-alia', 'jd-bd', 'jd-bldsa', 'jd-tx', 'jd-txa']
preferred_upos_cdn = None
if not self._auto_os:
if lines in cs_upcdn:
self._auto_os = {"os": "upos", "query": f"upcdn={lines}&probe_version=20221109",
"probe_url": f"//upos-cs-upcdn{lines}.bilivideo.com/OK"}
preferred_upos_cdn = lines
elif lines in jd_upcdn:
lines = lines.split('-')[1]
self._auto_os = {"os": "upos", "query": f"upcdn={lines}&probe_version=20221109",
"probe_url": f"//upos-jd-upcdn{lines}.bilivideo.com/OK"}
preferred_upos_cdn = lines
else:
self._auto_os = self.probe()
logger.info(f"线路选择 => {self._auto_os['os']}: {self._auto_os['query']}. time: {self._auto_os.get('cost')}")
if self._auto_os['os'] == 'upos':
upload = self.upos_stream
else:
logger.error(f"NoSearch:{self._auto_os['os']}")
raise NotImplementedError(self._auto_os['os'])
logger.info(f"os: {self._auto_os['os']}")
query = {
'r': self._auto_os['os'],
'profile': 'ugcupos/bup',
'ssl': 0,
'version': '2.8.12',
'build': 2081200,
'name': file_name,
'size': total_size,
}
resp = self.__session.get(
f"https://member.bilibili.com/preupload?{self._auto_os['query']}", params=query,
timeout=5)
ret = resp.json()
logger.debug(f"preupload: {ret}")
if preferred_upos_cdn:
# 如果返回的endpoint不在probe_url中，则尝试在endpoints中校验probe_url是否可用
if ret['endpoint'] not in self._auto_os['probe_url']:
for endpoint in ret['endpoints']:
if endpoint in self._auto_os['probe_url']:
ret['endpoint'] = endpoint
logger.info(f"修改endpoint: {ret['endpoint']}")
break
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:
stop_event.set()
# print("异常流 直接退出")
return
video_part['title'] = video_part['title'][:80]
videos.append(video_part) # 添加已经上传的视频
edit = False if videos.aid is None else True
ret = self.submit(submit_api=submit_api, edit=edit)
# logger.info(f"上传成功: {ret}")
if edit:
logger.info(f"编辑添加成功: {ret}")
else:
logger.info(f"上传成功: {ret}")
aid = ret['data']['aid']
videos.aid = aid
if file_name_callback:
file_name_callback(self.save_path)
async def upos_stream(self, stream_queue, file_name, total_size, ret):
# print("--------------, ", file_name)
chunk_size = ret['chunk_size']
auth = ret["auth"]
endpoint = ret["endpoint"]
biz_id = ret["biz_id"]
upos_uri = ret["upos_uri"]
url = f"https:{endpoint}/{upos_uri.replace('upos://', '')}" # 视频上传路径
headers = {
"X-Upos-Auth": auth
}
# 向上传地址申请上传，得到上传id等信息
upload_id = self.__session.post(f'{url}?uploads&output=json', timeout=15,
headers=headers).json()["upload_id"]
# 开始上传
parts = [] # 分块信息
chunks = math.ceil(total_size / chunk_size) # 获取分块数量
start = time.perf_counter()
# print("-----------")
# print(upload_id, chunks, chunk_size, total_size)
logger.info(
f"{file_name} - upload_id: {upload_id}, chunks: {chunks}, chunk_size: {chunk_size}, total_size: {total_size}")
n = 0
st = time.perf_counter()
max_workers = 3
semaphore = threading.Semaphore(max_workers)
with ThreadPoolExecutor(max_workers=max_workers) as executor:
futures = []
for index, chunk in enumerate(self.queue_reader_generator(stream_queue, chunk_size, total_size)):
if not chunk:
break
const_time = time.perf_counter() - st
speed = len(chunk) * 8 / 1024 / 1024 / const_time
logger.info(f"{file_name} - chunks-({index+1}/{chunks}) - down - speed: {speed:.2f}Mbps")
n += len(chunk)
params = {
'uploadId': upload_id,
'chunks': chunks,
'total': total_size,
'chunk': index,
'size': chunk_size,
'partNumber': index + 1,
'start': index * chunk_size,
'end': index * chunk_size + chunk_size
}
params_clone = params.copy()
semaphore.acquire()
future = executor.submit(self.upload_chunk_thread,
url, chunk, params_clone, headers, file_name)
future.add_done_callback(lambda x: semaphore.release())
futures.append(future)
st = time.perf_counter()
for f in list(futures):
if f.done():
futures.remove(f)
# 等待所有分片上传完成，并按顺序收集结果
for future in concurrent.futures.as_completed(futures):
pass
results = [{
"partNumber": i + 1,
"eTag": "etag"
} for i in range(chunks)]
parts.extend(results)
if len(parts) == 0:
return None
logger.info(f"{file_name} - total_size: {total_size}, n: {n}")
cost = time.perf_counter() - start
p = {
'name': file_name,
'uploadId': upload_id,
'biz_id': biz_id,
'output': 'json',
'profile': 'ugcupos/bup'
}
attempt = 0
while attempt <= 5: # 一旦放弃就会丢失前面所有的进度，多试几次吧
try:
r = self.__session.post(url, params=p, json={"parts": parts}, headers=headers, timeout=15).json()
if r.get('OK') == 1:
logger.info(f'{file_name} uploaded >> {total_size / 1000 / 1000 / cost:.2f}MB/s. {r}')
return {"title": splitext(file_name)[0], "filename": splitext(basename(upos_uri))[0], "desc": ""}
raise IOError(r)
except IOError:
attempt += 1
logger.info(f"请求合并分片时出现问题，尝试重连，次数：" + str(attempt))
time.sleep(15)
pass
def upload_chunk_thread(self, url, chunk, params_clone, headers, file_name, max_retries=3, backoff_factor=1):
st = time.perf_counter()
retries = 0
while retries < max_retries:
try:
r = requests.put(url=url, params=params_clone, data=chunk, headers=headers)
# 如果上传成功，退出重试循环
if r.status_code == 200:
const_time = time.perf_counter() - st
speed = len(chunk) * 8 / 1024 / 1024 / const_time
logger.info(
f"{file_name} - chunks-{params_clone['chunk'] +1 } - up status: {r.status_code} - speed: {speed:.2f}Mbps"
)
return {
"partNumber": params_clone['chunk'] + 1,
"eTag": "etag"
}
# 如果上传失败，但未达到最大重试次数，等待一段时间后重试
else:
retries += 1
logger.warning(
f"{file_name} - chunks-{params_clone['chunk']} - up failed: {r.status_code}. Retrying {retries}/{max_retries}")
# 计算退避时间，逐步增加重试间隔
backoff_time = backoff_factor ** retries
time.sleep(backoff_time)
except Exception as e:
retries += 1
logger.error(f"upload_chunk_thread err {str(e)}. Retrying {retries}/{max_retries}")
# 计算退避时间，逐步增加重试间隔
backoff_time = backoff_factor ** retries
time.sleep(backoff_time)
# 如果重试了所有次数仍然失败，记录错误
logger.error(f"{file_name} - chunks-{params_clone['chunk']} - Upload failed after {max_retries} attempts.")
return None
def queue_reader_generator(self, simple_queue: queue.SimpleQueue, chunk_size: int, max_size: int):
"""
从 simple_queue 中读取数据并按 chunk_size 大小分块产出 (yield)
当队列中获取到 None 或者数据总量达到 max_size 后，就用 0x00 补齐到 chunk_size
:param simple_queue: queue.SimpleQueue 实例，数据流会以多个分包 (bytes) 入队，最后以 None 表示结束
:param chunk_size: 消费者每次想要获取的数据块大小
:param max_size: 需要最终补齐的总大小（单位：字节），必须是 chunk_size 的整数倍
:return: 生成器，按 chunk_size 大小分批次产出数据
"""
if max_size % chunk_size != 0:
raise ValueError("max_size must be a multiple of chunk_size")
total_chunks = max_size // chunk_size
chunks_yielded = 0
current_buffer = bytearray()
save_file = None
if self.save_dir:
save_file = open(self.save_path, "wb")
while chunks_yielded < total_chunks:
try:
data = simple_queue.get(timeout=10)
except queue.Empty:
break
if data is None:
# 数据流结束，用0x00填充剩余的块
remaining_chunks = total_chunks - chunks_yielded
if remaining_chunks == total_chunks:
# print("空包跳过")
break
if len(current_buffer) > 0:
# 处理当前缓冲区中的最后一块数据
padding_size = chunk_size - len(current_buffer)
if padding_size > 0:
current_buffer += b'\x00' * padding_size
logger.info(f"最后一个包差了 {padding_size} 个字节")
yield bytes(current_buffer)
chunks_yielded += 1
remaining_chunks -= 1
break
logger.info(f"还差 {remaining_chunks} 个完整包")
# 输出剩余的全0块
for _ in range(remaining_chunks):
yield b'\x00' * chunk_size
chunks_yielded += 1
break
save_file and save_file.write(data)
# 将新数据添加到缓冲区
current_buffer.extend(data)
# 输出完整的块
while len(current_buffer) >= chunk_size and chunks_yielded < total_chunks:
yield bytes(current_buffer[:chunk_size])
current_buffer = current_buffer[chunk_size:]
chunks_yielded += 1
# print("本段分p完成")
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"]
# 不能提交 extra_fields 字段，提前处理
post_data = asdict(self.video)
if post_data.get('extra_fields'):
for key, value in json.loads(post_data.pop('extra_fields')).items():
post_data.setdefault(key, value)
self.__session.get('https://member.bilibili.com/x/geetest/pre/add', timeout=5)
if submit_api is None:
total_info = self.myinfo()
if total_info.get('data') is None:
logger.error(total_info)
total_info = total_info.get('data')
if total_info['level'] > 3 and total_info['follower'] > 1000:
user_weight = 2
else:
user_weight = 1
logger.info(f'推测的用户权重: {user_weight}')
# submit_api = 'web' if user_weight == 2 else 'client'
# web 目前（2025-01-26）全量分p功能
submit_api = 'web'
ret = None
if submit_api == 'web':
ret = self.submit_web(post_data, edit=edit)
if ret["code"] == 21138:
logger.info(f'改用客户端接口提交{ret}')
submit_api = 'client'
if submit_api == 'client':
ret = self.submit_client(post_data, edit=edit)
if not ret:
raise Exception(f'不存在的选项：{submit_api}')
if ret["code"] == 0:
return ret
else:
raise Exception(ret)
def submit_web(self, post_data, edit=False):
logger.info('使用网页端api提交')
if not self.__bili_jct:
raise RuntimeError("bili_jct is required!")
api = 'https://member.bilibili.com/x/vu/web/add?csrf=' + self.__bili_jct
if edit:
api = 'https://member.bilibili.com/x/vu/web/edit?csrf=' + self.__bili_jct
return self.__session.post(api, timeout=5,
json=post_data).json()
def submit_client(self, post_data, edit=False):
logger.info('使用客户端api端提交')
if not self.access_token:
if self.account is None:
raise RuntimeError("Access token is required, but account and access_token does not exist!")
self.login_by_password(**self.account)
self.store()
api = 'http://member.bilibili.com/x/vu/client/add?access_key=' + self.access_token
if edit:
api = 'http://member.bilibili.com/x/vu/client/edit?access_key=' + self.access_token
logger.debug(f"client api submit: {post_data}")
while True:
ret = self.__session.post(api, timeout=5, json=post_data).json()
if ret['code'] == -101:
logger.info(f'刷新token{ret}')
self.login_by_password(**config['user']['account'])
self.store()
continue
return ret
def cover_up(self, img: str):
"""
:param img: img path or stream
:return: img URL
"""
from PIL import Image
from io import BytesIO
with Image.open(img) as im:
# 宽和高,需要16：10
xsize, ysize = im.size
if xsize / ysize > 1.6:
delta = xsize - ysize * 1.6
region = im.crop((delta / 2, 0, xsize - delta / 2, ysize))
else:
delta = ysize - xsize * 10 / 16
region = im.crop((0, delta / 2, xsize, ysize - delta / 2))
buffered = BytesIO()
region.save(buffered, format=im.format)
r = self.__session.post(
url='https://member.bilibili.com/x/vu/web/cover/up',
data={
'cover': b'data:image/jpeg;base64,' + (base64.b64encode(buffered.getvalue())),
'csrf': self.__bili_jct
}, timeout=30
)
buffered.close()
res = r.json()
if res.get('data') is None:
raise Exception(res)
return res['data']['url']
def get_tags(self, upvideo, typeid="", desc="", cover="", groupid=1, vfea=""):
"""
上传视频后获得推荐标签
:param vfea:
:param groupid:
:param typeid:
:param desc:
:param cover:
: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}'
return self.__session.get(url=url, timeout=5).json()
def __enter__(self):
return self
def __exit__(self, e_t, e_v, t_b):
self.close()
def close(self):
"""Closes all adapters and as such the session"""
self.__session.close()
@dataclass
class Data:
"""
cover: 封面图片，可由recovers方法得到视频的帧截图
"""
copyright: int = 2
source: str = ''
tid: int = 21
cover: str = ''
title: str = ''
desc_format_id: int = 0
desc: str = ''
desc_v2: list = field(default_factory=list)
dynamic: str = ''
subtitle: dict = field(init=False)
tag: Union[list, str] = ''
videos: list = field(default_factory=list)
dtime: Any = None
open_subtitle: InitVar[bool] = False
dolby: int = 0
hires: int = 0
no_reprint = 0
open_elec = 0
extra_fields: str = ""
aid: int = None
# interactive: int = 0
# no_reprint: int 1
# open_elec: int 1
def __post_init__(self, open_subtitle):
self.subtitle = {"open": int(open_subtitle), "lan": ""}
if self.dtime and self.dtime - int(time.time()) <= 14400:
self.dtime = None
if isinstance(self.tag, list):
self.tag = ','.join(self.tag)
def delay_time(self, dtime: int):
"""设置延时发布时间，距离提交大于2小时，格式为10位时间戳"""
if dtime - int(time.time()) > 7200:
self.dtime = dtime
def set_tag(self, tag: list):
"""设置标签，tag为数组"""
self.tag = ','.join(tag)
def append(self, video):
self.videos.append(video)
biliup/plugins/bilibili.py
+34 -33
View File
@@ -22,21 +22,17 @@ class Bililive(DownloadBase):
self.bilibili_danmaku = config.get('bilibili_danmaku', False)
self.bilibili_danmaku_detail = config.get('bilibili_danmaku_detail', False)
self.__real_room_id = None
self.__is_login = False
self.__bili_uid = 0
# self.fake_headers['referer'] = url
self.__login_mid = 0
if config.get('user', {}).get('bili_cookie'):
self.fake_headers['cookie'] = config.get('user', {}).get('bili_cookie')
if config.get('user', {}).get('bili_cookie_file'):
cookie_file_name = config.get('user', {}).get('bili_cookie_file')
try:
with open(cookie_file_name, encoding='utf-8') as stream:
user_data = json.load(stream)
cookies = user_data["cookie_info"]["cookies"]
cookies = json.load(stream)["cookie_info"]["cookies"]
cookies_str = ''
for i in cookies:
cookies_str += f"{i['name']}={i['value']};"
self.__bili_uid = user_data['token_info']['mid']
self.fake_headers['cookie'] = cookies_str
except Exception:
logger.exception("load_cookies error")
@@ -58,6 +54,7 @@ class Bililive(DownloadBase):
room_id = match1(self.url, r'bilibili.com/(\d+)')
quality_number = int(config.get('bili_qn', 10000))
self.fake_headers['referer'] = self.url
# 获取直播状态与房间标题
info_by_room_url = f"{OFFICIAL_API}/xlive/web-room/v1/index/getInfoByRoom?room_id={room_id}"
@@ -89,8 +86,9 @@ class Bililive(DownloadBase):
is_new_live = False
if is_check:
self.__is_login = await self.check_login_status()
return True
else:
self.__login_mid = await self.check_login_status()
# 复用原画 m3u8 流
if self.raw_stream_url is not None \
@@ -144,23 +142,27 @@ class Bililive(DownloadBase):
stream_url = stream_info['url']
logger.debug(f"{self.plugin_msg}: 使用 {current_cdn} 流")
self.raw_stream_url = normalize_cn204(
f"{stream_url['host']}{stream_url['base_url']}{stream_url['extra']}"
)
# 替换 cn-gotcha01 节点
cn01_sids = config.get('bili_replace_cn01', [])
if cn01_sids:
if cn01_sids and "cn-gotcha01" in current_cdn:
if isinstance(cn01_sids, str): cn01_sids = cn01_sids.split(',')
if "cn-gotcha01" in current_cdn:
for sid in cn01_sids:
__host = f"https://{sid}.bilivideo.com"
__url = f"{__host}{stream_url['base_url']}{stream_url['extra']}"
if (await self.acheck_url_healthy(__url)) is not None:
stream_url['host'] = __host
break
else:
logger.debug(f"{self.plugin_msg}: {sid} is not available")
for sid in cn01_sids:
new_url = f"https://{sid}.bilivideo.com{stream_url['base_url']}{stream_url['extra']}"
new_url = await self.acheck_url_healthy(new_url)
if new_url is not None:
self.raw_stream_url = new_url
break
else:
logger.debug(f"{self.plugin_msg}: {sid} is not available")
# 强制原画
if quality_number == 10000 and config.get('bili_force_source'):
if stream_info['suffix'] != 'origin':
if quality_number <= 10000 and config.get('bili_force_source'):
# 不处理 qn 20000
if stream_info['suffix'] not in {'origin', 'uhd', 'maxhevc'}:
__base_url = stream_url['base_url'].replace(f"_{stream_info['suffix']}", "")
__sk = match1(stream_url['extra'], r'sk=([^&]+)')
__extra = stream_url['extra'].replace(__sk, __sk[:32])
@@ -173,10 +175,6 @@ class Bililive(DownloadBase):
f"{self.plugin_msg}: force_source 处理 {current_cdn} 失败 {stream_info['stream_name']}"
)
if not self.raw_stream_url:
self.raw_stream_url = normalize_cn204(
f"{stream_url['host']}{stream_url['base_url']}{stream_url['extra']}"
)
if config.get('bili_cdn_fallback'):
__url = await self.acheck_url_healthy(self.raw_stream_url)
@@ -208,13 +206,14 @@ class Bililive(DownloadBase):
if self.bilibili_danmaku:
self.danmaku = DanmakuClient(
self.url, self.gen_download_filename(), {
'room_id': self.__real_room_id,
'cookie': self.fake_headers.get("cookie"),
'room_id': self.__real_room_id,
'cookie': self.fake_headers.get('cookie', ''),
'detail': self.bilibili_danmaku_detail,
'bili_uid': self.__bili_uid
'uid': self.__login_mid
}
)
async def get_play_info(self, api: str, qn: int = 10000) -> dict:
full_url = f"{api}/xlive/web-room/v2/index/getRoomPlayInfo"
params = {
@@ -248,7 +247,7 @@ class Bililive(DownloadBase):
full_url = f"{api}/live-bvc/master.m3u8"
params = {
"cid": self.__real_room_id,
# "mid": {self.__login_info.get('mid')},
"mid": self.__login_mid,
"pt": "html5", # platform
# "p2p_type": "-1",
}
@@ -278,8 +277,7 @@ class Bililive(DownloadBase):
streams = play_info['playurl_info']['playurl']['stream']
if protocol == 'hls_fmp4':
if config.get('bili_anonymous_origin'):
# 开播检测运行在主线程，拿不到保存的登录状态
if special_type in play_info['all_special_types']:
if special_type in play_info['all_special_types'] and not self.__login_mid:
logger.warn(f"{self.plugin_msg}: 特殊直播{special_type}")
else:
stream_urls = await self.get_master_m3u8(api)
@@ -302,19 +300,22 @@ class Bililive(DownloadBase):
# 空字典照常返回，重试交给上层方法处理
return stream_urls
async def check_login_status(self):
"""检查B站登录状态"""
async def check_login_status(self) -> int:
"""
检查B站登录状态
:return: 当前登录用户 mid
"""
try:
_res = await client.get('https://api.bilibili.com/x/web-interface/nav', headers=self.fake_headers)
user_data = json.loads(_res.text).get('data', {})
if user_data.get('isLogin'):
logger.info(f"用户名：{user_data['uname']}, mid：{user_data['mid']}, isLogin：{user_data['isLogin']}")
return True
return user_data['mid']
else:
logger.warning(f"{self.plugin_msg}: 未登录，或将只能录制到最低画质。")
except Exception as e:
logger.error(f"{self.plugin_msg}: 登录态校验失败 {e}")
return False
return 0
# Copy from room-player.js
def check_areablock(data):
biliup/plugins/biliuprs.py
+4 -3
View File
@@ -14,7 +14,7 @@ class BiliWeb(UploadBase):
self, principal, data, submit_api=None, copyright=2, postprocessor=None, dtime=None,
dynamic='', lines='AUTO', threads=3, tid=122, tags=None, cover_path=None, description='',
dolby=0, hires=0, no_reprint=0, open_elec=0, credits=None,
user_cookie='cookies.json', copyright_source=None
user_cookie='cookies.json', copyright_source=None, extra_fields = ""
):
super().__init__(principal, data, persistence_path='bili.cookie', postprocessor=postprocessor)
if tags is None:
@@ -43,7 +43,8 @@ class BiliWeb(UploadBase):
self.open_elec = open_elec
self.user_cookie = user_cookie
self.copyright_source = copyright_source
self.extra_fields = extra_fields
def upload(self, file_list: List[UploadBase.FileInfo]) -> List[UploadBase.FileInfo]:
if self.credits:
desc_v2 = self.creditsToDesc_v2()
@@ -74,9 +75,9 @@ class BiliWeb(UploadBase):
"open_elec": self.open_elec,
"limit": self.threads,
"desc_v2": desc_v2,
"extra_fields": self.extra_fields,
"dtime": int(time.time() + self.dtime) if self.dtime else None,
}
upload_process = mp.get_context('spawn').Process(target=stream_gears_upload, daemon=True, kwargs=upload_args)
upload_process.start()
upload_process.join()
biliup/plugins/inke.py
+1 -1
View File
@@ -9,7 +9,7 @@ from ..engine.download import DownloadBase
@Plugin.download(regexp=r'(?:https?://)?(?:(?:www)\.)?inke\.cn')
class inke(DownloadBase):
class Inke(DownloadBase):
def __init__(self, fname, url, suffix='flv'):
super().__init__(fname, url, suffix)
biliup/plugins/stream_gears.py
+6 -2
View File
@@ -13,7 +13,7 @@ class BiliWeb(UploadBase):
self, principal, data, submit_api=None, copyright=2, postprocessor=None, dtime=None,
dynamic='', lines='AUTO', threads=3, tid=122, tags=None, cover_path=None, description='',
dolby=0, hires=0, no_reprint=0, open_elec=0, credits=None,
user_cookie='cookies.json', copyright_source=None
user_cookie='cookies.json', copyright_source=None, extra_fields = ""
):
super().__init__(principal, data, persistence_path='bili.cookie', postprocessor=postprocessor)
if credits is None:
@@ -43,6 +43,9 @@ class BiliWeb(UploadBase):
self.user_cookie = user_cookie
self.copyright_source = copyright_source
self.extra_fields = extra_fields
def upload(self, file_list) -> List[UploadBase.FileInfo]:
line = None
if self.lines == 'kodo':
@@ -90,7 +93,8 @@ class BiliWeb(UploadBase):
self.threads,
desc_v2,
dtime,
line
line,
self.extra_fields
)
logger.info(f"上传成功: {self.principal}")
return file_list
biliup/web/__init__.py
+1
View File
@@ -480,6 +480,7 @@ async def m_upload(request):
json_data = await request.json()
json_data['params']['uploader'] = 'stream_gears'
json_data['params']['name'] = json_data['params']['template_name']
# json_data['params']['extra_fields'] = "{\"is_only_self\": 1}"
threading.Thread(target=biliup_uploader, args=(json_data['files'], json_data['params'])).start()
return web.json_response({'status': 'ok'})