Refactor imports and formatting across modules

This commit cleans up and reorders imports for consistency, improves formatting (especially for multi-line expressions and function signatures), and fixes minor style issues in several files. It also updates the server command to pass the 'auth' parameter and makes minor logic clarifications in the stream-gears Python entrypoint.
This commit is contained in:
ForgQi committed 2025-10-24 03:21:44 +08:00
1 parent b829731ad9
commit 9e155493f0
12 files changed
+61 -46

No files matched your search

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