From 247f231640b8dd566976551656343fd38f9977bb Mon Sep 17 00:00:00 2001
From: ForgQi <34411314+ForgQi@users.noreply.github.com>
Date: Tue, 30 Sep 2025 01:44:46 +0800
Subject: [PATCH] Refactor log tabs and improve download/upload handling
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.
---
Cargo.lock | 4 +-
Cargo.toml | 2 +-
app/(app)/logviewer/page.tsx | 8 +-
biliup/engine/download.py | 11 --
biliup/engine/upload.py | 135 ------------------
.../biliup-cli/src/server/common/download.rs | 2 +-
.../src/server/core/download_manager.rs | 3 +-
crates/biliup-cli/src/server/core/monitor.rs | 14 +-
.../server/infrastructure/service_register.rs | 3 +-
crates/stream-gears/src/server.rs | 22 ++-
10 files changed, 35 insertions(+), 169 deletions(-)
diff --git a/Cargo.lock b/Cargo.lock
index d17ac20..843353f 100644
--- a/Cargo.lock
+++ b/Cargo.lock
@@ -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",
diff --git a/Cargo.toml b/Cargo.toml
index 2aaaf7c..33fd5e4 100644
--- a/Cargo.toml
+++ b/Cargo.toml
@@ -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"] }
diff --git a/app/(app)/logviewer/page.tsx b/app/(app)/logviewer/page.tsx
index 036a88b..6c4c23f 100644
--- a/app/(app)/logviewer/page.tsx
+++ b/app/(app)/logviewer/page.tsx
@@ -249,12 +249,12 @@ export default function LogViewer() {
-
-
-
-
+
+ {/**/}
+ {/* */}
+ {/**/}
diff --git a/biliup/engine/download.py b/biliup/engine/download.py
index 49ad0c1..6c24fa2 100644
--- a/biliup/engine/download.py
+++ b/biliup/engine/download.py
@@ -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):
# 获取封面
diff --git a/biliup/engine/upload.py b/biliup/engine/upload.py
index b90f2bc..fcb620b 100644
--- a/biliup/engine/upload.py
+++ b/biliup/engine/upload.py
@@ -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()
diff --git a/crates/biliup-cli/src/server/common/download.rs b/crates/biliup-cli/src/server/common/download.rs
index a607696..15cb25c 100644
--- a/crates/biliup-cli/src/server/common/download.rs
+++ b/crates/biliup-cli/src/server/common/download.rs
@@ -346,7 +346,7 @@ impl DownloadTask {
async fn start_danmaku(&self, client: &Arc) -> AppResult<()> {
// 启动弹幕下载逻辑
- println!(
+ info!(
"Starting danmaku client for stream: {}",
self.ctx.stream_info.streamer_info.url
);
diff --git a/crates/biliup-cli/src/server/core/download_manager.rs b/crates/biliup-cli/src/server/core/download_manager.rs
index 687e260..6bc969f 100644
--- a/crates/biliup-cli/src/server/core/download_manager.rs
+++ b/crates/biliup-cli/src/server/core/download_manager.rs
@@ -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);
}
}
diff --git a/crates/biliup-cli/src/server/core/monitor.rs b/crates/biliup-cli/src/server/core/monitor.rs
index 623ad96..cbf0465 100644
--- a/crates/biliup-cli/src/server/core/monitor.rs
+++ b/crates/biliup-cli/src/server/core/monitor.rs
@@ -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 _ =` 忽略发送时的任何错误
diff --git a/crates/biliup-cli/src/server/infrastructure/service_register.rs b/crates/biliup-cli/src/server/infrastructure/service_register.rs
index ff1c7b3..bce7e4a 100644
--- a/crates/biliup-cli/src/server/infrastructure/service_register.rs
+++ b/crates/biliup-cli/src/server/infrastructure/service_register.rs
@@ -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(()))
}
diff --git a/crates/stream-gears/src/server.rs b/crates/stream-gears/src/server.rs
index 5605c4c..c686dd2 100644
--- a/crates/stream-gears/src/server.rs
+++ b/crates/stream-gears/src/server.rs
@@ -72,7 +72,7 @@ impl DownloadPlugin for PyPlugin {
}
async fn check_status(&self, ctx: &mut Context) -> Result> {
- 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>,
- url: &str,
+ ctx: &mut Context,
) -> PyResult>)>> {
- 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>)>> {
@@ -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::() {
+ 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!(