diff --git a/crates/biliup-cli/src/cli.rs b/crates/biliup-cli/src/cli.rs index 60a33de..7d2d95b 100644 --- a/crates/biliup-cli/src/cli.rs +++ b/crates/biliup-cli/src/cli.rs @@ -123,8 +123,8 @@ pub enum Commands { /// Port to use #[arg(short, long, default_value = "19159")] - port: u16, - + port: u16, + /// Port to use #[arg(short, long, default_value = "false")] auth: bool, diff --git a/crates/biliup-cli/src/downloader.rs b/crates/biliup-cli/src/downloader.rs index 5184522..75cb40e 100644 --- a/crates/biliup-cli/src/downloader.rs +++ b/crates/biliup-cli/src/downloader.rs @@ -1,3 +1,4 @@ +use crate::server::errors::{AppError, AppResult}; use biliup::downloader::extractor::find_extractor; use biliup::downloader::flv_parser::{ CodecId, SoundFormat, TagData, aac_audio_packet_header, avc_video_packet_header, header, @@ -7,7 +8,6 @@ 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 crate::server::errors::{AppError, AppResult}; use bytes::{Buf, BufMut, Bytes, BytesMut}; use error_stack::ResultExt; use std::io::{BufReader, BufWriter, ErrorKind, Read}; diff --git a/crates/biliup-cli/src/lib.rs b/crates/biliup-cli/src/lib.rs index ee40184..e3a4a19 100644 --- a/crates/biliup-cli/src/lib.rs +++ b/crates/biliup-cli/src/lib.rs @@ -1,6 +1,6 @@ -pub mod server; pub mod cli; pub mod downloader; +pub mod server; pub mod uploader; // use crate::server::api::router::ApplicationController; diff --git a/crates/biliup-cli/src/main.rs b/crates/biliup-cli/src/main.rs index c02d516..5f085ce 100644 --- a/crates/biliup-cli/src/main.rs +++ b/crates/biliup-cli/src/main.rs @@ -1,9 +1,9 @@ use time::macros::format_description; +use biliup::uploader::util::SubmitOption; use biliup_cli::cli::{Cli, Commands}; use biliup_cli::downloader::{download, generate_json}; use biliup_cli::uploader::{append, list, login, renew, show, upload_by_command, upload_by_config}; -use biliup::uploader::util::SubmitOption; use clap::Parser; @@ -97,7 +97,7 @@ async fn main() -> AppResult<()> { split_size, split_time, } => download(&url, output, split_size, split_time).await?, - Commands::Server { bind, port, auth } => biliup_cli::run((&bind, port)).await?, + Commands::Server { bind, port, auth } => biliup_cli::run((&bind, port), auth).await?, Commands::List { is_pubing, pubed, diff --git a/crates/biliup-cli/src/server/api/endpoints.rs b/crates/biliup-cli/src/server/api/endpoints.rs index e9feb08..df88764 100644 --- a/crates/biliup-cli/src/server/api/endpoints.rs +++ b/crates/biliup-cli/src/server/api/endpoints.rs @@ -19,10 +19,10 @@ use crate::server::infrastructure::repositories::{ del_streamer, get_all_streamer, get_upload_config, }; 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 axum::Json; use biliup::credential::Credential; use chrono::Utc; use clap::ValueEnum; @@ -34,7 +34,7 @@ use std::path::PathBuf; use std::sync::{Arc, RwLock}; use std::time::{Duration, UNIX_EPOCH}; use tokio::fs; -use tokio::io::{AsyncWriteExt}; +use tokio::io::AsyncWriteExt; use tracing::info; pub async fn get_streamers_endpoint( diff --git a/crates/biliup-cli/src/server/app.rs b/crates/biliup-cli/src/server/app.rs index 10f9b4a..2162ab6 100644 --- a/crates/biliup-cli/src/server/app.rs +++ b/crates/biliup-cli/src/server/app.rs @@ -26,7 +26,11 @@ pub struct ApplicationController; impl ApplicationController { /// 启动Web服务器 - pub async fn serve(addr: &SocketAddr, enable_login_guard: bool, service_register: ServiceRegister) -> AppResult<()> { + pub async fn serve( + addr: &SocketAddr, + enable_login_guard: bool, + service_register: ServiceRegister, + ) -> AppResult<()> { // 会话层配置 // 使用 tower-sessions 建立会话层,将会话作为请求扩展提供 let session_store = SqliteStore::new(service_register.pool.clone()); diff --git a/crates/biliup-cli/src/server/common/download.rs b/crates/biliup-cli/src/server/common/download.rs index 6c3ab42..d29d4e7 100644 --- a/crates/biliup-cli/src/server/common/download.rs +++ b/crates/biliup-cli/src/server/common/download.rs @@ -58,9 +58,10 @@ impl DownloadGuard { let worker = self.worker.clone(); if let Some(client) = danmaku - && let Err(e) = client.stop().await { - error!("Error stopping danmaku client: {}", e); - } + && let Err(e) = client.stop().await + { + error!("Error stopping danmaku client: {}", e); + } // 确保状态更新和资源清理 rooms_handle @@ -219,9 +220,9 @@ impl SegmentEventProcessor { if let Some(ref client) = danmaku && let Err(e) = client .rolling(&next_file_path.with_extension("xml").display().to_string()) - { - error!("Danmaku rolling error: {}", e); - } + { + error!("Danmaku rolling error: {}", e); + } } SegmentEvent::Segment(event) => { // 触发弹幕滚动保存 @@ -232,9 +233,10 @@ impl SegmentEventProcessor { .with_extension("xml") .display() .to_string(), - ) { - error!("Danmaku rolling error: {}", e); - } + ) + { + error!("Danmaku rolling error: {}", e); + } // 异步处理事件 let processor = processor.clone(); if let Err(e) = processor.process(event) { diff --git a/crates/biliup-cli/src/server/common/upload.rs b/crates/biliup-cli/src/server/common/upload.rs index 11081f4..68c7e46 100644 --- a/crates/biliup-cli/src/server/common/upload.rs +++ b/crates/biliup-cli/src/server/common/upload.rs @@ -417,7 +417,9 @@ impl UActor { // 可以添加错误通知机制 } info!(url=ctx.stream_info.streamer_info.url, result=?result, "后处理执行完毕:Finished processing segment event"); - ctx.worker.change_status(Stage::Upload, WorkerStatus::Idle).await; + ctx.worker + .change_status(Stage::Upload, WorkerStatus::Idle) + .await; } } } diff --git a/crates/biliup-cli/src/server/infrastructure/models/hook_step.rs b/crates/biliup-cli/src/server/infrastructure/models/hook_step.rs index 55a1968..d8fd59e 100644 --- a/crates/biliup-cli/src/server/infrastructure/models/hook_step.rs +++ b/crates/biliup-cli/src/server/infrastructure/models/hook_step.rs @@ -88,9 +88,10 @@ impl HookStep { // 等待进程完成并检查退出状态 let status = process.wait().await.change_context(AppError::Unknown)?; if !status.success() { - bail!(AppError::Custom( - format!("Command failed with status: {}", status) - )); + bail!(AppError::Custom(format!( + "Command failed with status: {}", + status + ))); } Ok(()) diff --git a/crates/biliup-cli/src/uploader.rs b/crates/biliup-cli/src/uploader.rs index 4654250..31f4f79 100644 --- a/crates/biliup-cli/src/uploader.rs +++ b/crates/biliup-cli/src/uploader.rs @@ -1,3 +1,5 @@ +use crate::UploadLine; +use crate::server::errors::{AppError, AppResult}; use biliup::client::StatelessClient; use biliup::error::Kind; use biliup::uploader::bilibili::{BiliBili, Studio, Vid, Video}; @@ -5,8 +7,6 @@ use biliup::uploader::credential::{Credential, LoginInfo}; use biliup::uploader::line::Probe; use biliup::uploader::util::SubmitOption; use biliup::uploader::{VideoFile, credential, line, load_config}; -use crate::UploadLine; -use crate::server::errors::{AppError, AppResult}; use bytes::{Buf, Bytes}; use clap::ValueEnum; use dialoguer::Input; diff --git a/crates/stream-gears/src/lib.rs b/crates/stream-gears/src/lib.rs index 8dd8417..89cda97 100644 --- a/crates/stream-gears/src/lib.rs +++ b/crates/stream-gears/src/lib.rs @@ -16,6 +16,7 @@ use biliup::credential::Credential; use biliup::downloader::extractor::CallbackFn; use biliup::downloader::util::{LifecycleFile, Segmentable}; use biliup::downloader::{hls, httpflv}; +use clap::Parser; use pyo3::types::{PyList, PyType}; use std::collections::HashMap; use std::env::args; @@ -23,7 +24,6 @@ use std::path::PathBuf; use std::str::FromStr; use std::sync::{Mutex, Once}; use std::time::Duration; -use clap::Parser; use tracing::{debug, error, info}; use tracing_subscriber::EnvFilter; @@ -463,7 +463,6 @@ fn upload( #[pyfunction] pub fn main_loop(py: Python<'_>) -> PyResult<()> { - // 获取 Python 的 sys.argv let sys = py.import("sys")?; let bound = sys.getattr("argv")?; @@ -481,13 +480,17 @@ pub fn main_loop(py: Python<'_>) -> PyResult<()> { &[] => todo!(), &[_] => { args.push("server".to_string()); - }, - [_, command, ..] => if command == "start" { - args[1] = "server".to_string(); + } + [_, command, ..] => { + if command == "start" { + args[1] = "server".to_string(); + } } } - py.detach(|| server::_main(args.as_slice()).map_err(|e| PyRuntimeError::new_err(format!("{e:?}")))) + py.detach(|| { + server::_main(args.as_slice()).map_err(|e| PyRuntimeError::new_err(format!("{e:?}"))) + }) // for item in plugin_list.iter() { // // 每个元素都是一个类(type) diff --git a/crates/stream-gears/src/server.rs b/crates/stream-gears/src/server.rs index 12366b0..aca5b77 100644 --- a/crates/stream-gears/src/server.rs +++ b/crates/stream-gears/src/server.rs @@ -1,6 +1,9 @@ use crate::{DanmakuClient, construct_headers, download}; use async_trait::async_trait; use biliup::downloader::util::Segmentable; +use biliup::uploader::util::SubmitOption; +use biliup_cli::cli::{Cli, Commands}; +use biliup_cli::downloader::generate_json; use biliup_cli::server::app::ApplicationController; use biliup_cli::server::common::util::{Recorder, media_ext_from_url, parse_time}; use biliup_cli::server::config::{Config, default_segment_time}; @@ -16,7 +19,9 @@ use biliup_cli::server::infrastructure::models::StreamerInfo; use biliup_cli::server::infrastructure::repositories; use biliup_cli::server::infrastructure::repositories::get_upload_config; use biliup_cli::server::infrastructure::service_register::ServiceRegister; +use biliup_cli::uploader::{append, list, login, renew, show, upload_by_command, upload_by_config}; use chrono::Utc; +use clap::Parser; use error_stack::{Report, ResultExt}; use fancy_regex::Regex; use pyo3::prelude::PyDictMethods; @@ -31,12 +36,7 @@ use std::net::ToSocketAddrs; use std::ops::Deref; use std::path::PathBuf; use std::sync::{Arc, LazyLock, RwLock}; -use clap::Parser; use tracing::{debug, info}; -use biliup::uploader::util::SubmitOption; -use biliup_cli::cli::{Cli, Commands}; -use biliup_cli::downloader::generate_json; -use biliup_cli::uploader::{append, list, login, renew, show, upload_by_command, upload_by_config}; #[derive(Debug)] pub struct PyPlugin { @@ -85,7 +85,9 @@ impl DownloadPlugin for PyPlugin { Arc::new(DanmakuClient::new(Arc::new(danmaku))) as Arc; ctx.extension.insert(danmaku); } - Ok(StreamStatus::Live { stream_info: Box::new(info) }) + Ok(StreamStatus::Live { + stream_info: Box::new(info), + }) } None => Ok(StreamStatus::Offline), } @@ -353,9 +355,8 @@ fn cfg_arc() -> &'static Arc> { #[tokio::main] pub(crate) async fn _main(args: &[String]) -> AppResult<()> { - let cli = Cli::try_parse_from(args).change_context(AppError::Unknown)?; - + match cli.command { Commands::Login => login(cli.user_cookie, cli.proxy.as_deref()).await?, Commands::Renew => { @@ -378,7 +379,7 @@ pub(crate) async fn _main(args: &[String]) -> AppResult<()> { submit.unwrap_or(SubmitOption::App), cli.proxy.as_deref(), ) - .await? + .await? } Commands::Upload { video_path: _, @@ -405,7 +406,7 @@ pub(crate) async fn _main(args: &[String]) -> AppResult<()> { submit.unwrap_or(SubmitOption::App), cli.proxy.as_deref(), ) - .await? + .await? } Commands::Show { vid } => show(cli.user_cookie, vid, cli.proxy.as_deref()).await?, Commands::DumpFlv { file_name } => generate_json(file_name)?, @@ -417,8 +418,8 @@ pub(crate) async fn _main(args: &[String]) -> AppResult<()> { } => biliup_cli::downloader::download(&url, output, split_size, split_time).await?, Commands::Server { bind, port, auth } => { info!( - "environment loaded and configuration parsed, initializing Postgres connection and running migrations..." - ); + "environment loaded and configuration parsed, initializing Postgres connection and running migrations..." + ); let conn_pool = ConnectionManager::new_pool("data/data.sqlite3") .await .expect("could not initialize the database connection pool"); @@ -430,14 +431,16 @@ pub(crate) async fn _main(args: &[String]) -> AppResult<()> { )); let vec = from_py(actor_handle.clone()).unwrap(); - let service_register = ServiceRegister::new(conn_pool, CONFIG.clone(), actor_handle, vec); + let service_register = + ServiceRegister::new(conn_pool, CONFIG.clone(), actor_handle, vec); let all_streamer = repositories::get_all_streamer(&service_register.pool).await?; for streamer in all_streamer { // workers.push(Arc::new(Worker::new(streamer.id, service_register.pool.clone()))); if let Some(manager) = service_register.get_manager(&streamer.url) { - let upload_config = get_upload_config(&service_register.pool, streamer.id).await?; + let upload_config = + get_upload_config(&service_register.pool, streamer.id).await?; let _ = service_register .add_room(manager, streamer, upload_config) .await?; @@ -455,7 +458,7 @@ pub(crate) async fn _main(args: &[String]) -> AppResult<()> { .await .attach("could not initialize application routes")?; // biliup_cli::run((&bind, port)).await? - }, + } Commands::List { is_pubing, pubed, @@ -472,7 +475,7 @@ pub(crate) async fn _main(args: &[String]) -> AppResult<()> { from_page, max_pages, ) - .await? + .await? } };