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, },