diff --git a/.github/resource/Image.jpg b/.github/resource/Image.jpg new file mode 100644 index 0000000..120b5d4 Binary files /dev/null and b/.github/resource/Image.jpg differ diff --git a/Cargo.lock b/Cargo.lock index da158fe..361d578 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -3617,7 +3617,7 @@ checksum = "a8f112729512f8e442d81f95a8a7ddf2b7c6b8a1a6f509a95864142b30cab2d3" [[package]] name = "stream-gears" -version = "1.1.1" +version = "1.1.2" dependencies = [ "anyhow", "async-channel", diff --git a/Cargo.toml b/Cargo.toml index 29094d4..a98c51d 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -7,7 +7,7 @@ members = [ resolver = "3" [workspace.package] -version = "1.1.1" +version = "1.1.2" [workspace.dependencies] serde = { version = "1.0", features = ["derive"] } diff --git a/README.md b/README.md index 9cd0cd9..ba35e24 100644 --- a/README.md +++ b/README.md @@ -74,8 +74,8 @@ ## 💴捐赠 +![支付宝](.github/resource/Image.jpg) [爱发电 »](https://afdian.com/a/biliup) - ## ⭐Stars [![Star History Chart](https://api.star-history.com/svg?repos=biliup/biliup&type=Date)](https://star-history.com/#biliup/biliup&Date) diff --git a/app/lib/use-streamers.ts b/app/lib/use-streamers.ts index 45c02b5..34cb438 100644 --- a/app/lib/use-streamers.ts +++ b/app/lib/use-streamers.ts @@ -23,6 +23,7 @@ export function useBiliUsers() { const [list, setList] = useState([]); useEffect(() => { if (!data || data.length === 0) { + setList([]) return; } const updateList = async (item: User) => { diff --git a/app/ui/QRcode.tsx b/app/ui/QRcode.tsx index a5e94c5..4329248 100644 --- a/app/ui/QRcode.tsx +++ b/app/ui/QRcode.tsx @@ -29,16 +29,11 @@ const Qrcode: React.FC = ({ onSuccess }) => { body: JSON.stringify(qrData), signal: signal, }) - if (!(res.status >= 200 && res.status < 300)) { - throw new Error(await res.text()) - } const data = await res.json() - if (!res.ok) { - throw new Error(data.message) - } onSuccess(data['filename']) })().catch(e => { console.log(e) + setUrl(e.message) Notification.error({ title: 'QRcode', content: {e.message}, @@ -53,6 +48,10 @@ const Qrcode: React.FC = ({ onSuccess }) => { if (url === '') { return } + + if (!url.startsWith("http")) { + return <> {url} + } return (
= ({ onCancel, visible }) => { - const { trigger } = useSWRMutation('/v1/users', sendRequest) + const { trigger } = useSWRMutation('/v1/users', sendRequest) const { trigger: deleteUser } = useSWRMutation('/v1/users', requestDelete) const { biliUsers: list } = useBiliUsers() const [modalVisible, setVisible] = useState(false) @@ -55,10 +55,12 @@ const UserList: React.FC = ({ onCancel, visible }) => { throw new Error(ret.message) } await trigger({ - id: 0, - name: value, - value: value, - platform: 'bilibili-cookies', + // id: 0, + // name: value, + // value: value, + // platform: 'bilibili-cookies', + key: 'bilibili-cookies', + value: value }) setVisible(false) Toast.success('创建成功') diff --git a/crates/biliup-cli/src/downloader.rs b/crates/biliup-cli/src/downloader.rs index a5935fa..9fcddb6 100644 --- a/crates/biliup-cli/src/downloader.rs +++ b/crates/biliup-cli/src/downloader.rs @@ -137,7 +137,7 @@ pub fn generate_json(mut file_name: PathBuf) -> Result<()> { script_tag_count += 1; let (_, tag_data) = script_data(i).unwrap(); - + FlvTag { header: tag_header, data: TagDataHeader::Script(tag_data), diff --git a/crates/biliup-cli/src/server/api/endpoints.rs b/crates/biliup-cli/src/server/api/endpoints.rs index 841590e..36c8f1d 100644 --- a/crates/biliup-cli/src/server/api/endpoints.rs +++ b/crates/biliup-cli/src/server/api/endpoints.rs @@ -1,7 +1,6 @@ use crate::server::core::download_actor::DownloadActorHandle; use crate::server::core::live_streamers::{ - AddLiveStreamerDto, DynLiveStreamersService, LiveStreamerDto, - LiveStreamerEntity, + AddLiveStreamerDto, DynLiveStreamersService, LiveStreamerDto, LiveStreamerEntity, }; use crate::server::core::upload_streamers::{DynUploadStreamersRepository, StudioEntity}; use crate::server::core::users::User; diff --git a/crates/biliup-cli/src/server/config_manager.rs b/crates/biliup-cli/src/server/config_manager.rs index 5156929..5fc48d2 100644 --- a/crates/biliup-cli/src/server/config_manager.rs +++ b/crates/biliup-cli/src/server/config_manager.rs @@ -349,33 +349,35 @@ impl ConfigManager { 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); + && 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()); - } + && let Some(ref mut user) = self.config.user + { + user.access_token = Some(access_token.to_string()); + } Ok(()) } diff --git a/crates/biliup-cli/src/server/event_manager.rs b/crates/biliup-cli/src/server/event_manager.rs index 05a252f..8b3901a 100644 --- a/crates/biliup-cli/src/server/event_manager.rs +++ b/crates/biliup-cli/src/server/event_manager.rs @@ -6,7 +6,6 @@ use std::sync::Arc; use tokio::sync::{RwLock, mpsc}; use tracing::{debug, error, info}; - /// 事件类型枚举，对应Python版本的事件 #[derive(Debug, Clone, PartialEq, Eq, Hash, Serialize, Deserialize)] pub enum EventType { diff --git a/crates/biliup-cli/src/server/infrastructure/repositories/upload_streamers_repository.rs b/crates/biliup-cli/src/server/infrastructure/repositories/upload_streamers_repository.rs index 1c22576..ab8afc0 100644 --- a/crates/biliup-cli/src/server/infrastructure/repositories/upload_streamers_repository.rs +++ b/crates/biliup-cli/src/server/infrastructure/repositories/upload_streamers_repository.rs @@ -1,6 +1,5 @@ use crate::server::infrastructure::connection_pool::ConnectionPool; - #[derive(Clone)] pub struct SqliteUploadStreamersRepository { pool: ConnectionPool, diff --git a/crates/biliup-cli/src/server/infrastructure/service_register.rs b/crates/biliup-cli/src/server/infrastructure/service_register.rs index 555e0f4..f73abe9 100644 --- a/crates/biliup-cli/src/server/infrastructure/service_register.rs +++ b/crates/biliup-cli/src/server/infrastructure/service_register.rs @@ -1,7 +1,4 @@ -use crate::server::core::live_streamers::{ - DynDownloadRecordsRepository, - DynVideosRepository, -}; +use crate::server::core::live_streamers::{DynDownloadRecordsRepository, DynVideosRepository}; use crate::server::core::upload_streamers::DynUploadRecordsRepository; use crate::server::infrastructure::connection_pool::ConnectionPool; use crate::server::infrastructure::repositories::download_records_repository::SqliteDownloadRecordsRepository; diff --git a/crates/biliup/src/downloader.rs b/crates/biliup/src/downloader.rs index 6f8a2b2..2af9d60 100644 --- a/crates/biliup/src/downloader.rs +++ b/crates/biliup/src/downloader.rs @@ -1,6 +1,3 @@ - - - pub mod error; pub mod extractor; pub mod flv_parser; diff --git a/crates/biliup/src/downloader/extractor/douyu.rs b/crates/biliup/src/downloader/extractor/douyu.rs index f89975a..93cbd60 100644 --- a/crates/biliup/src/downloader/extractor/douyu.rs +++ b/crates/biliup/src/downloader/extractor/douyu.rs @@ -76,20 +76,20 @@ impl SiteDefinition for DouyuLive { && let Some(key) = regex::Regex::new(r"(\d{1,8}[0-9a-zA-Z]+)_?\d{0,4}(/playlist|.m3u8)") .unwrap() .captures(&result["data"]["rtmp_live"].to_string()) - { - return Ok(Site { - name: "douyu", - title: room_info - .get("room") - .and_then(|room| room.get("room_name")) - .and_then(|name| name.as_str()) - .unwrap_or_default() - .to_string(), - direct_url: format!("https://hw-tct.douyucdn.cn/live/{}.flv?uuid=", &key[1]), - extension: Extension::Flv, - client, - }); - } + { + return Ok(Site { + name: "douyu", + title: room_info + .get("room") + .and_then(|room| room.get("room_name")) + .and_then(|name| name.as_str()) + .unwrap_or_default() + .to_string(), + direct_url: format!("https://hw-tct.douyucdn.cn/live/{}.flv?uuid=", &key[1]), + extension: Extension::Flv, + client, + }); + } Err(Error::Custom(result.to_string())) } diff --git a/crates/biliup/src/downloader/httpflv.rs b/crates/biliup/src/downloader/httpflv.rs index fc61c04..f355c5e 100644 --- a/crates/biliup/src/downloader/httpflv.rs +++ b/crates/biliup/src/downloader/httpflv.rs @@ -129,7 +129,6 @@ pub(crate) async fn parse_flv( } on_meta_data = Some((tag_header, bytes.clone(), previous_tag_size.clone())); - FlvTag { header: tag_header, data: TagDataHeader::Script(tag_data), diff --git a/crates/biliup/src/uploader/credential.rs b/crates/biliup/src/uploader/credential.rs index 8d62f45..9ae1f3e 100644 --- a/crates/biliup/src/uploader/credential.rs +++ b/crates/biliup/src/uploader/credential.rs @@ -113,7 +113,7 @@ pub struct LoginInfo { pub struct TokenInfo { pub access_token: String, expires_in: u32, - mid: u64, + pub mid: u64, refresh_token: String, } diff --git a/crates/stream-gears/src/server.rs b/crates/stream-gears/src/server.rs index 5556ae8..3a21859 100644 --- a/crates/stream-gears/src/server.rs +++ b/crates/stream-gears/src/server.rs @@ -56,9 +56,8 @@ async fn _main() -> AppResult<()> { let conn_pool = ConnectionManager::new_pool("data/data.sqlite3") .await .expect("could not initialize the database connection pool"); - - let configs = repositories::get_config(&conn_pool).await?; - let service_register = ServiceRegister::new(conn_pool, configs); + *config::CONFIG.write().unwrap() = repositories::get_config(&conn_pool).await?; + let service_register = ServiceRegister::new(conn_pool, config::CONFIG.clone()); let all_streamer = repositories::get_all_streamer(&service_register.pool).await?; let all_uploader = repositories::get_all_uploader(&service_register.pool).await?; diff --git a/crates/stream-gears/src/server/api/auth.rs b/crates/stream-gears/src/server/api/auth.rs index 1e543f2..840ac9a 100644 --- a/crates/stream-gears/src/server/api/auth.rs +++ b/crates/stream-gears/src/server/api/auth.rs @@ -92,7 +92,7 @@ mod post { mod get { use super::*; - + use tracing::error; // pub async fn login( diff --git a/crates/stream-gears/src/server/api/bilibili_endpoints.rs b/crates/stream-gears/src/server/api/bilibili_endpoints.rs index b5b8d18..25edb11 100644 --- a/crates/stream-gears/src/server/api/bilibili_endpoints.rs +++ b/crates/stream-gears/src/server/api/bilibili_endpoints.rs @@ -53,7 +53,7 @@ pub async fn get_myinfo_endpoint( } pub async fn get_proxy_endpoint( - Extension(client): Extension, + State(client): State, Query(params): Query>, ) -> Result { info!(params = ¶ms["url"]); diff --git a/crates/stream-gears/src/server/api/endpoints.rs b/crates/stream-gears/src/server/api/endpoints.rs index 9269b83..82c9907 100644 --- a/crates/stream-gears/src/server/api/endpoints.rs +++ b/crates/stream-gears/src/server/api/endpoints.rs @@ -1,23 +1,23 @@ -use crate::server::errors::{AppError, report_to_response}; -use std::sync::{Arc, RwLock}; - use crate::server::config::Config; +use crate::server::errors::{AppError, report_to_response, AppResult}; 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, InsertLiveStreamer, InsertUploadStreamer, LiveStreamer, StreamerInfo, - UploadStreamer, -}; +use crate::server::infrastructure::models::{Configuration, FileItem, InsertConfiguration, InsertLiveStreamer, InsertUploadStreamer, LiveStreamer, StreamerInfo, UploadStreamer}; use crate::server::infrastructure::repositories::{del_streamer, get_all_streamer}; use crate::server::infrastructure::service_register::ServiceRegister; use axum::Json; use axum::extract::{Path, State}; use axum::http::StatusCode; use axum::response::{IntoResponse, Response}; +use biliup::credential::{Credential, LoginInfo}; use error_stack::ResultExt; use ormlite::{Insert, Model}; use serde_json::json; +use std::sync::{Arc, RwLock}; +use std::time::{Duration, UNIX_EPOCH}; +use tokio::fs; +use tokio::io::AsyncWriteExt; use tracing::info; pub async fn get_streamers_endpoint( @@ -244,6 +244,98 @@ pub async fn get_users_endpoint( Ok(Json(res)) } -pub async fn delete_user_endpoint(Path(id): Path) -> Result, Response> { +pub async fn add_user_endpoint( + State(pool): State, + Json(user): Json, +) -> Result, Response> { + let res = user.insert(&pool).await.change_context(AppError::Unknown).map_err(report_to_response)?; + Ok(Json(res)) +} + +pub async fn delete_user_endpoint(Path(id): Path, State(pool): State) -> Result, Response> { + let x = sqlx::query("DELETE FROM configuration WHERE id = ?") + .bind(id) + .execute(&pool).await.change_context(AppError::Unknown).map_err(report_to_response)?; + info!("{:?}", x); Ok(Json(())) } + +pub async fn get_qrcode() -> Result, Response> { + let qrcode = Credential::new(None) + .get_qrcode() + .await + .change_context(AppError::Unknown) + .map_err(report_to_response)?; + Ok(Json(qrcode)) +} + +pub async fn login_by_qrcode( + Json(value): Json, +) -> Result, Response> { + let info = tokio::time::timeout( + Duration::from_secs(300), + Credential::new(None).login_by_qrcode(value), + // std::future::pending::>(), + ) + .await + .change_context(AppError::Custom("deadline has elapsed".to_string())) + .map_err(report_to_response)? + .change_context(AppError::Unknown) + .map_err(report_to_response)?; + + // extract mid + let mid = info.token_info.mid; + let filename = format!("data/{}.json", mid); + + let mut file = fs::File::create(&filename) + .await + .change_context(AppError::Unknown) + .map_err(report_to_response)?; + file.write_all(&serde_json::to_vec_pretty(&info).unwrap()) + .await + .change_context(AppError::Unknown) + .map_err(report_to_response)?; + + Ok(Json(json!({ "filename": filename }))) +} + +pub async fn get_videos() -> Result>, Response> { + let media_extensions = [".mp4", ".flv", ".3gp", ".webm", ".mkv", ".ts"]; + let blacklist = ["next-env.d.ts"]; + + let mut file_list = Vec::new(); + let mut index = 1; + + // **use tokio::fs::read_dir** + if let Ok(mut entries) = fs::read_dir(".").await { + while let Ok(Some(entry)) = entries.next_entry().await { + let path = entry.path(); + let file_name = entry.file_name().to_string_lossy().into_owned(); + + if blacklist.contains(&file_name.as_str()) { + continue; + } + + if let Some(ext) = path.extension().and_then(|e| e.to_str()) { + if media_extensions.iter().any(|allowed| &ext == &allowed.trim_start_matches('.')) { + if let Ok(metadata) = entry.metadata().await { + let mtime = metadata.modified() + .ok() + .and_then(|time| time.duration_since(UNIX_EPOCH).ok()) + .map(|d| d.as_secs()) + .unwrap_or(0); + + file_list.push(serde_json::json!({ + "key": index, + "name": file_name, + "updateTime": mtime, + "size": metadata.len(), + })); + index += 1; + } + } + } + } + } + Ok(Json(file_list)) +} diff --git a/crates/stream-gears/src/server/app.rs b/crates/stream-gears/src/server/app.rs index dd14ae4..e33d37f 100644 --- a/crates/stream-gears/src/server/app.rs +++ b/crates/stream-gears/src/server/app.rs @@ -58,10 +58,9 @@ impl ApplicationController { // let app = protected::router() - let client = StatelessClient::default(); // let vec = service_register.streamers_service.get_streamers().await?; // build our application with a route - let enable_login_guard = true; + let enable_login_guard = false; let mut app = server::router::router(service_register); if enable_login_guard { app = app @@ -82,7 +81,7 @@ impl ApplicationController { .allow_origin("http://localhost:3000".parse::().unwrap()) .allow_methods(AllowMethods::any()), ) - .layer(Extension(client.clone())) + // .layer(Extension(client.clone())) .fallback(static_handler); // run our app with hyper // `axum::Server` is a re-export of `hyper::Server` diff --git a/crates/stream-gears/src/server/config.rs b/crates/stream-gears/src/server/config.rs index 436af3f..5e40d9f 100644 --- a/crates/stream-gears/src/server/config.rs +++ b/crates/stream-gears/src/server/config.rs @@ -7,8 +7,9 @@ use pyo3::types::{PyAnyMethods, PyDict}; use pyo3::{Bound, FromPyObject, PyAny, PyResult, Python, pyclass, pyfunction, pymethods}; use pythonize::pythonize; use serde::{Deserialize, Serialize}; +use std::cell::LazyCell; use std::ops::Deref; -use std::sync::{Arc, OnceLock, RwLock}; +use std::sync::{Arc, LazyLock, OnceLock, RwLock}; use std::{collections::HashMap, path::Path, path::PathBuf}; #[derive(bon::Builder, Debug, Clone, Serialize, Deserialize)] @@ -329,10 +330,14 @@ impl Config { } // 进程级全局单例（安全）：OnceLock + Arc + RwLock -pub(crate) static CONFIG: OnceLock>> = OnceLock::new(); +pub(crate) static CONFIG: LazyLock>> = LazyLock::new(|| { + Arc::new(RwLock::new( + Config::builder().streamers(Default::default()).build(), + )) +}); fn cfg_arc() -> &'static Arc> { - CONFIG.get().expect("Config not initialized") + &*CONFIG } #[pyclass] diff --git a/crates/stream-gears/src/server/core/download_manager.rs b/crates/stream-gears/src/server/core/download_manager.rs index 54f1906..489838a 100644 --- a/crates/stream-gears/src/server/core/download_manager.rs +++ b/crates/stream-gears/src/server/core/download_manager.rs @@ -126,7 +126,7 @@ impl UActor { return; }; let config = worker.get_config().await.unwrap(); - let client = StatelessClient::default(); + let client = worker.context.client.clone(); let line = config.lines; let line = match line.as_str() { "bda2" => line::bda2(), @@ -199,9 +199,10 @@ impl UActor { if !studio.cover.is_empty() && let Ok(c) = &std::fs::read(&studio.cover).map_err(|e| error!(e=?e)) - && let Ok(url) = bilibili.cover_up(c).await.map_err(|e| error!(e=?e)) { - studio.cover = url; - }; + && let Ok(url) = bilibili.cover_up(c).await.map_err(|e| error!(e=?e)) + { + studio.cover = url; + }; let submit = match config.submit_api { Some(submit) => SubmitOption::from_str(&submit).unwrap_or(SubmitOption::App), diff --git a/crates/stream-gears/src/server/infrastructure/connection_pool.rs b/crates/stream-gears/src/server/infrastructure/connection_pool.rs index 7148d7d..a191015 100644 --- a/crates/stream-gears/src/server/infrastructure/connection_pool.rs +++ b/crates/stream-gears/src/server/infrastructure/connection_pool.rs @@ -1,7 +1,7 @@ -use std::path::Path; use anyhow::{Context, Result}; use sqlx::sqlite::SqlitePoolOptions; use sqlx::{Pool, Sqlite}; +use std::path::Path; use tracing::info; pub type ConnectionPool = Pool; diff --git a/crates/stream-gears/src/server/infrastructure/context.rs b/crates/stream-gears/src/server/infrastructure/context.rs index 2c0a08d..6e6bc8a 100644 --- a/crates/stream-gears/src/server/infrastructure/context.rs +++ b/crates/stream-gears/src/server/infrastructure/context.rs @@ -3,29 +3,26 @@ use crate::server::errors::{AppError, AppResult}; use crate::server::infrastructure::connection_pool::ConnectionPool; use crate::server::infrastructure::models::{Configuration, LiveStreamer, UploadStreamer}; use crate::server::infrastructure::repositories::{get_config, get_streamer}; +use biliup::client::StatelessClient; use error_stack::ResultExt; use ormlite::Model; use serde::{Deserialize, Serialize}; use std::sync::RwLock; use tracing::info; +#[derive(Debug)] pub struct Context { - uploads: Vec, - configuration: Configuration, - streamers: Vec, + pool: ConnectionPool, + pub client: StatelessClient, } impl Context { - pub fn new( - uploads: Vec, - configuration: Configuration, - streamers: Vec, - ) -> Self { - Self { - uploads, - configuration, - streamers, - } + pub fn new(pool: ConnectionPool, client: StatelessClient) -> Self { + Self { pool, client } + } + + pub async fn get_config(&self) -> AppResult { + get_config(&self.pool).await } } @@ -33,16 +30,16 @@ impl Context { pub struct Worker { pub(crate) id: i64, pub url: String, - pool: ConnectionPool, + pub context: Context, pub downloader_status: RwLock, pub uploader_status: RwLock, } impl Worker { - pub fn new(id: i64, url: &str, pool: ConnectionPool) -> Self { + pub fn new(id: i64, url: &str, context: Context) -> Self { Self { id, - pool, + context, downloader_status: RwLock::new(Default::default()), uploader_status: Default::default(), url: url.to_string(), @@ -50,11 +47,11 @@ impl Worker { } pub async fn get_streamer(&self) -> AppResult { - get_streamer(&self.pool, self.id).await + get_streamer(&self.context.pool, self.id).await } pub async fn get_config(&self) -> AppResult { - get_config(&self.pool).await + self.context.get_config().await } pub async fn get_upload_config(&self) -> AppResult> { @@ -65,7 +62,7 @@ impl Worker { UploadStreamer::select() .where_("id = ?") .bind(id) - .fetch_optional(&self.pool) + .fetch_optional(&self.context.pool) .await .change_context(AppError::Unknown) } diff --git a/crates/stream-gears/src/server/infrastructure/models.rs b/crates/stream-gears/src/server/infrastructure/models.rs index 7c9027a..93b8880 100644 --- a/crates/stream-gears/src/server/infrastructure/models.rs +++ b/crates/stream-gears/src/server/infrastructure/models.rs @@ -61,12 +61,14 @@ pub struct LiveStreamer { // #[ormlite(json)] 数据库空与json空有区别所以这里不能用 // pub preprocessor: Option>, // 只能使用 sqlx::types::Json - pub preprocessor: sqlx::types::Json>>, - - pub segment_processor: sqlx::types::Json>>, - - pub downloaded_processor: sqlx::types::Json>>, - pub postprocessor: 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, } @@ -80,7 +82,7 @@ pub struct FileItem { } #[derive(Model, Debug, Clone, Serialize, Deserialize)] -#[ormlite(table = "configuration", insert = "InsertConfiguration")] +#[ormlite(table = "configuration")] pub struct Configuration { pub id: i64, pub key: String, @@ -110,12 +112,14 @@ pub struct InsertLiveStreamer { // #[ormlite(json)] 数据库空与json空有区别所以这里不能用 // pub preprocessor: Option>, // 只能使用 sqlx::types::Json - pub preprocessor: sqlx::types::Json>>, - - pub segment_processor: sqlx::types::Json>>, - - pub downloaded_processor: sqlx::types::Json>>, - pub postprocessor: 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, } @@ -148,3 +152,10 @@ pub struct InsertUploadStreamer { pub extra_fields: Option, pub is_only_self: Option, } + +#[derive(Insert, Debug, Clone, Serialize, Deserialize)] +#[ormlite(returns = "Configuration")] +pub struct InsertConfiguration { + pub key: String, + pub value: String, // TEXT +} diff --git a/crates/stream-gears/src/server/infrastructure/repositories.rs b/crates/stream-gears/src/server/infrastructure/repositories.rs index f03bb80..832c9f9 100644 --- a/crates/stream-gears/src/server/infrastructure/repositories.rs +++ b/crates/stream-gears/src/server/infrastructure/repositories.rs @@ -1,7 +1,9 @@ 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}; +use crate::server::infrastructure::models::{ + Configuration, InsertConfiguration, LiveStreamer, UploadStreamer, +}; use error_stack::ResultExt; use ormlite::{Insert, Model}; @@ -41,18 +43,20 @@ pub async fn get_config(pool: &ConnectionPool) -> AppResult { serde_json::from_str(&configuration.value).change_context(AppError::Unknown)?; Ok(json) } else { - let config = Config::builder() - .streamers(Default::default()) - .build(); - let configuration = InsertConfiguration { - key: "config".to_string(), - value: serde_json::to_string(&config).unwrap(), - }.insert(pool).await.change_context(AppError::Unknown)?; - let json: Config = - serde_json::from_str(&configuration.value).change_context(AppError::Unknown)?; - Ok(json) + let config = Config::builder().streamers(Default::default()).build(); + Ok(config) } +} +pub async fn insert_config(pool: &ConnectionPool, config: &Config) -> AppResult { + let configuration = InsertConfiguration { + key: "config".to_string(), + value: serde_json::to_string(config).unwrap(), + } + .insert(pool) + .await + .change_context(AppError::Unknown)?; + Ok(configuration) } pub async fn get_all_uploader(pool: &ConnectionPool) -> AppResult> { diff --git a/crates/stream-gears/src/server/infrastructure/service_register.rs b/crates/stream-gears/src/server/infrastructure/service_register.rs index f8a4dfd..51e6f63 100644 --- a/crates/stream-gears/src/server/infrastructure/service_register.rs +++ b/crates/stream-gears/src/server/infrastructure/service_register.rs @@ -5,8 +5,11 @@ use crate::server::core::monitor::Monitor; use crate::server::core::plugin; use crate::server::errors::{AppError, AppResult}; use crate::server::infrastructure::connection_pool::ConnectionPool; -use crate::server::infrastructure::context::Worker; +use crate::server::infrastructure::context::{Context, Worker}; +use crate::server::infrastructure::repositories; use axum::extract::FromRef; +use biliup::client::StatelessClient; +use biliup::credential::Credential; use error_stack::bail; use ormlite::Model; use std::sync::{Arc, RwLock}; @@ -19,27 +22,19 @@ pub struct ServiceRegister { pub managers: Arc>, pub config: Arc>, pub actor_handle: Arc, + pub client: StatelessClient, } /// A simple service container responsible for managing the various services our API endpoints will pull from through axum extensions. impl ServiceRegister { - pub fn new(pool: ConnectionPool, config: Config) -> Self { + pub fn new(pool: ConnectionPool, config: Arc>) -> Self { info!("initializing utility services..."); + let client = StatelessClient::default(); - let cf = config::CONFIG.get_or_init(|| { - Arc::new(RwLock::new(config)) - // Arc::new(RwLock::new(Config::builder() - // // .user(UserConfig::default()) - // // .douyu_cdn("hw-h5".to_string()) - // // .douyu_rate(0) - // .streamers(HashMap::::new()) - // .build())) - }); - - info!(config=?cf); + info!(config=?config); let actor_handle = Arc::new(ActorHandle::new( - cf.read().unwrap().pool1_size, - cf.read().unwrap().pool2_size, + config.read().unwrap().pool1_size, + config.read().unwrap().pool2_size, )); info!("utility services initialized, building feature services..."); @@ -52,13 +47,17 @@ impl ServiceRegister { pool, workers: Arc::new(Default::default()), managers: Arc::new(vec), - config: cf.clone(), + config: config.clone(), actor_handle, + client, } } pub fn get_manager(&self, url: &str) -> Option<&DownloadManager> { - self.managers.iter().find(|&manager| manager.matches(url)).map(|v| v as _) + self.managers + .iter() + .find(|&manager| manager.matches(url)) + .map(|v| v as _) } pub async fn add_room( @@ -67,7 +66,11 @@ impl ServiceRegister { url: &str, monitor: Arc, ) -> AppResult> { - let worker = Arc::new(Worker::new(id, url, self.pool.clone())); + let worker = Arc::new(Worker::new( + id, + url, + Context::new(self.pool.clone(), self.client.clone()), + )); monitor.rooms_handle.add(worker.clone()).await; self.workers.write().unwrap().push(worker); info!("add {url} success"); diff --git a/crates/stream-gears/src/server/router.rs b/crates/stream-gears/src/server/router.rs index 35308c0..7b220fa 100644 --- a/crates/stream-gears/src/server/router.rs +++ b/crates/stream-gears/src/server/router.rs @@ -1,15 +1,10 @@ use crate::server::api::bilibili_endpoints::{ archive_pre_endpoint, get_myinfo_endpoint, get_proxy_endpoint, }; -use crate::server::api::endpoints::{ - add_upload_streamer_endpoint, delete_streamers_endpoint, delete_template_endpoint, - delete_user_endpoint, get_configuration, get_streamer_info, get_streamers_endpoint, - get_upload_streamer_endpoint, get_upload_streamers_endpoint, get_users_endpoint, - post_streamers_endpoint, put_streamers_endpoint, -}; +use crate::server::api::endpoints::{add_upload_streamer_endpoint, add_user_endpoint, delete_streamers_endpoint, delete_template_endpoint, delete_user_endpoint, get_configuration, get_qrcode, get_streamer_info, get_streamers_endpoint, get_upload_streamer_endpoint, get_upload_streamers_endpoint, get_users_endpoint, get_videos, login_by_qrcode, post_streamers_endpoint, put_streamers_endpoint}; use crate::server::infrastructure::service_register::ServiceRegister; -use axum::routing::{delete, get, post}; use axum::Router; +use axum::routing::{delete, get, post}; pub fn router(service_register: ServiceRegister) -> Router<()> { Router::new() @@ -42,13 +37,16 @@ pub fn router(service_register: ServiceRegister) -> Router<()> { .route("/v1/upload/streamers", post(add_upload_streamer_endpoint)) .route( "/v1/users", - get(get_users_endpoint), // .post(add_user_endpoint) + get(get_users_endpoint) + .post(add_user_endpoint), ) .route("/v1/users/{id}", delete(delete_user_endpoint)) .route("/bili/archive/pre", get(archive_pre_endpoint)) .route("/bili/space/myinfo", get(get_myinfo_endpoint)) .route("/bili/proxy", get(get_proxy_endpoint)) - // .route("/", get(root)) + .route("/v1/get_qrcode", get(get_qrcode)) + .route("/v1/login_by_qrcode", post(login_by_qrcode)) + .route("/v1/videos", get(get_videos)) // .layer(Extension(client.clone())) .with_state(service_register) } diff --git a/crates/stream-gears/src/server/util.rs b/crates/stream-gears/src/server/util.rs index 7a0929a..4fcf270 100644 --- a/crates/stream-gears/src/server/util.rs +++ b/crates/stream-gears/src/server/util.rs @@ -104,9 +104,10 @@ pub fn media_ext_from_url(input: &str) -> Option { // a) 从最后一个 path segment 取扩展名 if let Some(seg) = url.path_segments().and_then(|mut s| s.next_back()) && let Some((_, ext)) = seg.rsplit_once('.') - && let Some(ext) = clean_ext(ext) { - return Some(ext); - } + && let Some(ext) = clean_ext(ext) + { + return Some(ext); + } // b) 常见 query 参数中找一次（不重复多轮扫描），忽略大小写 let keys = ["format", "type", "ext", "filetype", "fmt"];