>(&mut self, path: Option) -> Result<()> {
- let config_path = if let Some(path) = path {
- path.as_ref().to_path_buf()
- } else {
- // 自动检测配置文件
- self.find_config_file().await?
- };
-
- info!("Loading config from: {:?}", config_path);
-
- let content = fs::read_to_string(&config_path).await?;
- let format = self.detect_format(&config_path)?;
-
- self.config = match format {
- ConfigFormat::Yaml => serde_yaml::from_str(&content)?,
- ConfigFormat::Toml => toml::from_str(&content)?,
- ConfigFormat::Json => serde_json::from_str(&content)?,
- };
-
- self.config_path = Some(config_path);
- self.format = Some(format);
-
- info!("Config loaded successfully");
- Ok(())
- }
-
- /// 保存配置到文件，对应Python的save方法
- pub async fn save(&self) -> Result<()> {
- let config_path = self
- .config_path
- .as_ref()
- .ok_or_else(|| anyhow::anyhow!("No config path set"))?;
- let format = self
- .format
- .as_ref()
- .ok_or_else(|| anyhow::anyhow!("No config format set"))?;
-
- let content = match format {
- ConfigFormat::Yaml => serde_yaml::to_string(&self.config)?,
- ConfigFormat::Toml => toml::to_string_pretty(&self.config)?,
- ConfigFormat::Json => serde_json::to_string_pretty(&self.config)?,
- };
-
- fs::write(config_path, content).await?;
- info!("Config saved to: {:?}", config_path);
- Ok(())
- }
-
- /// 保存到指定路径，对应Python的dump方法
- pub async fn dump>(&self, path: P) -> Result<()> {
- let path = path.as_ref();
- let format = self.detect_format(path)?;
-
- // 创建备份
- if path.exists() {
- let backup_path = format!(
- "{}.backup.{}",
- path.to_string_lossy(),
- chrono::Utc::now().format("%Y%m%d%H%M%S")
- );
- if let Err(e) = fs::copy(path, &backup_path).await {
- warn!("Failed to create backup: {}", e);
- } else {
- info!("Created backup: {}", backup_path);
- }
- }
-
- let content = match format {
- ConfigFormat::Yaml => serde_yaml::to_string(&self.config)?,
- ConfigFormat::Toml => toml::to_string_pretty(&self.config)?,
- ConfigFormat::Json => serde_json::to_string_pretty(&self.config)?,
- };
-
- fs::write(path, content).await?;
- info!("Config dumped to: {:?}", path);
- Ok(())
- }
-
- /// 自动查找配置文件
- async fn find_config_file(&self) -> Result {
- let candidates = vec!["config.yaml", "config.yml", "config.toml", "config.json"];
-
- for candidate in candidates {
- let path = PathBuf::from(candidate);
- if path.exists() {
- return Ok(path);
- }
- }
-
- Err(anyhow::anyhow!("No config file found"))
- }
-
- /// 检测配置文件格式
- fn detect_format>(&self, path: P) -> Result {
- let path = path.as_ref();
- let extension = path
- .extension()
- .and_then(|s| s.to_str())
- .ok_or_else(|| anyhow::anyhow!("Cannot determine file format"))?;
-
- match extension.to_lowercase().as_str() {
- "yaml" | "yml" => Ok(ConfigFormat::Yaml),
- "toml" => Ok(ConfigFormat::Toml),
- "json" => Ok(ConfigFormat::Json),
- _ => Err(anyhow::anyhow!("Unsupported config format: {}", extension)),
- }
- }
-
- /// 获取配置的只读引用
- pub fn get_config(&self) -> &BiliUpConfig {
- &self.config
- }
-
- /// 获取配置的可变引用
- pub fn get_config_mut(&mut self) -> &mut BiliUpConfig {
- &mut self.config
- }
-
- /// 获取流配置
- pub fn get_streamer_config(&self, name: &str) -> Option<&StreamerConfig> {
- self.config.streamers.as_ref()?.get(name)
- }
-
- /// 设置流配置
- pub fn set_streamer_config(&mut self, name: String, config: StreamerConfig) {
- if self.config.streamers.is_none() {
- self.config.streamers = Some(HashMap::new());
- }
- if let Some(ref mut streamers) = self.config.streamers {
- streamers.insert(name, config);
- }
- }
-
- /// 删除流配置
- pub fn remove_streamer_config(&mut self, name: &str) -> Option {
- self.config.streamers.as_mut()?.remove(name)
- }
-
- /// 获取所有流名称
- pub fn get_streamer_names(&self) -> Vec {
- self.config
- .streamers
- .as_ref()
- .map(|streamers| streamers.keys().cloned().collect())
- .unwrap_or_default()
- }
-
- /// 获取配置值（通用方法）
- pub fn get(&self, key: &str) -> Option
- where
- T: for<'de> Deserialize<'de> + Clone,
- {
- // 这是一个简化版本，实际实现需要处理嵌套的键路径
- // 例如：get("streamers.streamer1.url")
- None
- }
-
- /// 从cookies文件加载，对应Python的load_cookies
- pub async fn load_cookies>(&mut self, path: P) -> Result<()> {
- let content = fs::read_to_string(path).await?;
- let cookies_data: serde_json::Value = serde_json::from_str(&content)?;
-
- if let Some(cookie_info) = cookies_data.get("cookie_info")
- && let Some(cookies) = cookie_info.get("cookies").and_then(|c| c.as_array())
- {
- let mut cookie_map = HashMap::new();
- for cookie in cookies {
- if let (Some(name), Some(value)) = (
- cookie.get("name").and_then(|n| n.as_str()),
- cookie.get("value").and_then(|v| v.as_str()),
- ) {
- cookie_map.insert(name.to_string(), value.to_string());
- }
- }
-
- if self.config.user.is_none() {
- self.config.user = Some(UserConfig {
- cookies: None,
- access_token: None,
- });
- }
- if let Some(ref mut user) = self.config.user {
- user.cookies = Some(cookie_map);
- }
- }
-
- if let Some(token_info) = cookies_data.get("token_info")
- && let Some(access_token) = token_info.get("access_token").and_then(|t| t.as_str())
- && let Some(ref mut user) = self.config.user
- {
- user.access_token = Some(access_token.to_string());
- }
-
- Ok(())
- }
-
- /// 合并配置，用于处理覆盖字段
- pub fn merge(&mut self, other: BiliUpConfig) {
- // TODO: 实现深度合并逻辑
- if other.streamers.is_some() {
- self.config.streamers = other.streamers;
- }
- if other.user.is_some() {
- self.config.user = other.user;
- }
- // 合并其他字段...
- }
-}
-
-impl Default for ConfigManager {
- fn default() -> Self {
- Self::new()
- }
-}
-
-// 全局配置管理器实例
-
-#[cfg(test)]
-mod tests {
- use super::*;
- use tempfile::NamedTempFile;
- use tokio::io::AsyncWriteExt;
-
- #[tokio::test]
- async fn test_config_yaml_loading() {
- let yaml_content = r#"
-streamers:
- test_streamer:
- url: "https://live.bilibili.com/12345"
- format: "mp4"
- tags: ["测试", "直播"]
-threads: 4
-pool1_size: 8
-"#;
-
- let mut temp_file = NamedTempFile::new().unwrap();
-
- let mut manager = ConfigManager::new();
- manager.load(Some(temp_file.path())).await.unwrap();
-
- let config = manager.get_config();
- assert_eq!(config.threads, Some(4));
- assert_eq!(config.pool1_size, Some(8));
-
- let streamer = config
- .streamers
- .as_ref()
- .unwrap()
- .get("test_streamer")
- .unwrap();
- assert_eq!(streamer.url, "https://live.bilibili.com/12345");
- assert_eq!(streamer.format, Some("mp4".to_string()));
- }
-
- #[tokio::test]
- async fn test_config_toml_loading() {
- let toml_content = r#"
-threads = 4
-pool1_size = 8
-
-[streamers.test_streamer]
-url = "https://live.bilibili.com/12345"
-format = "mp4"
-tags = ["测试", "直播"]
-"#;
-
- let mut temp_file = NamedTempFile::new().unwrap();
-
- let mut manager = ConfigManager::new();
- manager.load(Some(temp_file.path())).await.unwrap();
-
- let config = manager.get_config();
- assert_eq!(config.threads, Some(4));
- assert_eq!(config.pool1_size, Some(8));
- }
-}
diff --git a/crates/biliup-cli/src/server/core.rs b/crates/biliup-cli/src/server/core.rs
deleted file mode 100644
index 1252701..0000000
--- a/crates/biliup-cli/src/server/core.rs
+++ /dev/null
@@ -1,32 +0,0 @@
-use serde::{Deserialize, Serialize};
-
-pub mod download_actor;
-pub mod live_streamers;
-pub mod main_loop;
-pub mod upload_actor;
-pub mod upload_streamers;
-pub mod users;
-pub mod util;
-
-/// Status of the live stream
-pub enum LiveStatus {
- /// Stream is online.
- Online,
- /// Stream is offline.
- Offline,
- /// The status of the stream could not be determined.
- Unknown,
-}
-
-/// Status of the live stream
-#[derive(Clone, Copy, Debug, Serialize, Deserialize, Default, PartialEq)]
-pub enum StreamStatus {
- /// Stream is online.
- Working,
- /// Stream is offline.
- Inspecting,
- /// The status of the stream could not be determined.
- #[default]
- Pending,
- Idle,
-}
diff --git a/crates/biliup-cli/src/server/core/download_actor.rs b/crates/biliup-cli/src/server/core/download_actor.rs
deleted file mode 100644
index 773e828..0000000
--- a/crates/biliup-cli/src/server/core/download_actor.rs
+++ /dev/null
@@ -1,210 +0,0 @@
-use biliup::client::StatelessClient;
-
-use crate::server::core::StreamStatus;
-use crate::server::core::live_streamers::LiveStreamerDto;
-use crate::server::core::util::{AnyMap, Cycle};
-use biliup::downloader::extractor::{SiteDefinition, find_extractor};
-
-use indexmap::indexmap;
-
-use std::collections::HashMap;
-use std::error::Error;
-use std::ops::DerefMut;
-use std::sync::{Arc, RwLock};
-use std::time::Duration;
-use tokio::task::JoinHandle;
-use tracing::debug;
-
-async fn start_monitor(
- task: Cycle,
- extractor: &(dyn SiteDefinition + Send + Sync),
- client: StatelessClient,
- // live_streamers_service: DynLiveStreamersService,
-) {
- let n = &mut 0;
- loop {
- let (url, status) = task.get(n);
- status != StreamStatus::Working;
-
- match status {
- StreamStatus::Working => {}
- StreamStatus::Inspecting => {}
- StreamStatus::Pending => {}
- StreamStatus::Idle => {
- task.change(&url, StreamStatus::Inspecting);
- }
- }
-
- match (extractor.get_site(&url, client.clone()).await, status) {
- (Ok(site), StreamStatus::Idle | StreamStatus::Inspecting) => {
- println!("Idle\n {url} \n{site}");
- // let (filename, split_size, split_time) = if let Ok(LiveStreamerDto {
- // filename,
- // split_size,
- // split_time,
- // ..
- // }) =
- // live_streamers_service.get_streamer_by_url(&url).await
- // {
- // (filename, split_size, split_time.map(Duration::from_secs))
- // } else {
- // ("./video/%Y-%m-%d/%H_%M_%S{title}".to_string(), None, None)
- // };
- // let live_streamers_service = live_streamers_service.clone();
- // {
- // let client = client.clone();
- // let url = url.clone();
- // let task = task.clone();
- // logging_spawn(async move {
- // let hook = live_streamers_service
- // .get_studio_by_url(&url)
- // .await
- // .unwrap_or_default()
- // .map(|studio| -> Box {
- // let handle = UploadActorHandle::new(client, studio);
- // Box::new(move |file_name| {
- // if let Ok(metadata) = std::fs::metadata(file_name)
- // .map_err(|err| error!("{}", err))
- // {
- // if metadata.len() > 10 * 1024 * 1024 {
- // info!("开始上传: {}", file_name);
- // handle.send_file_path(file_name);
- // }
- // }
- // })
- // });
- // if hook.is_none() {
- // debug!(url = %url, "upload template not set.");
- // }
- //
- // let segmentable = Segmentable::new(split_time, split_size);
- // // let segmentable = Segmentable::new( None, Some(16*1024*1024));
- // site.download(&filename, segmentable, hook).await?;
- // task.change(&url, StreamStatus::Idle);
- // Ok::<_, Box>(())
- // });
- // }
- task.change(&url, StreamStatus::Working);
- }
- (Ok(_site), StreamStatus::Pending) => {
- println!("Pending");
- }
- (Ok(_site), StreamStatus::Working) => {
- debug!("Working");
- }
- (Err(e), _) => {
- task.change(&url, StreamStatus::Idle);
- debug!(url, "{e}")
- }
- }
- tokio::time::sleep(Duration::from_secs(30)).await;
- }
-}
-
-#[derive(Clone)]
-struct DownloadActor {
- client: StatelessClient,
-}
-
-impl DownloadActor {
- fn new(client: StatelessClient) -> Self {
- Self { client }
- }
-
- fn run(
- &mut self,
- list: Vec,
- extensions: StreamActorMap,
- // client: StatelessClient,
- ) {
- for streamer in list {
- // let Some(extractor) = find_extractor(&streamer.url) else { continue; };
- let mut guard = extensions.write().unwrap();
- self.add_streamer(guard.deref_mut(), streamer.url)
- }
- println!("{:?}", extensions);
- }
-
- fn add_streamer(
- &self,
- map: &mut AnyMap<(Cycle, JoinHandle<()>)>,
- url: String,
- // client: StatelessClient,
- ) {
- let Some(extractor) = find_extractor(&url) else {
- return;
- };
- let _entry = map
- .entry(extractor.as_any().type_id())
- .and_modify(|(cy, _)| cy.insert(url.clone(), StreamStatus::Idle))
- .or_insert_with(|| {
- let cycle = Cycle::new(indexmap![url => StreamStatus::Idle]);
- let task = cycle.clone();
- let client = self.client.clone();
- let handle =
- tokio::spawn(async move { start_monitor(task, extractor, client).await });
- (cycle, handle)
- });
- }
-}
-
-type StreamActorMap = Arc, JoinHandle<()>)>>>;
-
-#[derive(Clone)]
-pub struct DownloadActorHandle {
- platform_map: StreamActorMap,
- // client: StatelessClient,
- actor: DownloadActor,
-}
-
-impl DownloadActorHandle {
- pub fn new(list: Vec, client: StatelessClient) -> Self {
- let mut actor = DownloadActor::new(client);
- let platform_map = Arc::new(RwLock::new(HashMap::default()));
- let platform = Arc::clone(&platform_map);
- // let client_c = client.clone();
- actor.run(list, platform);
- Self {
- platform_map,
- actor,
- }
- }
-
- pub fn add_streamer(&self, url: &str) {
- self.actor.add_streamer(
- self.platform_map.write().unwrap().deref_mut(),
- url.to_string(),
- // self.client.clone(),
- );
- }
-
- pub fn get_streamers(&self) -> HashMap {
- let read_guard = self.platform_map.read().unwrap();
- let mut map = HashMap::new();
- for (key, val) in read_guard.iter() {
- map.extend(val.0.get_all());
- }
- map
- }
-
- pub fn update_streamer(&self, url: &str) {
- self.remove_streamer(url);
- self.add_streamer(url);
- }
-
- pub fn remove_streamer(&self, url: &str) {
- find_extractor(url).and_then(|extractor| {
- self.platform_map
- .read()
- .unwrap()
- .get(&extractor.as_any().type_id())
- .and_then(|(cy, join_handle)| {
- let mut guard = cy.write();
- if guard.len() <= 1 {
- join_handle.abort()
- }
- guard.swap_remove(url)
- })
- });
- }
-}
diff --git a/crates/stream-gears/src/server/core/download_manager.rs b/crates/biliup-cli/src/server/core/download_manager.rs
similarity index 96%
rename from crates/stream-gears/src/server/core/download_manager.rs
rename to crates/biliup-cli/src/server/core/download_manager.rs
index 5830e88..c8ae1bc 100644
--- a/crates/stream-gears/src/server/core/download_manager.rs
+++ b/crates/biliup-cli/src/server/core/download_manager.rs
@@ -3,6 +3,7 @@ use crate::server::core::monitor::{Monitor, RoomsHandle};
use crate::server::core::plugin::{DownloadPlugin, StreamInfo};
use crate::server::errors::{AppError, AppResult};
use crate::server::infrastructure::context::{Worker, WorkerStatus};
+use crate::server::infrastructure::models::hook_step::process_video;
use async_channel::{Receiver, Sender, bounded};
use biliup::bilibili::{BiliBili, Studio, Video};
use biliup::client::StatelessClient;
@@ -19,7 +20,6 @@ use std::sync::{Arc, Mutex};
use std::time::Instant;
use tokio::task::JoinHandle;
use tracing::{error, info, warn};
-use crate::server::infrastructure::models::hook_step::process_video;
pub struct DownloadManager {
pub monitor: Mutex>>,
@@ -39,7 +39,7 @@ impl DownloadManager {
}
}
- pub(crate) fn ensure_monitor(&self) -> Arc {
+ pub fn ensure_monitor(&self) -> Arc {
// Monitor::new(self.plugin.name(), url)
self.monitor
.lock()
@@ -91,10 +91,10 @@ impl DActor {
move |event: SegmentEvent| {
if event.segment_index == 0 {
match sender.force_send(UploaderMessage::SegmentEvent(
- event,
- rx.clone(),
- room.clone(),
- )) {
+ event,
+ rx.clone(),
+ room.clone(),
+ )) {
Ok(Some(ret)) => {
warn!(SegmentEvent = ?ret, "replace an existing message in the channel");
}
@@ -175,7 +175,8 @@ impl UActor {
&line,
&event.file_path,
)
- .await.map_err(|e| error!(e=?e))
+ .await
+ .map_err(|e| error!(e=?e))
{
videos.push(video);
}
@@ -189,7 +190,8 @@ impl UActor {
&line,
&se.file_path,
)
- .await.map_err(|e| error!(e=?e))
+ .await
+ .map_err(|e| error!(e=?e))
{
videos.push(video);
}
@@ -254,7 +256,9 @@ impl UActor {
let streamer = worker.get_streamer().await.unwrap();
if let Some(post_processor) = streamer.postprocessor {
for video_path in &video_paths {
- process_video(video_path, &post_processor).await.map_err(|e| error!(e=?e));
+ process_video(video_path, &post_processor)
+ .await
+ .map_err(|e| error!(e=?e));
}
}
}
diff --git a/crates/stream-gears/src/server/core/downloader.rs b/crates/biliup-cli/src/server/core/downloader.rs
similarity index 100%
rename from crates/stream-gears/src/server/core/downloader.rs
rename to crates/biliup-cli/src/server/core/downloader.rs
diff --git a/crates/stream-gears/src/server/core/downloader/ffmpeg_downloader.rs b/crates/biliup-cli/src/server/core/downloader/ffmpeg_downloader.rs
similarity index 100%
rename from crates/stream-gears/src/server/core/downloader/ffmpeg_downloader.rs
rename to crates/biliup-cli/src/server/core/downloader/ffmpeg_downloader.rs
diff --git a/crates/stream-gears/src/server/core/downloader/stream_gears.rs b/crates/biliup-cli/src/server/core/downloader/stream_gears.rs
similarity index 100%
rename from crates/stream-gears/src/server/core/downloader/stream_gears.rs
rename to crates/biliup-cli/src/server/core/downloader/stream_gears.rs
diff --git a/crates/biliup-cli/src/server/core/live_streamers.rs b/crates/biliup-cli/src/server/core/live_streamers.rs
deleted file mode 100644
index aadd880..0000000
--- a/crates/biliup-cli/src/server/core/live_streamers.rs
+++ /dev/null
@@ -1,124 +0,0 @@
-use crate::server::core::StreamStatus;
-use async_trait::async_trait;
-use biliup::uploader::bilibili::Studio;
-use chrono::Utc;
-use serde::{Deserialize, Serialize};
-use sqlx::FromRow;
-use std::sync::Arc;
-
-/// Similar to above, we want to keep a reference count across threads so we can manage our connection pool.
-pub type DynLiveStreamersRepository = Arc;
-pub type DynDownloadRecordsRepository = Arc;
-pub type DynVideosRepository = Arc;
-
-#[async_trait]
-pub trait LiveStreamersRepository {
- async fn create_streamer(
- &self,
- entity: AddLiveStreamerDto,
- ) -> anyhow::Result;
- async fn delete_streamer(&self, id: i64) -> anyhow::Result<()>;
- async fn update_streamer(
- &self,
- entity: LiveStreamerEntity,
- ) -> anyhow::Result;
- async fn get_streamers(&self) -> anyhow::Result>;
- async fn get_streamer_by_url(&self, url: &str) -> anyhow::Result;
- async fn get_streamer_by_id(&self, id: i64) -> anyhow::Result;
-}
-
-#[async_trait]
-pub trait DownloadRecordsRepository {
- async fn create(&self, entity: DownloadRecords) -> anyhow::Result;
- // async fn delete(&self, id: i64) -> anyhow::Result<()>;
- // async fn update(&self, entity: DownloadRecords) -> anyhow::Result;
- async fn get_all(&self) -> anyhow::Result>;
- // async fn get_by_id(&self, id: i64) -> anyhow::Result;
-}
-
-#[async_trait]
-pub trait VideosRepository {
- async fn create(&self, entity: Videos) -> anyhow::Result;
- // async fn delete(&self, id: i64) -> anyhow::Result<()>;
- async fn update(&self, entity: Videos) -> anyhow::Result;
- // async fn get_all(&self) -> anyhow::Result>;
- async fn get_by_id(&self, id: i64) -> anyhow::Result;
-}
-
-#[derive(Serialize, Deserialize, FromRow)]
-pub struct LiveStreamerEntity {
- pub id: i64,
- pub url: String,
- pub remark: String,
- pub filename: String,
- pub split_time: Option,
- pub split_size: Option,
- pub upload_id: Option,
-}
-
-#[derive(FromRow)]
-pub struct DownloadRecords {
- pub id: i64,
- pub title: String,
- pub live_id: i64,
- pub upload_id: Option,
- pub status: String,
-}
-
-#[derive(FromRow)]
-pub struct Videos {
- pub id: i64,
- pub start_time: chrono::DateTime,
- pub end_time: chrono::DateTime,
- pub path: String,
- pub record_id: i64,
- pub status: String,
-}
-
-impl LiveStreamerEntity {
- pub fn into_dto(self) -> LiveStreamerDto {
- LiveStreamerDto {
- id: self.id,
- url: self.url,
- remark: self.remark,
- filename: self.filename,
- split_time: self.split_time.map(|t| t as u64),
- split_size: self.split_size.map(|s| s as u64),
- status: Default::default(),
- }
- }
-}
-
-/// A reference counter for our user service allows us safely pass instances user utils
-/// around which themselves depend on the user repostiory, and ultimately, our Posgres connection pool.
-pub type DynLiveStreamersService = Arc;
-
-#[async_trait]
-pub trait LiveStreamersService {
- async fn add_streamer(&self, request: AddLiveStreamerDto) -> anyhow::Result;
- async fn get_streamer_by_url(&self, url: &str) -> anyhow::Result;
- async fn get_streamer_by_id(&self, id: i64) -> anyhow::Result;
- async fn get_streamers(&self) -> anyhow::Result>;
- async fn get_studio_by_url(&self, url: &str) -> anyhow::Result>;
-}
-
-#[derive(Serialize, Deserialize, Default, Debug)]
-pub struct AddLiveStreamerDto {
- pub url: String,
- pub remark: String,
- pub filename: String,
- pub split_time: Option,
- pub split_size: Option,
- pub upload_id: Option,
-}
-
-#[derive(Serialize, Deserialize, Default, Debug)]
-pub struct LiveStreamerDto {
- pub id: i64,
- pub url: String,
- pub remark: String,
- pub filename: String,
- pub split_time: Option,
- pub split_size: Option,
- pub status: StreamStatus,
-}
diff --git a/crates/biliup-cli/src/server/core/main_loop.rs b/crates/biliup-cli/src/server/core/main_loop.rs
deleted file mode 100644
index 6f656a9..0000000
--- a/crates/biliup-cli/src/server/core/main_loop.rs
+++ /dev/null
@@ -1,115 +0,0 @@
-use crate::server::core::StreamStatus;
-use crate::server::core::live_streamers::Videos;
-use crate::server::core::util::Cycle;
-use anyhow::Result;
-use biliup::downloader::extractor::Site;
-use tokio::sync::mpsc::Sender;
-use tokio::sync::mpsc::{Receiver, channel};
-use tokio::task::JoinHandle;
-
-/// This struct is used by client actors to send messages to the main loop. The
-/// message type is `ToServer`.
-#[derive(Clone, Debug)]
-pub struct ServerHandle {
- chan: Sender,
-}
-impl ServerHandle {
- pub async fn send(&mut self, msg: ToMain) {
- if self.chan.send(msg).await.is_err() {
- panic!("Main loop has shut down.");
- }
- }
-}
-
-/// The message type used when a client actor sends messages to the main loop.
-pub enum ToMain {
- NewRecording(Site, Cycle),
- FileClosed(Videos),
- // FatalError(io::Error),
-}
-
-pub fn spawn_main_loop() -> (ServerHandle, JoinHandle<()>) {
- let (send, recv) = channel(64);
-
- let handle = ServerHandle { chan: send };
-
- let join = tokio::spawn(async move {
- let res = main_loop(recv).await;
- match res {
- Ok(()) => {}
- Err(err) => {
- eprintln!("Oops {}.", err);
- }
- }
- });
-
- (handle, join)
-}
-
-async fn main_loop(mut recv: Receiver) -> Result<()> {
- while let Some(msg) = recv.recv().await {
- match msg {
- ToMain::NewRecording(_site, _task) => {}
- ToMain::FileClosed(_) => {}
- }
- }
-
- Ok(())
-}
-
-// async fn recording(
-// url: &str,
-// mut site: Site,
-// task: Cycle,
-// client: StatelessClient,
-// live_streamers_service: DynLiveStreamersService,
-// ) {
-// println!("Idle\n {url} \n{site}");
-// let (filename, split_size, split_time) = if let Ok(LiveStreamerDto {
-// filename,
-// split_size,
-// split_time,
-// ..
-// }) =
-// live_streamers_service.get_streamer_by_url(url).await
-// {
-// (filename, split_size, split_time.map(Duration::from_secs))
-// } else {
-// ("./video/%Y-%m-%d/%H_%M_%S{title}".to_string(), None, None)
-// };
-//
-// logging_spawn({
-// // let client = client.clone();
-// let url = url.to_string();
-// let task = task.clone();
-// // let live_streamers_service = live_streamers_service.clone();
-// async move {
-// let hook = live_streamers_service
-// .get_studio_by_url(&url)
-// .await
-// .unwrap_or_default()
-// .map(|studio| -> Box {
-// let handle = UploadActorHandle::new(client, studio);
-// Box::new(move |file_name| {
-// if let Ok(metadata) =
-// std::fs::metadata(file_name).map_err(|err| error!("{}", err))
-// {
-// if metadata.len() > 10 * 1024 * 1024 {
-// info!("开始上传: {}", file_name);
-// handle.send_file_path(file_name);
-// }
-// }
-// })
-// });
-// if hook.is_none() {
-// debug!(url = %url, "upload template not set.");
-// }
-// let segmentable = Segmentable::new(split_time, split_size);
-// // let segmentable = Segmentable::new( None, Some(16*1024*1024));
-// site.download(&filename, segmentable, hook).await?;
-// task.change(&url, StreamStatus::Idle);
-// Ok::<_, Box>(())
-// }
-// });
-// task.change(url, StreamStatus::Working);
-// }
diff --git a/crates/stream-gears/src/server/core/mod.rs b/crates/biliup-cli/src/server/core/mod.rs
similarity index 100%
rename from crates/stream-gears/src/server/core/mod.rs
rename to crates/biliup-cli/src/server/core/mod.rs
diff --git a/crates/stream-gears/src/server/core/monitor.rs b/crates/biliup-cli/src/server/core/monitor.rs
similarity index 100%
rename from crates/stream-gears/src/server/core/monitor.rs
rename to crates/biliup-cli/src/server/core/monitor.rs
diff --git a/crates/biliup-cli/src/server/core/plugin.rs b/crates/biliup-cli/src/server/core/plugin.rs
new file mode 100644
index 0000000..3436f9c
--- /dev/null
+++ b/crates/biliup-cli/src/server/core/plugin.rs
@@ -0,0 +1,61 @@
+use crate::server::core::download_manager::{ActorHandle, DownloadManager};
+use crate::server::core::downloader::ffmpeg_downloader::FfmpegDownloader;
+use crate::server::core::downloader::stream_gears::StreamGears;
+use crate::server::core::downloader::{DownloadConfig, Downloader, DownloaderType};
+use crate::server::errors::{AppError, AppResult};
+use crate::server::infrastructure::context::Worker;
+use crate::server::util::{Recorder, media_ext_from_url, parse_time};
+use async_trait::async_trait;
+use biliup::downloader::util::Segmentable;
+use error_stack::{Report, ResultExt};
+use serde::{Deserialize, Serialize};
+use std::collections::HashMap;
+use std::path::{Path, PathBuf};
+use std::sync::Arc;
+use time::OffsetDateTime;
+use tracing::{debug, info};
+
+// Stream information structures
+#[derive(Debug, Clone, Serialize, Deserialize)]
+pub struct StreamInfo {
+ pub name: String,
+ pub url: String,
+ pub raw_stream_url: String,
+ pub title: String,
+ #[serde(with = "time::serde::rfc3339")]
+ pub date: OffsetDateTime, // 保存 Python 的时间对象（如 time.struct_time）
+ // pub end_time: PyObject, // 同上
+ pub live_cover_path: Option,
+ pub platform: String,
+ pub stream_headers: HashMap,
+ pub suffix: String,
+}
+
+#[derive(Debug, Clone, Serialize, Deserialize)]
+pub enum StreamStatus {
+ Live { stream_info: StreamInfo },
+ Offline,
+ Unknown,
+}
+
+#[async_trait]
+pub trait DownloadBase: Send + Sync {
+ async fn check_stream(&self) -> Result>;
+ async fn get_stream_info(&self) -> Result>;
+ async fn download(&self, output_path: impl AsRef) -> Result<(), Report>;
+ fn should_record(&self, room_title: &str) -> bool;
+ fn get_platform_name(&self) -> &'static str;
+}
+
+#[async_trait]
+pub trait DownloadPlugin {
+ fn matches(&self, url: &str) -> bool;
+ async fn check_status(&self, url: &str) -> Result>;
+ async fn create_downloader(
+ &self,
+ stream_info: &StreamInfo,
+ worker: &Worker,
+ ) -> AppResult>;
+
+ fn name(&self) -> &str;
+}
diff --git a/crates/biliup-cli/src/server/core/upload_actor.rs b/crates/biliup-cli/src/server/core/upload_actor.rs
deleted file mode 100644
index bddd352..0000000
--- a/crates/biliup-cli/src/server/core/upload_actor.rs
+++ /dev/null
@@ -1,156 +0,0 @@
-use biliup::client::StatelessClient;
-use biliup::error::Kind;
-use biliup::uploader::VideoFile;
-use biliup::uploader::bilibili::{BiliBili, Studio, Vid, Video};
-use biliup::uploader::credential::login_by_cookies;
-use biliup::uploader::line::Line;
-use futures::StreamExt;
-use std::path::{Path, PathBuf};
-
-use std::time::Instant;
-use tokio::sync::mpsc;
-use tracing::{error, info};
-
-struct UploadActor {
- receiver: mpsc::UnboundedReceiver,
- client: StatelessClient,
- studio: Studio,
- vid: Option,
-}
-enum ActorMessage {
- Upload { path: PathBuf },
-}
-
-impl UploadActor {
- fn new(
- studio: Studio,
- client: StatelessClient,
- receiver: mpsc::UnboundedReceiver,
- ) -> Self {
- UploadActor {
- receiver,
- client,
- studio,
- vid: None,
- }
- }
-
- async fn upload(
- &self,
- video_paths: &[&Path],
- bili: &BiliBili,
- line: Line,
- limit: usize,
- ) -> biliup::error::Result> {
- let mut videos = Vec::new();
- for video_path in video_paths {
- println!("{:?}", video_path.canonicalize()?.to_str());
- info!("{line:?}");
- let video_file = VideoFile::new(video_path)?;
- let total_size = video_file.total_size;
- let file_name = video_file.file_name.clone();
- let uploader = line.pre_upload(bili, video_file).await?;
-
- let instant = Instant::now();
-
- let video = uploader
- .upload(self.client.clone(), limit, |vs| {
- vs.map(|vs| {
- let chunk = vs?;
- let len = chunk.len();
- Ok((chunk, len))
- })
- })
- .await?;
- let t = instant.elapsed().as_millis();
- info!(
- "Upload completed: {file_name} => cost {:.2}s, {:.2} MB/s.",
- t as f64 / 1000.,
- total_size as f64 / 1000. / t as f64
- );
- videos.push(video);
- }
- Ok(videos)
- }
-
- async fn handle_message(&mut self, msg: ActorMessage) -> biliup::error::Result<()> {
- match msg {
- ActorMessage::Upload { path } => {
- let bili = login_by_cookies("cookies.json", None).await?;
- let videos = self
- .upload(&[path.as_path()], &bili, Default::default(), 3)
- .await?;
-
- if let Some(vid) = &self.vid {
- let mut studio = bili.studio_data(vid, None).await?;
- studio.videos.extend(videos);
- bili.edit(&studio, None).await?;
- } else {
- let studio = &mut self.studio;
- studio.videos.extend(videos);
- if studio.title.is_empty() {
- studio.title = path
- .file_stem()
- .and_then(|fname| fname.to_str())
- .unwrap_or_default()
- .to_string();
- }
- if studio.tag.is_empty() {
- studio.tag = bili
- .recommend_tag(studio.tid, &studio.title, "")
- .await?
- .get(0)
- .and_then(|tag| tag.get("tag"))
- .and_then(|tag| tag.as_str())
- .unwrap_or("录播")
- .to_string();
- }
- let result = bili.submit(studio, None).await?;
- self.vid = Some(
- result
- .data
- .as_ref()
- .and_then(|data| data.get("bvid"))
- .and_then(|vid| vid.as_str())
- .map(|vid| Vid::Bvid(String::from(vid)))
- .ok_or_else(|| Kind::Custom(format!("{:?}", result)))?,
- );
- }
- }
- }
- Ok(())
- }
-}
-
-async fn run_download_actor(mut actor: UploadActor) {
- while let Some(msg) = actor.receiver.recv().await {
- match actor.handle_message(msg).await {
- Ok(_) => {}
- Err(e) => {
- error!("{}", e)
- }
- };
- }
-}
-
-#[derive(Clone)]
-pub struct UploadActorHandle {
- sender: mpsc::UnboundedSender,
-}
-
-impl UploadActorHandle {
- pub fn new(client: StatelessClient, studio: Studio) -> Self {
- let (sender, receiver) = mpsc::unbounded_channel();
- let actor = UploadActor::new(studio, client, receiver);
- tokio::spawn(run_download_actor(actor));
-
- Self { sender }
- }
-
- pub fn send_file_path>(&self, path: T) {
- let msg = ActorMessage::Upload {
- path: PathBuf::from(path.as_ref()),
- };
- let _ = self.sender.send(msg);
- }
-}
diff --git a/crates/biliup-cli/src/server/core/upload_streamers.rs b/crates/biliup-cli/src/server/core/upload_streamers.rs
deleted file mode 100644
index 2a784d5..0000000
--- a/crates/biliup-cli/src/server/core/upload_streamers.rs
+++ /dev/null
@@ -1,129 +0,0 @@
-use async_trait::async_trait;
-use biliup::uploader::bilibili::Studio;
-use serde::{Deserialize, Serialize};
-use sqlx::FromRow;
-use std::sync::Arc;
-
-pub type DynUploadStreamersRepository = Arc;
-pub type DynUploadRecordsRepository = Arc;
-
-#[async_trait]
-pub trait UploadStreamersRepository {
- async fn create_streamer(&self, studio: StudioEntity) -> anyhow::Result;
- async fn delete_streamer(&self, id: i64) -> anyhow::Result<()>;
- async fn update_streamer(&self, studio: StudioEntity) -> anyhow::Result;
- async fn get_streamers(&self) -> anyhow::Result>;
- async fn get_streamer_by_id(&self, id: i64) -> anyhow::Result;
-}
-
-#[async_trait]
-pub trait UploadRecordsRepository {
- async fn create(&self, entity: UploadRecords) -> anyhow::Result;
- async fn delete(&self, id: i64) -> anyhow::Result<()>;
- // async fn update(&self, entity: UploadRecords) -> anyhow::Result;
- async fn get_all(&self) -> anyhow::Result>;
- async fn get_by_id(&self, id: i64) -> anyhow::Result;
-}
-
-// #[serde(default)]
-#[derive(FromRow, Serialize, Deserialize)]
-pub struct StudioEntity {
- #[serde(default)]
- pub id: i64,
- pub template_name: String,
- pub user: i64,
- pub copyright: u8,
- pub source: String,
- pub tid: u16,
- pub cover: String,
- pub title: String,
- pub desc: String,
- pub dynamic: String,
- pub tag: String,
- pub dtime: Option,
- pub interactive: u8,
- pub mission_id: Option,
- pub dolby: u8,
- pub lossless_music: u8,
- pub no_reprint: u8,
- pub charging_pay: u8,
- pub up_selection_reply: bool,
- pub up_close_reply: bool,
- pub up_close_danmu: bool,
-}
-
-impl StudioEntity {
- pub fn into_dto(self) -> Studio {
- Studio {
- copyright: self.copyright,
- source: self.source,
- tid: self.tid,
- cover: self.cover,
- title: self.title,
- desc_format_id: 0,
- desc: self.desc,
- dynamic: self.dynamic,
- subtitle: Default::default(),
- tag: self.tag,
- videos: vec![],
- desc_v2: None,
- dtime: self.dtime,
- open_subtitle: false,
- interactive: self.interactive,
- mission_id: self.mission_id,
- dolby: self.dolby,
- lossless_music: self.lossless_music,
- no_reprint: self.no_reprint,
- charging_pay: self.charging_pay,
- aid: None,
- up_selection_reply: self.up_selection_reply,
- up_close_reply: self.up_close_reply,
- up_close_danmu: self.up_close_danmu,
- extra_fields: None,
- }
- }
-
- // pub fn into_profile(self, following: bool) -> ProfileDto {
- // ProfileDto {
- // username: self.username,
- // bio: self.bio,
- // image: self.image,
- // following,
- // }
- // }
-}
-
-#[derive(FromRow, Serialize, Deserialize)]
-pub struct UploadRecords {
- pub id: i64,
- pub identity: String,
- pub status: String,
-}
-
-impl Default for StudioEntity {
- fn default() -> Self {
- StudioEntity {
- id: 0,
- template_name: "".to_string(),
- user: 0,
- copyright: 1,
- source: "".to_string(),
- tid: 0,
- cover: "".to_string(),
- title: "".to_string(),
- desc: "".to_string(),
- dynamic: "".to_string(),
- tag: "".to_string(),
- dtime: None,
- interactive: 0,
- mission_id: None,
- dolby: 0,
- lossless_music: 0,
- no_reprint: 0,
- charging_pay: 0,
- up_selection_reply: false,
- up_close_reply: false,
- up_close_danmu: false,
- }
- }
-}
diff --git a/crates/biliup-cli/src/server/core/users.rs b/crates/biliup-cli/src/server/core/users.rs
deleted file mode 100644
index 6a664da..0000000
--- a/crates/biliup-cli/src/server/core/users.rs
+++ /dev/null
@@ -1,23 +0,0 @@
-use async_trait::async_trait;
-use serde::{Deserialize, Serialize};
-use sqlx::FromRow;
-use std::sync::Arc;
-
-pub type DynUsersRepository = Arc;
-
-#[derive(FromRow, Serialize, Deserialize)]
-pub struct User {
- pub id: i64,
- pub name: String,
- pub value: String,
- pub platform: String,
-}
-
-#[async_trait]
-pub trait UsersRepository {
- async fn create_user(&self, user: User) -> anyhow::Result;
- async fn get_users(&self) -> anyhow::Result>;
- async fn delete_user(&self, id: i64) -> anyhow::Result<()>;
- async fn update_user(&self, user: User) -> anyhow::Result;
- async fn get_user_by_id(&self, id: i64) -> anyhow::Result;
-}
diff --git a/crates/biliup-cli/src/server/core/util.rs b/crates/biliup-cli/src/server/core/util.rs
deleted file mode 100644
index 5dc7071..0000000
--- a/crates/biliup-cli/src/server/core/util.rs
+++ /dev/null
@@ -1,114 +0,0 @@
-use indexmap::IndexMap;
-use std::any::TypeId;
-use std::collections::HashMap;
-use std::error::Error;
-use std::fmt::{Debug, Formatter};
-use std::future::Future;
-use std::hash::{BuildHasherDefault, Hasher};
-use std::sync::{Arc, RwLock, RwLockWriteGuard};
-use tokio::task::JoinHandle;
-use tracing::error;
-
-pub type AnyMap
+
[爱发电 »](https://afdian.com/a/biliup)
## ⭐Stars
diff --git a/crates/biliup-cli/Cargo.toml b/crates/biliup-cli/Cargo.toml
index 20bc174..59e4a44 100644
--- a/crates/biliup-cli/Cargo.toml
+++ b/crates/biliup-cli/Cargo.toml
@@ -23,7 +23,7 @@ bytes = { workspace = true }
futures = { workspace = true }
clap = { version = "4", features = ["derive"] }
time = { workspace = true }
-anyhow = { workspace = true }
+
glob = "0.3.2"
qrcode = "0.14"
indicatif = "0.17.11"
@@ -33,7 +33,7 @@ chrono = { workspace = true, features = ["serde"] }
indexmap = { version = "2.8.0" }
async-trait = { workspace = true }
humantime = "2.2.0"
-axum = { version = "0.8" }
+axum = { version = "0.8", features = ["macros"] }
tower-http = { version = "0.6.2", features = ["cors"] }
reqwest = { workspace = true }
sqlx = { version = "0.8", features = [ "runtime-tokio-rustls", "sqlite" ] }
@@ -44,7 +44,18 @@ tempfile = "3.14"
rust-embed = "8.7.2"
mime_guess = "2.0.5"
ormlite = { version = "0.24.1", features = ["sqlite", "chrono"] }
-
+url = { workspace = true }
+error-stack = { workspace = true }
+bon = { workspace = true }
+axum-login = "0.18.0"
+password-auth = "1.0.0"
+tower = "0.5.2"
+tower-sessions = { version = "0.14.0", default-features = false, features = [
+ "signed",
+] }
+tower-sessions-sqlx-store = { version = "0.15.0", features = ["sqlite"] }
+async-channel = { version = "2.5.0" }
+nom = "8"
[features]
default = []
python-bridge = []
diff --git a/crates/biliup-cli/src/downloader.rs b/crates/biliup-cli/src/downloader.rs
index 9fcddb6..7172848 100644
--- a/crates/biliup-cli/src/downloader.rs
+++ b/crates/biliup-cli/src/downloader.rs
@@ -1,4 +1,3 @@
-use anyhow::{Context, Result};
use biliup::downloader::extractor::find_extractor;
use biliup::downloader::flv_parser::{
CodecId, SoundFormat, TagData, aac_audio_packet_header, avc_video_packet_header, header,
@@ -8,10 +7,11 @@ use biliup::downloader::flv_writer;
use biliup::downloader::flv_writer::{FlvTag, TagDataHeader};
use biliup::downloader::httpflv::map_parse_err;
use biliup::downloader::util::Segmentable;
+use biliup_cli::server::errors::{AppError, AppResult};
use bytes::{Buf, BufMut, Bytes, BytesMut};
+use error_stack::ResultExt;
use std::io::{BufReader, BufWriter, ErrorKind, Read};
use std::path::PathBuf;
-
use tracing::{error, info, warn};
pub async fn download(
@@ -19,22 +19,27 @@ pub async fn download(
output: String,
split_size: Option