From d4b6a5fb220439360893f43aee533aa5a010a0fa Mon Sep 17 00:00:00 2001 From: ForgQi <34411314+ForgQi@users.noreply.github.com> Date: Sat, 27 Sep 2025 16:42:53 +0800 Subject: [PATCH] Refactor models and context for streamer separation Split LiveStreamer and UploadStreamer models into separate modules and update references throughout the codebase. Refactor StreamInfo to StreamInfoExt, embedding StreamerInfo for improved database integration. Update context, plugin, monitor, and related logic to support new model structure and database operations. Bump version to 1.1.9. --- Cargo.lock | 4 +- Cargo.toml | 2 +- crates/biliup-cli/src/server/api/endpoints.rs | 14 +- .../biliup-cli/src/server/common/download.rs | 6 +- crates/biliup-cli/src/server/common/upload.rs | 26 ++- .../src/server/core/download_manager.rs | 9 +- crates/biliup-cli/src/server/core/monitor.rs | 31 ++- crates/biliup-cli/src/server/core/plugin.rs | 21 +-- .../src/server/infrastructure/context.rs | 11 +- .../src/server/infrastructure/dto.rs | 2 +- .../src/server/infrastructure/models.rs | 178 +----------------- .../infrastructure/models/live_streamer.rs | 80 ++++++++ .../infrastructure/models/upload_streamer.rs | 88 +++++++++ .../src/server/infrastructure/repositories.rs | 4 +- .../server/infrastructure/service_register.rs | 7 +- crates/stream-gears/src/server.rs | 32 ++-- 16 files changed, 280 insertions(+), 235 deletions(-) create mode 100644 crates/biliup-cli/src/server/infrastructure/models/live_streamer.rs create mode 100644 crates/biliup-cli/src/server/infrastructure/models/upload_streamer.rs diff --git a/Cargo.lock b/Cargo.lock index 49921fa..55ca2f7 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -395,7 +395,7 @@ dependencies = [ [[package]] name = "biliup-cli" -version = "1.1.8" +version = "1.1.9" dependencies = [ "async-channel", "async-trait", @@ -3641,7 +3641,7 @@ checksum = "a8f112729512f8e442d81f95a8a7ddf2b7c6b8a1a6f509a95864142b30cab2d3" [[package]] name = "stream-gears" -version = "1.1.8" +version = "1.1.9" dependencies = [ "async-channel", "async-trait", diff --git a/Cargo.toml b/Cargo.toml index 57fa007..30c2188 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -7,7 +7,7 @@ members = [ resolver = "3" [workspace.package] -version = "1.1.8" +version = "1.1.9" [workspace.dependencies] serde = { version = "1.0", features = ["derive"] } diff --git a/crates/biliup-cli/src/server/api/endpoints.rs b/crates/biliup-cli/src/server/api/endpoints.rs index 52fd022..3f60854 100644 --- a/crates/biliup-cli/src/server/api/endpoints.rs +++ b/crates/biliup-cli/src/server/api/endpoints.rs @@ -3,13 +3,13 @@ use crate::server::common::upload::{build_studio, submit_to_bilibili, upload}; use crate::server::common::util::Recorder; use crate::server::config::Config; use crate::server::core::download_manager::ActorHandle; -use crate::server::errors::{AppError, report_to_response}; +use crate::server::errors::{report_to_response, AppError}; use crate::server::infrastructure::connection_pool::ConnectionPool; use crate::server::infrastructure::context::Worker; use crate::server::infrastructure::dto::LiveStreamerResponse; use crate::server::infrastructure::models::{ - Configuration, FileItem, InsertConfiguration, InsertLiveStreamer, InsertUploadStreamer, - LiveStreamer, StreamerInfo, UploadStreamer, + Configuration, FileItem, InsertConfiguration, + StreamerInfo, }; use crate::server::infrastructure::repositories::{ del_streamer, get_all_streamer, get_upload_config, @@ -19,7 +19,7 @@ use axum::extract::ws::{Message, Utf8Bytes, WebSocket}; use axum::extract::{Path, Query, State, WebSocketUpgrade}; use axum::http::StatusCode; use axum::response::{IntoResponse, Response}; -use axum::{Json, debug_handler}; +use axum::{debug_handler, Json}; use biliup::credential::Credential; use clap::ValueEnum; use error_stack::{Report, ResultExt}; @@ -34,8 +34,10 @@ use std::sync::{Arc, RwLock}; use std::time::{Duration, UNIX_EPOCH}; use tokio::fs; use tokio::io::{AsyncBufReadExt, AsyncReadExt, AsyncSeekExt, AsyncWriteExt, BufReader}; -use tokio::time::{MissedTickBehavior, interval}; +use tokio::time::{interval, MissedTickBehavior}; use tracing::{debug, error, info}; +use crate::server::infrastructure::models::live_streamer::{InsertLiveStreamer, LiveStreamer}; +use crate::server::infrastructure::models::upload_streamer::{InsertUploadStreamer, UploadStreamer}; pub async fn get_streamers_endpoint( State(pool): State, @@ -489,7 +491,7 @@ pub struct PostUploads { params: UploadStreamer, } -#[debug_handler] +// #[debug_handler] pub async fn post_uploads( State(config): State>>, Json(json_data): Json, diff --git a/crates/biliup-cli/src/server/common/download.rs b/crates/biliup-cli/src/server/common/download.rs index e8008f1..a509574 100644 --- a/crates/biliup-cli/src/server/common/download.rs +++ b/crates/biliup-cli/src/server/common/download.rs @@ -2,7 +2,7 @@ use crate::server::common::util::Recorder; use crate::server::core::download_manager::{DownloaderMessage, UploaderMessage}; use crate::server::core::downloader::{DownloadStatus, Downloader, SegmentEvent, SegmentInfo}; use crate::server::core::monitor::RoomsHandle; -use crate::server::core::plugin::{DownloadPlugin, StreamInfo}; +use crate::server::core::plugin::{DownloadPlugin, StreamInfoExt}; use crate::server::errors::{AppError, AppResult}; use crate::server::infrastructure::context::{Context, Stage, Worker, WorkerStatus}; use async_channel::{Receiver, Sender}; @@ -320,7 +320,7 @@ impl DownloadTask { let recorder = Recorder::new( streamer.filename_prefix, &streamer.remark, - &stream_info.title, + &stream_info.streamer_info.title, &suffix, ); // 可选的弹幕客户端 @@ -347,7 +347,7 @@ impl DownloadTask { // 启动弹幕下载逻辑 println!( "Starting danmaku client for stream: {}", - self.ctx.stream_info.url + self.ctx.stream_info.streamer_info.url ); client.download(Box::new(|_| {})).await?; Ok(()) diff --git a/crates/biliup-cli/src/server/common/upload.rs b/crates/biliup-cli/src/server/common/upload.rs index fe8cdf4..df4da47 100644 --- a/crates/biliup-cli/src/server/common/upload.rs +++ b/crates/biliup-cli/src/server/common/upload.rs @@ -4,7 +4,7 @@ use crate::server::core::download_manager::UActor; use crate::server::core::downloader::{SegmentEvent, SegmentInfo}; use crate::server::errors::{AppError, AppResult}; use crate::server::infrastructure::context::{Context, Worker}; -use crate::server::infrastructure::models::UploadStreamer; +use crate::server::infrastructure::models::upload_streamer::UploadStreamer; use crate::server::infrastructure::models::hook_step::process_video; use async_channel::Receiver; use biliup::bilibili::{BiliBili, Credit, ResponseData, Studio, Video}; @@ -20,7 +20,9 @@ use futures::StreamExt; use std::path::{Path, PathBuf}; use std::str::FromStr; use std::time::Instant; +use ormlite::Insert; use tracing::{error, info}; +use crate::server::infrastructure::models::InsertFileItem; // 辅助结构体 struct UploadContext { @@ -54,7 +56,7 @@ pub async fn process_with_upload( let recorder = Recorder::new( upload_context.upload_config.title.clone(), &ctx.worker.live_streamer.remark, - &ctx.stream_info.title, + &ctx.stream_info.streamer_info.title, "", ); let studio = build_studio( @@ -70,7 +72,7 @@ pub async fn process_with_upload( // 4. 执行后处理 if !uploaded_videos.paths.is_empty() { - execute_postprocessor(uploaded_videos.paths, &ctx.worker).await?; + execute_postprocessor(uploaded_videos.paths, &ctx).await?; } Ok(()) @@ -263,9 +265,21 @@ pub(crate) async fn build_studio( Ok(studio) } -pub async fn execute_postprocessor(video_paths: Vec, worker: &Worker) -> AppResult<()> { - if let Some(processor) = worker.get_streamer().postprocessor { - let paths: Vec<&Path> = video_paths.iter().map(|p| p.as_path()).collect(); +pub async fn execute_postprocessor(video_paths: Vec, ctx: &Context) -> AppResult<()> { + if let Some(processor) = ctx.worker.get_streamer().postprocessor { + let paths: Vec<&Path> = video_paths.iter().map(|p| { + let pool = ctx.pool.clone(); + let file = p.display().to_string(); + tokio::spawn(async move { + let result = InsertFileItem { + file, + streamer_info_id: 0, + }.insert(&pool).await; + info!(result=?result); + }); + + p.as_path() + }).collect(); process_video(&paths, &processor).await?; } Ok(()) diff --git a/crates/biliup-cli/src/server/core/download_manager.rs b/crates/biliup-cli/src/server/core/download_manager.rs index 90c41ed..2815df2 100644 --- a/crates/biliup-cli/src/server/core/download_manager.rs +++ b/crates/biliup-cli/src/server/core/download_manager.rs @@ -3,7 +3,7 @@ use crate::server::common::upload::{execute_postprocessor, process_with_upload}; use crate::server::common::util::Recorder; use crate::server::core::downloader::{Downloader, SegmentEvent, SegmentInfo}; use crate::server::core::monitor::{Monitor, RoomsHandle}; -use crate::server::core::plugin::{DownloadPlugin, StreamInfo}; +use crate::server::core::plugin::{DownloadPlugin, StreamInfoExt}; use crate::server::errors::{AppError, AppResult}; use crate::server::infrastructure::context::{Context, Stage, Worker, WorkerStatus}; use crate::server::infrastructure::models::hook_step::process_video; @@ -23,6 +23,7 @@ use std::sync::{Arc, Mutex}; use std::time::Instant; use tokio::task::JoinHandle; use tracing::{error, info, warn}; +use crate::server::infrastructure::connection_pool::ConnectionPool; /// 下载管理器 /// 负责管理特定平台的下载任务，包括监控器和插件 @@ -56,12 +57,12 @@ impl DownloadManager { /// /// # 返回 /// 返回监控器的Arc引用 - pub fn ensure_monitor(&self) -> Arc { + pub fn ensure_monitor(&self, pool: ConnectionPool) -> Arc { self.monitor .lock() .unwrap() .get_or_insert_with(|| { - Arc::new(Monitor::new(self.plugin.clone(), self.actor_handle.clone())) + Arc::new(Monitor::new(self.plugin.clone(), self.actor_handle.clone(), pool)) }) .clone() } @@ -163,7 +164,7 @@ impl UActor { paths.push(event.prev_file_path); } // 无上传配置时，直接执行后处理 - execute_postprocessor(paths, &ctx.worker).await + execute_postprocessor(paths, &ctx).await } }; diff --git a/crates/biliup-cli/src/server/core/monitor.rs b/crates/biliup-cli/src/server/core/monitor.rs index 8ea614d..23c15cd 100644 --- a/crates/biliup-cli/src/server/core/monitor.rs +++ b/crates/biliup-cli/src/server/core/monitor.rs @@ -4,9 +4,13 @@ use crate::server::infrastructure::context::{Context, Stage, Worker, WorkerStatu use async_channel::{Receiver, Sender, bounded}; use std::sync::Arc; use std::time::Duration; +use ormlite::Model; +use ormlite::model::ModelBuilder; use tokio::sync::oneshot; use tokio::task::JoinHandle; -use tracing::info; +use tracing::{error, info}; +use crate::server::infrastructure::connection_pool::ConnectionPool; +use crate::server::infrastructure::models::StreamerInfo; /// 启动客户端监控循环 /// @@ -19,6 +23,7 @@ async fn start_client( rooms_handle: Arc, plugin: Arc, actor_handle: Arc, + pool: ConnectionPool, mut interval: u64, ) { let platform_name = &rooms_handle.name; @@ -29,19 +34,34 @@ async fn start_client( let ulr = room.get_streamer().url; interval = room.get_config().event_loop_interval; info!("[{platform_name}] room: {ulr}"); - let mut ctx = Context::new(room.clone(), Default::default()); + let mut ctx = Context::new(room.clone(), Default::default(), pool.clone()); // 检查直播状态 match plugin.check_status(&mut ctx).await.unwrap() { - StreamStatus::Live { stream_info } => { + StreamStatus::Live { mut stream_info } => { + let sql_no_id = &stream_info.streamer_info; + let insert = match StreamerInfo::builder() + .url(sql_no_id.url.clone()) + .name(room.live_streamer.remark.clone()) + .title(sql_no_id.title.clone()) + .date(sql_no_id.date.clone()) + .live_cover_path(sql_no_id.live_cover_path.clone()) + .insert(&ctx.pool).await { + Ok(insert) => {insert} + Err(e) => { + error!(e=?e, "插入数据库失败"); + continue + } + }; info!("room: {ulr} is live -> {:?}", stream_info); // 更新状态为等待中 room.change_status(Stage::Download, WorkerStatus::Pending); + stream_info.streamer_info = insert; // 发送下载开始消息 if actor_handle .down_sender .send(DownloaderMessage::Start( plugin.clone(), - Context::new(room.clone(), stream_info), + Context::new(room.clone(), stream_info, pool.clone()), rooms_handle.clone(), )) .await @@ -79,6 +99,7 @@ impl Monitor { pub fn new( plugin: Arc, actor_handle: Arc, + pool: ConnectionPool ) -> Self { // 创建房间处理器 let handle = Arc::new(RoomsHandle::new(plugin.name())); @@ -86,7 +107,7 @@ impl Monitor { let join_handle = tokio::spawn({ let handle = Arc::clone(&handle); async move { - start_client(handle, plugin, actor_handle, 10).await; + start_client(handle, plugin, actor_handle, pool,10).await; } }); Self { diff --git a/crates/biliup-cli/src/server/core/plugin.rs b/crates/biliup-cli/src/server/core/plugin.rs index a0b5202..098e762 100644 --- a/crates/biliup-cli/src/server/core/plugin.rs +++ b/crates/biliup-cli/src/server/core/plugin.rs @@ -11,24 +11,15 @@ use std::collections::HashMap; use std::path::Path; use std::sync::Arc; use time::OffsetDateTime; +use crate::server::infrastructure::models::StreamerInfo; /// 流信息结构 /// 包含直播流的详细信息 #[derive(Debug, Clone, Serialize, Deserialize, Default)] -pub struct StreamInfo { - /// 主播名称 - pub name: String, - /// 直播间URL - pub url: String, +pub struct StreamInfoExt { + pub streamer_info: StreamerInfo, /// 原始流URL pub raw_stream_url: String, - /// 直播标题 - pub title: String, - /// 直播开始时间 - // #[serde(with = "time::serde::rfc3339")] - pub date: DateTime, - /// 直播封面路径（可选） - pub live_cover_path: Option, /// 平台名称 pub platform: String, /// 流请求头 @@ -42,7 +33,7 @@ pub struct StreamInfo { #[derive(Debug, Clone, Serialize, Deserialize)] pub enum StreamStatus { /// 正在直播，包含流信息 - Live { stream_info: StreamInfo }, + Live { stream_info: StreamInfoExt }, /// 离线状态 Offline, /// 未知状态 @@ -56,7 +47,7 @@ pub trait DownloadBase: Send + Sync { /// 检查流是否可用 async fn check_stream(&self) -> Result>; /// 获取流信息 - async fn get_stream_info(&self) -> Result>; + async fn get_stream_info(&self) -> Result>; /// 下载到指定路径 async fn download(&self, output_path: impl AsRef) -> Result<(), Report>; /// 判断是否应该录制 @@ -76,7 +67,7 @@ pub trait DownloadPlugin { /// 创建下载器实例 async fn create_downloader( &self, - stream_info: &StreamInfo, + stream_info: &StreamInfoExt, config: Config, recorder: Recorder, ) -> Arc; diff --git a/crates/biliup-cli/src/server/infrastructure/context.rs b/crates/biliup-cli/src/server/infrastructure/context.rs index 9703a23..bd9f39a 100644 --- a/crates/biliup-cli/src/server/infrastructure/context.rs +++ b/crates/biliup-cli/src/server/infrastructure/context.rs @@ -1,9 +1,9 @@ use crate::server::config::Config; use crate::server::core::downloader::Downloader; -use crate::server::core::plugin::StreamInfo; +use crate::server::core::plugin::StreamInfoExt; use crate::server::errors::{AppError, AppResult}; use crate::server::infrastructure::connection_pool::ConnectionPool; -use crate::server::infrastructure::models::{LiveStreamer, UploadStreamer}; +use crate::server::infrastructure::models::live_streamer::LiveStreamer; use crate::server::infrastructure::repositories::{get_config, get_streamer}; use axum::http::Extensions; use biliup::client::StatelessClient; @@ -13,15 +13,17 @@ use ormlite::Model; use serde::{Deserialize, Serialize, Serializer}; use std::sync::{Arc, RwLock}; use tracing::info; +use crate::server::infrastructure::models::upload_streamer::UploadStreamer; /// 应用程序上下文，包含工作器和扩展信息 #[derive(Debug, Clone)] pub struct Context { /// 工作器实例 pub worker: Arc, - pub stream_info: StreamInfo, + pub stream_info: StreamInfoExt, /// 扩展数据容器 pub extension: Extensions, + pub pool: ConnectionPool, } impl Context { @@ -29,11 +31,12 @@ impl Context { /// /// # 参数 /// * `worker` - 工作器实例的Arc引用 - pub fn new(worker: Arc, stream_info: StreamInfo) -> Self { + pub fn new(worker: Arc, stream_info: StreamInfoExt, pool: ConnectionPool) -> Self { Self { worker, stream_info, extension: Default::default(), + pool, } } } diff --git a/crates/biliup-cli/src/server/infrastructure/dto.rs b/crates/biliup-cli/src/server/infrastructure/dto.rs index 85a3a88..79aae6e 100644 --- a/crates/biliup-cli/src/server/infrastructure/dto.rs +++ b/crates/biliup-cli/src/server/infrastructure/dto.rs @@ -1,5 +1,5 @@ use crate::server::infrastructure::context::WorkerStatus; -use crate::server::infrastructure::models::LiveStreamer; +use crate::server::infrastructure::models::live_streamer::LiveStreamer; use serde::{Deserialize, Serialize}; /// 直播主播响应数据传输对象 diff --git a/crates/biliup-cli/src/server/infrastructure/models.rs b/crates/biliup-cli/src/server/infrastructure/models.rs index fee5db8..a0cc628 100644 --- a/crates/biliup-cli/src/server/infrastructure/models.rs +++ b/crates/biliup-cli/src/server/infrastructure/models.rs @@ -1,69 +1,14 @@ /// 钩子步骤模块 pub mod hook_step; +pub mod upload_streamer; +pub mod live_streamer; -use chrono::NaiveDateTime; -use hook_step::HookStep; +use chrono::{DateTime, Utc}; use ormlite::{Insert, Model}; use serde::{Deserialize, Serialize}; -use serde_json::Value; - -/// 上传配置模型 -/// 存储视频上传到B站时的各种配置参数 -#[derive(Model, Debug, Clone, Serialize, Deserialize)] -#[ormlite(table = "uploadstreamers")] -pub struct UploadStreamer { - /// 主键ID - pub id: i64, - /// 模板名称 - pub template_name: String, - /// 视频标题 - pub title: Option, - /// 分区ID - pub tid: Option, - /// 版权类型（1-自制，2-转载） - pub copyright: Option, - /// 转载来源 - pub copyright_source: Option, - /// 封面路径 - pub cover_path: Option, - /// 视频简介 - pub description: Option, - /// 动态内容 - pub dynamic: Option, - /// 定时发布时间 - pub dtime: Option, - /// 杜比音效 - pub dolby: Option, - /// Hi-Res音质 - pub hires: Option, - /// 充电专属 - pub charging_pay: Option, - /// 禁止转载 - pub no_reprint: Option, - /// 上传者 - pub uploader: Option, - /// 用户Cookie - pub user_cookie: Option, - /// 标签列表（JSON格式） - #[ormlite(json)] - pub tags: Vec, // not null - /// 制作人员信息 - pub credits: Option, - /// 开启精选评论 - pub up_selection_reply: Option, - /// 关闭评论 - pub up_close_reply: Option, - /// 关闭弹幕 - pub up_close_danmu: Option, - /// 额外字段 - pub extra_fields: Option, - /// 仅自己可见 - pub is_only_self: Option, -} - /// 主播信息模型 /// 存储主播的基本信息和直播状态 -#[derive(Model, Debug, Clone, Serialize, Deserialize)] +#[derive(Model, Debug, Clone, Serialize, Deserialize, Default)] #[ormlite(table = "streamerinfo")] pub struct StreamerInfo { /// 主键ID @@ -74,61 +19,16 @@ pub struct StreamerInfo { pub url: String, /// 直播标题 pub title: String, - /// 直播时间 - pub date: NaiveDateTime, - /// 直播封面路径 + /// 直播开始时间 + pub date: DateTime, + /// 直播封面路径（可选） pub live_cover_path: String, } -/// 直播主播模型 -/// 存储直播主播的配置信息和录制参数 -#[derive(Model, Debug, Clone, Serialize, Deserialize)] -#[ormlite(table = "livestreamers")] -pub struct LiveStreamer { - /// 主键ID - #[ormlite(primary_key)] - pub id: i64, - /// 直播间URL - pub url: String, - /// 备注名称 - pub remark: String, - /// 文件名前缀 - pub filename_prefix: Option, - /// 录制时间范围 - pub time_range: Option, - /// 关联的上传配置ID（外键，可空） - pub upload_streamers_id: Option, - /// 录制格式 - pub format: Option, - /// 覆盖配置（"override"为关键字，字段名避让） - #[ormlite(column = "override")] - #[serde(rename = "override")] - pub override_cfg: Option, - - /// 预处理器列表（JSON格式） - /// 注意：数据库空与json空有区别，所以这里不能用#[ormlite(json)] - /// 只能使用 sqlx::types::Json - #[ormlite(json)] - pub preprocessor: Option>, - /// 分段处理器列表（JSON格式） - #[ormlite(json)] - pub segment_processor: Option>, - /// 下载完成处理器列表（JSON格式） - #[ormlite(json)] - pub downloaded_processor: Option>, - /// 后处理器列表（JSON格式） - #[ormlite(json)] - pub postprocessor: Option>, - /// 可选参数 - pub opt_args: Option, - /// 排除关键词 - pub excluded_keywords: Option, -} - /// 文件列表模型 /// 存储录制文件的信息 #[derive(Model, Debug, Clone, Serialize, Deserialize)] -#[ormlite(table = "filelist")] +#[ormlite(table = "filelist", insert = "InsertFileItem")] pub struct FileItem { /// 主键ID pub id: i64, @@ -151,68 +51,6 @@ pub struct Configuration { pub value: String, } -/// 插入直播主播的数据结构 -/// 用于创建新的直播主播记录 -#[derive(Insert, Debug, Serialize, Deserialize)] -#[ormlite(returns = "LiveStreamer")] -pub struct InsertLiveStreamer { - pub url: String, - pub remark: String, - pub filename_prefix: Option, - pub time_range: Option, - pub upload_streamers_id: Option, // FK，可空 - pub format: Option, - #[ormlite(column = "override")] - // “override” 是字段名，这里改为 override_cfg 避免与保留字混淆 - #[serde(rename = "override")] - pub override_cfg: Option, // "override" 为关键字，字段名避让 - - // #[ormlite(json)] 数据库空与json空有区别所以这里不能用 - // pub preprocessor: Option>, - // 只能使用 sqlx::types::Json - #[ormlite(json)] - pub preprocessor: Option>, - #[ormlite(json)] - pub segment_processor: Option>, - #[ormlite(json)] - pub downloaded_processor: Option>, - #[ormlite(json)] - pub postprocessor: Option>, - pub opt_args: Option, - pub excluded_keywords: Option, -} - -/// 插入上传配置的数据结构 -/// 用于创建新的上传配置记录 -#[derive(Model, Insert, Debug, Clone, Serialize, Deserialize)] -#[ormlite(table = "uploadstreamers", returns = "UploadStreamer")] -pub struct InsertUploadStreamer { - pub id: Option, - pub template_name: String, - pub title: Option, - pub tid: Option, - pub copyright: Option, - pub copyright_source: Option, - pub cover_path: Option, - pub description: Option, - pub dynamic: Option, - pub dtime: Option, - pub dolby: Option, - pub hires: Option, - pub charging_pay: Option, - pub no_reprint: Option, - pub uploader: Option, - pub user_cookie: Option, - #[ormlite(json)] - pub tags: Vec, // not null - pub credits: Option, - pub up_selection_reply: Option, - pub up_close_reply: Option, - pub up_close_danmu: Option, - pub extra_fields: Option, - pub is_only_self: Option, -} - /// 插入配置的数据结构 /// 用于创建新的配置记录 #[derive(Insert, Debug, Clone, Serialize, Deserialize)] diff --git a/crates/biliup-cli/src/server/infrastructure/models/live_streamer.rs b/crates/biliup-cli/src/server/infrastructure/models/live_streamer.rs new file mode 100644 index 0000000..81ab8c5 --- /dev/null +++ b/crates/biliup-cli/src/server/infrastructure/models/live_streamer.rs @@ -0,0 +1,80 @@ +use ormlite::{Insert, Model}; +use serde::{Deserialize, Serialize}; +use serde_json::Value; +use crate::server::infrastructure::models::hook_step::HookStep; + +/// 直播主播模型 +/// 存储直播主播的配置信息和录制参数 +#[derive(Model, Debug, Clone, Serialize, Deserialize)] +#[ormlite(table = "livestreamers")] +pub struct LiveStreamer { + /// 主键ID + #[ormlite(primary_key)] + pub id: i64, + /// 直播间URL + pub url: String, + /// 备注名称 + pub remark: String, + /// 文件名前缀 + pub filename_prefix: Option, + /// 录制时间范围 + pub time_range: Option, + /// 关联的上传配置ID（外键，可空） + pub upload_streamers_id: Option, + /// 录制格式 + pub format: Option, + /// 覆盖配置（"override"为关键字，字段名避让） + #[ormlite(column = "override")] + #[serde(rename = "override")] + pub override_cfg: Option, + + /// 预处理器列表（JSON格式） + /// 注意：数据库空与json空有区别，所以这里不能用#[ormlite(json)] + /// 只能使用 sqlx::types::Json + #[ormlite(json)] + pub preprocessor: Option>, + /// 分段处理器列表（JSON格式） + #[ormlite(json)] + pub segment_processor: Option>, + /// 下载完成处理器列表（JSON格式） + #[ormlite(json)] + pub downloaded_processor: Option>, + /// 后处理器列表（JSON格式） + #[ormlite(json)] + pub postprocessor: Option>, + /// 可选参数 + pub opt_args: Option, + /// 排除关键词 + pub excluded_keywords: Option, +} + +/// 插入直播主播的数据结构 +/// 用于创建新的直播主播记录 +#[derive(Insert, Debug, Serialize, Deserialize)] +#[ormlite(returns = "LiveStreamer")] +pub struct InsertLiveStreamer { + pub url: String, + pub remark: String, + pub filename_prefix: Option, + pub time_range: Option, + pub upload_streamers_id: Option, // FK，可空 + pub format: Option, + #[ormlite(column = "override")] + // “override” 是字段名，这里改为 override_cfg 避免与保留字混淆 + #[serde(rename = "override")] + pub override_cfg: Option, // "override" 为关键字，字段名避让 + + // #[ormlite(json)] 数据库空与json空有区别所以这里不能用 + // pub preprocessor: Option>, + // 只能使用 sqlx::types::Json + #[ormlite(json)] + pub preprocessor: Option>, + #[ormlite(json)] + pub segment_processor: Option>, + #[ormlite(json)] + pub downloaded_processor: Option>, + #[ormlite(json)] + pub postprocessor: Option>, + pub opt_args: Option, + pub excluded_keywords: Option, +} \ No newline at end of file diff --git a/crates/biliup-cli/src/server/infrastructure/models/upload_streamer.rs b/crates/biliup-cli/src/server/infrastructure/models/upload_streamer.rs new file mode 100644 index 0000000..0b8d1df --- /dev/null +++ b/crates/biliup-cli/src/server/infrastructure/models/upload_streamer.rs @@ -0,0 +1,88 @@ +use ormlite::{Insert, Model}; +use serde::{Deserialize, Serialize}; +use serde_json::Value; + +/// 上传配置模型 +/// 存储视频上传到B站时的各种配置参数 +#[derive(Model, Debug, Clone, Serialize, Deserialize)] +#[ormlite(table = "uploadstreamers")] +pub struct UploadStreamer { + /// 主键ID + pub id: i64, + /// 模板名称 + pub template_name: String, + /// 视频标题 + pub title: Option, + /// 分区ID + pub tid: Option, + /// 版权类型（1-自制，2-转载） + pub copyright: Option, + /// 转载来源 + pub copyright_source: Option, + /// 封面路径 + pub cover_path: Option, + /// 视频简介 + pub description: Option, + /// 动态内容 + pub dynamic: Option, + /// 定时发布时间 + pub dtime: Option, + /// 杜比音效 + pub dolby: Option, + /// Hi-Res音质 + pub hires: Option, + /// 充电专属 + pub charging_pay: Option, + /// 禁止转载 + pub no_reprint: Option, + /// 上传者 + pub uploader: Option, + /// 用户Cookie + pub user_cookie: Option, + /// 标签列表（JSON格式） + #[ormlite(json)] + pub tags: Vec, // not null + /// 制作人员信息 + pub credits: Option, + /// 开启精选评论 + pub up_selection_reply: Option, + /// 关闭评论 + pub up_close_reply: Option, + /// 关闭弹幕 + pub up_close_danmu: Option, + /// 额外字段 + pub extra_fields: Option, + /// 仅自己可见 + pub is_only_self: Option, +} + +/// 插入上传配置的数据结构 +/// 用于创建新的上传配置记录 +#[derive(Model, Insert, Debug, Clone, Serialize, Deserialize)] +#[ormlite(table = "uploadstreamers", returns = "UploadStreamer")] +pub struct InsertUploadStreamer { + pub id: Option, + pub template_name: String, + pub title: Option, + pub tid: Option, + pub copyright: Option, + pub copyright_source: Option, + pub cover_path: Option, + pub description: Option, + pub dynamic: Option, + pub dtime: Option, + pub dolby: Option, + pub hires: Option, + pub charging_pay: Option, + pub no_reprint: Option, + pub uploader: Option, + pub user_cookie: Option, + #[ormlite(json)] + pub tags: Vec, // not null + pub credits: Option, + pub up_selection_reply: Option, + pub up_close_reply: Option, + pub up_close_danmu: Option, + pub extra_fields: Option, + pub is_only_self: Option, +} \ No newline at end of file diff --git a/crates/biliup-cli/src/server/infrastructure/repositories.rs b/crates/biliup-cli/src/server/infrastructure/repositories.rs index f5d4394..92b6d8a 100644 --- a/crates/biliup-cli/src/server/infrastructure/repositories.rs +++ b/crates/biliup-cli/src/server/infrastructure/repositories.rs @@ -2,10 +2,12 @@ use crate::server::config::Config; use crate::server::errors::{AppError, AppResult}; use crate::server::infrastructure::connection_pool::ConnectionPool; use crate::server::infrastructure::models::{ - Configuration, InsertConfiguration, LiveStreamer, UploadStreamer, + Configuration, InsertConfiguration, }; use error_stack::ResultExt; use ormlite::{Insert, Model}; +use crate::server::infrastructure::models::live_streamer::LiveStreamer; +use crate::server::infrastructure::models::upload_streamer::UploadStreamer; /// 根据ID获取直播主播信息 /// diff --git a/crates/biliup-cli/src/server/infrastructure/service_register.rs b/crates/biliup-cli/src/server/infrastructure/service_register.rs index eacd47f..745c6f7 100644 --- a/crates/biliup-cli/src/server/infrastructure/service_register.rs +++ b/crates/biliup-cli/src/server/infrastructure/service_register.rs @@ -4,12 +4,13 @@ use crate::server::core::monitor::Monitor; use crate::server::errors::{AppError, AppResult}; use crate::server::infrastructure::connection_pool::ConnectionPool; use crate::server::infrastructure::context::{Context, Worker, WorkerStatus}; -use crate::server::infrastructure::models::{LiveStreamer, UploadStreamer}; +use crate::server::infrastructure::models::live_streamer::LiveStreamer; use axum::extract::FromRef; use biliup::client::StatelessClient; use error_stack::bail; use std::sync::{Arc, RwLock}; use tracing::info; +use crate::server::infrastructure::models::upload_streamer::UploadStreamer; /// 服务注册器 /// 负责管理应用程序中的各种服务实例，包括数据库连接池、工作器、下载管理器等 @@ -98,7 +99,7 @@ impl ServiceRegister { self.client.clone(), )); // 将工作器添加到监控器和工作器列表中 - let monitor = manager.ensure_monitor(); + 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"); @@ -130,7 +131,7 @@ impl ServiceRegister { bail!(AppError::Unknown) }; // 从监控器中删除房间 - let monitor = manager.ensure_monitor(); + let monitor = manager.ensure_monitor(self.pool.clone()); let len = monitor.rooms_handle.del(id).await; info!("id: {id} removed, remained len {len}"); // 如果没有剩余房间，清理监控器 diff --git a/crates/stream-gears/src/server.rs b/crates/stream-gears/src/server.rs index 78a8576..ea78fe0 100644 --- a/crates/stream-gears/src/server.rs +++ b/crates/stream-gears/src/server.rs @@ -8,14 +8,14 @@ use biliup_cli::server::core::download_manager::{ActorHandle, DownloadManager}; use biliup_cli::server::core::downloader::ffmpeg_downloader::FfmpegDownloader; use biliup_cli::server::core::downloader::stream_gears::StreamGears; use biliup_cli::server::core::downloader::{DownloadConfig, Downloader, DownloaderType}; -use biliup_cli::server::core::plugin::{DownloadPlugin, StreamInfo, StreamStatus}; +use biliup_cli::server::core::plugin::{DownloadPlugin, StreamInfoExt, StreamStatus}; use biliup_cli::server::errors::{AppError, AppResult}; use biliup_cli::server::infrastructure::connection_pool::ConnectionManager; use biliup_cli::server::infrastructure::context::{Context, Worker}; use biliup_cli::server::infrastructure::repositories; use biliup_cli::server::infrastructure::repositories::get_upload_config; use biliup_cli::server::infrastructure::service_register::ServiceRegister; -use chrono::Local; +use chrono::{DateTime, Local, NaiveDateTime, Utc}; use error_stack::{Report, ResultExt}; use fancy_regex::Regex; use pyo3::exceptions::PyRuntimeError; @@ -33,6 +33,7 @@ use std::path::PathBuf; use std::sync::{Arc, LazyLock, RwLock}; use time::OffsetDateTime; use tracing::{debug, info}; +use biliup_cli::server::infrastructure::models::StreamerInfo; #[derive(Debug)] pub struct PyPlugin { @@ -90,7 +91,7 @@ impl DownloadPlugin for PyPlugin { async fn create_downloader( &self, - stream_info: &StreamInfo, + stream_info: &StreamInfoExt, // worker: &Worker, config: Config, recorder: Recorder, @@ -142,10 +143,10 @@ impl DownloadPlugin for PyPlugin { async fn call_via_threads( obj: Arc>, url: &str, -) -> PyResult>)>> { +) -> PyResult>)>> { let url = url.to_string(); tokio::task::spawn_blocking(move || { - Python::attach(|py| -> PyResult>)>> { + Python::attach(|py| -> PyResult>)>> { // 从 biliup.util 获取 loop（按你项目里真实的名字来取） let util = PyModule::import(py, "biliup.common.util")?; // 下面两行二选一（取决于 biliup.util 的 API）： @@ -216,7 +217,7 @@ pub fn from_py(actor_handle: Arc) -> PyResult> /// end_time 的语义与 Python 中一致：若未提供或为“假值”，则使用 time.localtime() pub fn stream_info_from_py( self_obj: &Bound<'_, PyAny>, -) -> PyResult<(StreamInfo, Option>)> { +) -> PyResult<(StreamInfoExt, Option>)> { // 从 self 上获取属性并抽取为 Rust 类型 let name: String = self_obj.getattr("fname")?.extract()?; let url: String = self_obj.getattr("url")?.extract()?; @@ -239,7 +240,7 @@ pub fn stream_info_from_py( // date 直接使用传入的 start_time（保留为 Python 对象） // let date = OffsetDateTime::now_utc(); - let date = Local::now(); + // end_time: 若传入 None 或“假值”，则使用 time.localtime() // let end_time_obj: PyObject = match end_time { // Some(et) if et.is_true()? => et.to_object(py), @@ -249,16 +250,19 @@ pub fn stream_info_from_py( // lt.to_object(py) // } // };self.update_headers(self.stream_headers) - + Ok(( - StreamInfo { - name, - url, + StreamInfoExt { + streamer_info: StreamerInfo { + id: 0, + name, + url, + title, + date: Utc::now(), + live_cover_path: live_cover_path.unwrap_or_default(), + }, suffix: media_ext_from_url(&raw_stream_url).unwrap(), raw_stream_url, - title, - date, - live_cover_path, platform, stream_headers, },