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

Refactor log tabs and improve download/upload handling

Browse Source 
Merged download and upload logs into a single tab in the log viewer and commented out the separate upload log tab. Removed unused and legacy methods from the Python upload and download engine classes. Improved logging and error handling in Rust server modules, including more informative log messages and better context propagation. Updated plugin interface to pass context instead of just URL, and enhanced dictionary filtering in config state. Bumped workspace version to 1.1.12.
This commit is contained in:
ForgQi committed 2025-09-30 01:44:46 +08:00
1 parent 3e76841206
commit 247f231640
10 files changed
+35 -169
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

Cargo.lock
Generated
+2 -2
View File
@@ -395,7 +395,7 @@ dependencies = [
[[package]]
name = "biliup-cli"
version = "1.1.11"
version = "1.1.12"
dependencies = [
"async-channel",
"async-trait",
@@ -3641,7 +3641,7 @@ checksum = "a8f112729512f8e442d81f95a8a7ddf2b7c6b8a1a6f509a95864142b30cab2d3"
[[package]]
name = "stream-gears"
version = "1.1.11"
version = "1.1.12"
dependencies = [
"async-channel",
"async-trait",
Cargo.toml
+1 -1
View File
@@ -7,7 +7,7 @@ members = [
resolver = "3"
[workspace.package]
version = "1.1.11"
version = "1.1.12"
[workspace.dependencies]
serde = { version = "1.0", features = ["derive"] }
app/(app)/logviewer/page.tsx
+4 -4
View File
@@ -249,12 +249,12 @@ export default function LogViewer() {
<TabPane tab="主程序运行日志" itemKey="ds_update">
<LogContent logs={logs} logContainerRef={logContainerRef} isLoading={isLoading} />
</TabPane>
<TabPane tab="stream-gears下载日志" itemKey="download">
<LogContent logs={logs} logContainerRef={logContainerRef} isLoading={isLoading} />
</TabPane>
<TabPane tab="biliup-rs上传日志" itemKey="upload">
<TabPane tab="biliup下载和上传日志" itemKey="download">
<LogContent logs={logs} logContainerRef={logContainerRef} isLoading={isLoading} />
</TabPane>
{/*<TabPane tab="biliup-rs上传日志" itemKey="upload">*/}
{/* <LogContent logs={logs} logContainerRef={logContainerRef} isLoading={isLoading} />*/}
{/*</TabPane>*/}
</Tabs>
</Card>
</Content>
biliup/engine/download.py
-11
View File
@@ -327,17 +327,6 @@ class DownloadBase(ABC):
def download_success_callback(self):
pass
def run(self):
try:
self.danmaku_init()
if self.danmaku:
self.danmaku.start()
retval = self.download()
return retval
finally:
if self.danmaku:
self.danmaku.stop()
self.danmaku = None
def download_cover(self, fmtname):
# 获取封面
biliup/engine/upload.py
-135
View File
@@ -1,12 +1,7 @@
import logging
import os
import pathlib
import shutil
from typing import NamedTuple, Optional, List
from sqlalchemy import desc
logger = logging.getLogger('biliup')
@@ -21,136 +16,6 @@ class UploadBase:
self.data: dict = data
self.post_processor = postprocessor
# 此方法会返回文件列表中上传次数最大文件的上传次数
@staticmethod
def get_max_upload_count(filelist: List[FileInfo]) -> int:
from biliup.handler import event_manager
with NamedLock("file_upload_count"):
file_upload_count = event_manager.context['file_upload_count']
max = 0
for item in filelist:
current = file_upload_count.setdefault(item.video, 0)
max = current if current > max else max
file_upload_count[item.video] += 1
return max
@staticmethod
def file_list(index) -> List[FileInfo]:
from biliup.handler import event_manager
media_extensions = ['.mp4', '.flv', '.3gp', '.webm', '.mkv', '.ts']
# 获取文件列表
file_list = []
# 数据库中保存的文件名, 不含后缀
save = []
with SessionLocal() as db:
dbinfo = db.query(models.StreamerInfo).filter(models.StreamerInfo.name == index).order_by(
desc(models.StreamerInfo.id)).first()
if dbinfo:
for dbfile in dbinfo.filelist:
save.append(dbfile.file)
for file_name in os.listdir('.'):
# 可能有两层后缀.with_suffix('')去掉一层.stem取文件名
if (index in file_name or pathlib.Path(file_name).with_suffix('').stem in save ) and os.path.isfile(file_name):
file_list.append(file_name)
if len(file_list) == 0:
return []
file_list = sorted(file_list, key=lambda x: get_file_create_timestamp(x))
# 正在上传的文件列表
upload_filename: list = event_manager.context['upload_filename']
results = []
for index, file in enumerate(file_list):
old_name = file
if file.endswith('.part'):
file_list[index] = os.path.splitext(file)[0]
file = os.path.splitext(file)[0]
name, ext = os.path.splitext(file)
# 过滤正在上传的
if name in upload_filename:
continue
# 过滤不是视频的
if ext not in media_extensions:
continue
if old_name != file:
logger.info(f'{old_name} 已更名为 {file}')
shutil.move(old_name, file)
file_size = os.path.getsize(file) / 1024 / 1024
threshold = config.get('filtering_threshold', 0)
if file_size <= threshold:
os.remove(file)
logger.info(f'过滤删除 - {file}')
continue
video = file
danmaku = None
if f'{name}.xml' in file_list:
danmaku = f'{name}.xml'
result = UploadBase.FileInfo(video=video, danmaku=danmaku)
results.append(result)
# 过滤弹幕
for file in file_list:
name, ext = os.path.splitext(file)
# 过滤正在上传的
if name in upload_filename:
continue
if ext == '.xml':
have_video = False
for result in results:
if result.danmaku == file:
have_video = True
break
if not have_video:
logger.info(f'无视频，过滤删除 - {file}')
UploadBase.remove_file(file)
return results
@staticmethod
def remove_filelist(file_list: List[FileInfo]):
for f in file_list:
UploadBase.remove_file(f.video)
if f.danmaku is not None:
UploadBase.remove_file(f.danmaku)
@staticmethod
def remove_file(file: str):
try:
os.remove(file)
logger.info(f'删除 - {file}')
except:
logger.warning(f'删除失败 - {file}')
def upload(self, file_list: List[FileInfo]) -> List[FileInfo]:
raise NotImplementedError()
def start(self):
from biliup.handler import event_manager
# 保证一个name同时只有一个上传线程扫描文件列表
lock = NamedLock(f'upload_file_list_{self.principal}')
upload_filename_list = []
try:
lock.acquire()
file_list = UploadBase.file_list(self.principal)
if len(file_list) > 0:
upload_filename_list = [os.path.splitext(file.video)[0] for file in file_list]
logger.info('准备上传' + self.data["format_title"])
with NamedLock('upload_filename'):
event_manager.context['upload_filename'].extend(upload_filename_list)
lock.release()
file_list = self.upload(file_list)
return file_list
finally:
with NamedLock('upload_filename'):
event_manager.context['upload_filename'] = list(
set(event_manager.context['upload_filename']) - set(upload_filename_list))
if lock.locked():
lock.release()
crates/biliup-cli/src/server/common/download.rs
+1 -1
View File
@@ -346,7 +346,7 @@ impl DownloadTask {
async fn start_danmaku(&self, client: &Arc<dyn Downloader>) -> AppResult<()> {
// 启动弹幕下载逻辑
println!(
info!(
"Starting danmaku client for stream: {}",
self.ctx.stream_info.streamer_info.url
);
crates/biliup-cli/src/server/core/download_manager.rs
+2 -1
View File
@@ -190,10 +190,11 @@ impl UActor {
}
};
if let Err(e) = result {
if let Err(e) = &result {
error!("Process segment event failed: {}", e);
// 可以添加错误通知机制
}
info!(url=ctx.stream_info.streamer_info.url, result=?result, "后处理执行完毕：Finished processing segment event");
ctx.worker.change_status(Stage::Upload, WorkerStatus::Idle);
}
}
crates/biliup-cli/src/server/core/monitor.rs
+8 -6
View File
@@ -35,8 +35,8 @@ async fn start_client(
interval = room.get_config().event_loop_interval;
let mut ctx = Context::new(room.clone(), Default::default(), pool.clone());
// 检查直播状态
match plugin.check_status(&mut ctx).await.unwrap() {
StreamStatus::Live { mut stream_info } => {
match plugin.check_status(&mut ctx).await {
Ok(StreamStatus::Live { mut stream_info }) => {
let sql_no_id = &stream_info.streamer_info;
let insert = match StreamerInfo::builder()
.url(sql_no_id.url.clone())
@@ -57,12 +57,13 @@ async fn start_client(
// 更新状态为等待中
room.change_status(Stage::Download, WorkerStatus::Pending);
stream_info.streamer_info = insert;
ctx.stream_info = stream_info;
// 发送下载开始消息
if actor_handle
.down_sender
.send(DownloaderMessage::Start(
plugin.clone(),
Context::new(room.clone(), stream_info, pool.clone()),
ctx,
rooms_handle.clone(),
))
.await
@@ -72,8 +73,9 @@ async fn start_client(
rooms_handle.toggle(room).await;
}
}
StreamStatus::Offline => {}
StreamStatus::Unknown => {}
Ok(StreamStatus::Offline) => {}
Ok(StreamStatus::Unknown) => {},
Err(e) => error!(e=?e, ctx=ctx.worker.live_streamer.url,"检查直播间出错")
};
}
// 等待下一次检查
@@ -272,8 +274,8 @@ impl RoomsActor {
let _ = respond_to.send(self.next());
}
ActorMessage::Add(worker) => {
info!("Added room [{}]", worker.live_streamer.url);
self.rooms.push(worker);
info!("Added room [{:?}]", self.rooms);
}
ActorMessage::Del { respond_to, id } => {
// `let _ =` 忽略发送时的任何错误
crates/biliup-cli/src/server/infrastructure/service_register.rs
+1 -2
View File
@@ -49,7 +49,6 @@ impl ServiceRegister {
// 创建默认的HTTP客户端
let client = StatelessClient::default();
info!(config=?config);
info!("utility services initialized, building feature services...");
@@ -102,7 +101,7 @@ impl ServiceRegister {
let monitor = manager.ensure_monitor(self.pool.clone());
monitor.rooms_handle.add(worker.clone()).await;
self.workers.write().unwrap().push(worker.clone());
info!("add {worker:?} success");
info!("add {} success", worker.live_streamer.url);
Ok(Some(()))
}
crates/stream-gears/src/server.rs
+16 -6
View File
@@ -72,7 +72,7 @@ impl DownloadPlugin for PyPlugin {
}
async fn check_status(&self, ctx: &mut Context) -> Result<StreamStatus, Report<AppError>> {
match call_via_threads(self.plugin.clone(), &ctx.worker.live_streamer.url)
match call_via_threads(self.plugin.clone(), ctx)
.await
.change_context(AppError::Unknown)?
{
@@ -141,9 +141,10 @@ impl DownloadPlugin for PyPlugin {
async fn call_via_threads(
obj: Arc<Py<PyType>>,
url: &str,
ctx: &mut Context,
) -> PyResult<Option<(StreamInfoExt, Option<Py<PyAny>>)>> {
let url = url.to_string();
let url = ctx.worker.live_streamer.url.to_string();
let remark = ctx.worker.live_streamer.remark.to_string();
tokio::task::spawn_blocking(move || {
Python::attach(
|py| -> PyResult<Option<(StreamInfoExt, Option<Py<PyAny>>)>> {
@@ -160,7 +161,7 @@ async fn call_via_threads(
let asyncio = PyModule::import(py, "asyncio")?;
// 生成协程 self.acheck_stream()
let instance = obj.bind(py).call1(("fname", url))?;
let instance = obj.bind(py).call1((remark, url))?;
let coro = instance.call_method0("acheck_stream")?;
// 调度到指定 loop
@@ -238,7 +239,6 @@ pub fn stream_info_from_py(
} else {
None
};
// date 直接使用传入的 start_time（保留为 Python 对象）
// let date = OffsetDateTime::now_utc();
@@ -300,7 +300,17 @@ impl ConfigState {
{
return Ok(d);
}
return Ok(bound);
// 尝试转换为字典并过滤
return match bound.downcast::<PyDict>() {
Ok(dict) => {
let filtered = PyDict::new(py);
dict.iter()
.filter(|(_, v)| !v.is_none())
.try_for_each(|(k, v)| filtered.set_item(k, v))?;
Ok(filtered.into_any())
}
Err(_) => Ok(bound), // 不是字典，直接返回
};
};
let Some(default) = default else {
return Err(pyo3::exceptions::PyAttributeError::new_err(format!(