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!(