From 50845dc6cd6a838fc0932a9a37d85183ecd81833 Mon Sep 17 00:00:00 2001 From: ForgQi <34411314+ForgQi@users.noreply.github.com> Date: Fri, 19 Sep 2025 23:19:17 +0800 Subject: [PATCH] Update .gitignore and refactor douyin.py imports Added 'logs' to .gitignore to exclude log files from version control. Cleaned up and reorganized imports in douyin.py for better readability and maintainability. Refactor and clean up imports, logic, and pattern usage This commit removes unused imports, simplifies logic with let-else and pattern matching, and updates code style for conciseness and clarity across multiple modules. It also adds Default implementations for handler structs, improves error handling, and streamlines repository and service registration code. --- .gitignore | 3 +- biliup/Danmaku/douyin.py | 11 +++---- crates/biliup-cli/src/downloader.rs | 6 ++-- crates/biliup-cli/src/server/api/endpoints.rs | 6 ++-- .../biliup-cli/src/server/config_manager.rs | 15 ++++----- .../src/server/core/download_actor.rs | 13 +++----- .../biliup-cli/src/server/core/main_loop.rs | 10 ++---- crates/biliup-cli/src/server/event_manager.rs | 1 - crates/biliup-cli/src/server/handlers.rs | 14 ++++++-- .../server/infrastructure/connection_pool.rs | 4 --- .../repositories/live_streamers_repository.rs | 6 ---- .../upload_streamers_repository.rs | 4 --- .../repositories/users_repository.rs | 4 --- .../server/infrastructure/service_register.rs | 10 ++---- crates/biliup-cli/src/server/integration.rs | 4 +-- crates/biliup/src/downloader.rs | 10 ------ .../biliup/src/downloader/extractor/douyu.rs | 5 ++- crates/biliup/src/downloader/httpflv.rs | 6 ++-- crates/stream-gears/src/server.rs | 32 ++++--------------- crates/stream-gears/src/server/api/auth.rs | 9 +++--- .../src/server/api/bilibili_endpoints.rs | 6 ++-- .../stream-gears/src/server/api/endpoints.rs | 11 +++---- crates/stream-gears/src/server/app.rs | 12 +------ crates/stream-gears/src/server/config.rs | 6 ++-- .../src/server/core/download_manager.rs | 26 ++++++--------- .../src/server/core/downloader.rs | 1 - .../core/downloader/ffmpeg_downloader.rs | 8 ++--- .../server/core/downloader/stream_gears.rs | 6 ++-- .../stream-gears/src/server/core/monitor.rs | 7 +--- crates/stream-gears/src/server/core/plugin.rs | 8 ++--- crates/stream-gears/src/server/errors.rs | 3 +- .../src/server/infrastructure/context.rs | 6 ++-- .../src/server/infrastructure/repositories.rs | 10 +++--- .../server/infrastructure/service_register.rs | 18 ++--------- .../src/server/infrastructure/users.rs | 5 ++- crates/stream-gears/src/server/router.rs | 5 +-- crates/stream-gears/src/server/util.rs | 12 +++---- 37 files changed, 106 insertions(+), 217 deletions(-) diff --git a/.gitignore b/.gitignore index cbd9897..0775881 100644 --- a/.gitignore +++ b/.gitignore @@ -50,4 +50,5 @@ _internal pnpm-lock.yaml cookies.json .augment/ -.cunzhi-memory/ \ No newline at end of file +.cunzhi-memory/ +logs \ No newline at end of file diff --git a/biliup/Danmaku/douyin.py b/biliup/Danmaku/douyin.py index 53918f6..229e68d 100644 --- a/biliup/Danmaku/douyin.py +++ b/biliup/Danmaku/douyin.py @@ -3,19 +3,18 @@ # 2024.06.22: 添加来自 https://github.com/hua0512/stream-rec 修改后的 webmssdk.js，以计算 signature import gzip +import logging import aiohttp -import json -from urllib.parse import unquote -from biliup.config import config -from .douyin_util.dy_pb2 import ChatMessage, PushFrame, Response -from biliup.plugins import match1 from google.protobuf import json_format -import logging + +from .douyin_util.dy_pb2 import ChatMessage, PushFrame, Response logger = logging.getLogger('biliup') class Douyin: + from biliup.config import config + headers = { # 'user-agent': random_user_agent(), 'Referer': 'https://live.douyin.com/', diff --git a/crates/biliup-cli/src/downloader.rs b/crates/biliup-cli/src/downloader.rs index 5a6e16c..a5935fa 100644 --- a/crates/biliup-cli/src/downloader.rs +++ b/crates/biliup-cli/src/downloader.rs @@ -137,11 +137,11 @@ pub fn generate_json(mut file_name: PathBuf) -> Result<()> { script_tag_count += 1; let (_, tag_data) = script_data(i).unwrap(); - let flv_tag = FlvTag { + + FlvTag { header: tag_header, data: TagDataHeader::Script(tag_data), - }; - flv_tag + } } }; flv_writer::to_json(&mut writer, &flv_tag)?; diff --git a/crates/biliup-cli/src/server/api/endpoints.rs b/crates/biliup-cli/src/server/api/endpoints.rs index f905764..841590e 100644 --- a/crates/biliup-cli/src/server/api/endpoints.rs +++ b/crates/biliup-cli/src/server/api/endpoints.rs @@ -1,20 +1,18 @@ use crate::server::core::download_actor::DownloadActorHandle; use crate::server::core::live_streamers::{ - AddLiveStreamerDto, DynLiveStreamersRepository, DynLiveStreamersService, LiveStreamerDto, + AddLiveStreamerDto, DynLiveStreamersService, LiveStreamerDto, LiveStreamerEntity, }; use crate::server::core::upload_streamers::{DynUploadStreamersRepository, StudioEntity}; -use crate::server::core::users::{DynUsersRepository, User}; +use crate::server::core::users::User; use crate::server::errors::{AppError, AppResult}; use crate::server::infrastructure::connection_pool::ConnectionPool; use crate::server::infrastructure::repositories::models::{ Configuration, FileItem, LiveStreamer, StreamerInfo, UploadStreamer, }; -use crate::server::infrastructure::service_register::ServiceRegister; use axum::extract::{Path, State}; use axum::{Extension, Json}; -use biliup::credential::login_by_cookies; use ormlite::Model; use serde_json::json; diff --git a/crates/biliup-cli/src/server/config_manager.rs b/crates/biliup-cli/src/server/config_manager.rs index 4da8ced..5156929 100644 --- a/crates/biliup-cli/src/server/config_manager.rs +++ b/crates/biliup-cli/src/server/config_manager.rs @@ -330,7 +330,7 @@ impl ConfigManager { .streamers .as_ref() .map(|streamers| streamers.keys().cloned().collect()) - .unwrap_or_else(Vec::new) + .unwrap_or_default() } /// 获取配置值（通用方法） @@ -348,8 +348,8 @@ impl ConfigManager { 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") { - if let Some(cookies) = cookie_info.get("cookies").and_then(|c| c.as_array()) { + 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)) = ( @@ -370,15 +370,12 @@ impl ConfigManager { user.cookies = Some(cookie_map); } } - } - if let Some(token_info) = cookies_data.get("token_info") { - if let Some(access_token) = token_info.get("access_token").and_then(|t| t.as_str()) { - if let Some(ref mut user) = self.config.user { + 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(()) } diff --git a/crates/biliup-cli/src/server/core/download_actor.rs b/crates/biliup-cli/src/server/core/download_actor.rs index 9b88267..773e828 100644 --- a/crates/biliup-cli/src/server/core/download_actor.rs +++ b/crates/biliup-cli/src/server/core/download_actor.rs @@ -1,11 +1,9 @@ use biliup::client::StatelessClient; use crate::server::core::StreamStatus; -use crate::server::core::live_streamers::{DynLiveStreamersService, LiveStreamerDto}; -use crate::server::core::upload_actor::UploadActorHandle; -use crate::server::core::util::{AnyMap, Cycle, logging_spawn}; +use crate::server::core::live_streamers::LiveStreamerDto; +use crate::server::core::util::{AnyMap, Cycle}; use biliup::downloader::extractor::{SiteDefinition, find_extractor}; -use biliup::downloader::util::Segmentable; use indexmap::indexmap; @@ -15,8 +13,7 @@ use std::ops::DerefMut; use std::sync::{Arc, RwLock}; use std::time::Duration; use tokio::task::JoinHandle; -use tracing::log::info; -use tracing::{debug, error}; +use tracing::debug; async fn start_monitor( task: Cycle, @@ -27,7 +24,7 @@ async fn start_monitor( let n = &mut 0; loop { let (url, status) = task.get(n); - if status != StreamStatus::Working {} + status != StreamStatus::Working; match status { StreamStatus::Working => {} @@ -39,7 +36,7 @@ async fn start_monitor( } match (extractor.get_site(&url, client.clone()).await, status) { - (Ok(mut site), StreamStatus::Idle | StreamStatus::Inspecting) => { + (Ok(site), StreamStatus::Idle | StreamStatus::Inspecting) => { println!("Idle\n {url} \n{site}"); // let (filename, split_size, split_time) = if let Ok(LiveStreamerDto { // filename, diff --git a/crates/biliup-cli/src/server/core/main_loop.rs b/crates/biliup-cli/src/server/core/main_loop.rs index 183922b..6f656a9 100644 --- a/crates/biliup-cli/src/server/core/main_loop.rs +++ b/crates/biliup-cli/src/server/core/main_loop.rs @@ -1,17 +1,11 @@ use crate::server::core::StreamStatus; -use crate::server::core::live_streamers::{DynLiveStreamersService, LiveStreamerDto, Videos}; -use crate::server::core::upload_actor::UploadActorHandle; -use crate::server::core::util::{Cycle, logging_spawn}; +use crate::server::core::live_streamers::Videos; +use crate::server::core::util::Cycle; use anyhow::Result; -use biliup::client::StatelessClient; use biliup::downloader::extractor::Site; -use biliup::downloader::util::Segmentable; -use std::error::Error; -use std::time::Duration; use tokio::sync::mpsc::Sender; use tokio::sync::mpsc::{Receiver, channel}; use tokio::task::JoinHandle; -use tracing::{debug, error, info}; /// This struct is used by client actors to send messages to the main loop. The /// message type is `ToServer`. diff --git a/crates/biliup-cli/src/server/event_manager.rs b/crates/biliup-cli/src/server/event_manager.rs index a517a87..05a252f 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}; -use crate::server::core::live_streamers::LiveStreamerDto; /// 事件类型枚举，对应Python版本的事件 #[derive(Debug, Clone, PartialEq, Eq, Hash, Serialize, Deserialize)] diff --git a/crates/biliup-cli/src/server/handlers.rs b/crates/biliup-cli/src/server/handlers.rs index b313e80..8ba0ffc 100644 --- a/crates/biliup-cli/src/server/handlers.rs +++ b/crates/biliup-cli/src/server/handlers.rs @@ -1,8 +1,6 @@ use anyhow::Result; use async_trait::async_trait; use chrono::Utc; -use std::collections::HashMap; -use std::path::Path; use tokio::fs; use tracing::{debug, error, info, warn}; @@ -132,6 +130,12 @@ impl EventHandler for DownloadHandler { /// 下载完成处理器，对应Python的processed pub struct DownloadedHandler {} +impl Default for DownloadedHandler { + fn default() -> Self { + Self::new() + } +} + impl DownloadedHandler { pub fn new() -> Self { Self {} @@ -264,6 +268,12 @@ impl EventHandler for UploadHandler { /// 上传完成处理器，对应Python的uploaded函数 pub struct UploadedHandler {} +impl Default for UploadedHandler { + fn default() -> Self { + Self::new() + } +} + impl UploadedHandler { pub fn new() -> Self { Self {} diff --git a/crates/biliup-cli/src/server/infrastructure/connection_pool.rs b/crates/biliup-cli/src/server/infrastructure/connection_pool.rs index e1882bb..1d6e80e 100644 --- a/crates/biliup-cli/src/server/infrastructure/connection_pool.rs +++ b/crates/biliup-cli/src/server/infrastructure/connection_pool.rs @@ -1,9 +1,5 @@ -use crate::server::infrastructure::repositories::models::{ - Configuration, FileItem, LiveStreamer, StreamerInfo, UploadStreamer, -}; use anyhow::{Context, Result}; use ormlite::model::*; -use ormlite::sqlite::SqliteConnection; use sqlx::sqlite::SqlitePoolOptions; use sqlx::{Pool, Sqlite}; use tracing::info; diff --git a/crates/biliup-cli/src/server/infrastructure/repositories/live_streamers_repository.rs b/crates/biliup-cli/src/server/infrastructure/repositories/live_streamers_repository.rs index c4d0f17..58d8ef9 100644 --- a/crates/biliup-cli/src/server/infrastructure/repositories/live_streamers_repository.rs +++ b/crates/biliup-cli/src/server/infrastructure/repositories/live_streamers_repository.rs @@ -1,10 +1,4 @@ -use crate::server::core::live_streamers::{ - AddLiveStreamerDto, LiveStreamerEntity, LiveStreamersRepository, -}; use crate::server::infrastructure::connection_pool::ConnectionPool; -use anyhow::Context; -use async_trait::async_trait; -use sqlx::query_as; #[derive(Clone)] pub struct SqliteLiveStreamersRepository { 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 8d0f0a6..1c22576 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,9 +1,5 @@ -use crate::server::core::upload_streamers::{StudioEntity, UploadStreamersRepository}; use crate::server::infrastructure::connection_pool::ConnectionPool; -use anyhow::Context; -use async_trait::async_trait; -use sqlx::query_as; #[derive(Clone)] pub struct SqliteUploadStreamersRepository { diff --git a/crates/biliup-cli/src/server/infrastructure/repositories/users_repository.rs b/crates/biliup-cli/src/server/infrastructure/repositories/users_repository.rs index 8375e6f..052ee77 100644 --- a/crates/biliup-cli/src/server/infrastructure/repositories/users_repository.rs +++ b/crates/biliup-cli/src/server/infrastructure/repositories/users_repository.rs @@ -1,8 +1,4 @@ -use crate::server::core::users::{User, UsersRepository}; use crate::server::infrastructure::connection_pool::ConnectionPool; -use anyhow::Context; -use async_trait::async_trait; -use sqlx::{query, query_as}; #[derive(Clone)] pub struct SqliteUsersStreamersRepository { diff --git a/crates/biliup-cli/src/server/infrastructure/service_register.rs b/crates/biliup-cli/src/server/infrastructure/service_register.rs index fa390db..555e0f4 100644 --- a/crates/biliup-cli/src/server/infrastructure/service_register.rs +++ b/crates/biliup-cli/src/server/infrastructure/service_register.rs @@ -1,17 +1,11 @@ use crate::server::core::live_streamers::{ - DynDownloadRecordsRepository, DynLiveStreamersRepository, DynLiveStreamersService, + DynDownloadRecordsRepository, DynVideosRepository, }; -use crate::server::core::upload_streamers::{ - DynUploadRecordsRepository, DynUploadStreamersRepository, -}; -use crate::server::core::users::DynUsersRepository; +use crate::server::core::upload_streamers::DynUploadRecordsRepository; use crate::server::infrastructure::connection_pool::ConnectionPool; use crate::server::infrastructure::repositories::download_records_repository::SqliteDownloadRecordsRepository; -use crate::server::infrastructure::repositories::live_streamers_repository::SqliteLiveStreamersRepository; use crate::server::infrastructure::repositories::upload_records_repository::SqliteUploadRecordsRepository; -use crate::server::infrastructure::repositories::upload_streamers_repository::SqliteUploadStreamersRepository; -use crate::server::infrastructure::repositories::users_repository::SqliteUsersStreamersRepository; use crate::server::infrastructure::repositories::videos_repository::SqliteVideosRepository; use axum::extract::FromRef; use std::sync::Arc; diff --git a/crates/biliup-cli/src/server/integration.rs b/crates/biliup-cli/src/server/integration.rs index ad82bfd..cf8b1ac 100644 --- a/crates/biliup-cli/src/server/integration.rs +++ b/crates/biliup-cli/src/server/integration.rs @@ -3,9 +3,7 @@ use tokio::task::JoinHandle; use tracing::{error, info}; use crate::server::{ - config_manager::ConfigManager, core::live_streamers::DynLiveStreamersService, - event_manager::EventManager, handlers::HandlerRegistry, - infrastructure::connection_pool::ConnectionManager, plugin_registry::initialize_plugins, + core::live_streamers::DynLiveStreamersService, plugin_registry::initialize_plugins, }; /// 集成管理器，负责协调所有组件 diff --git a/crates/biliup/src/downloader.rs b/crates/biliup/src/downloader.rs index 3a5ee01..6f8a2b2 100644 --- a/crates/biliup/src/downloader.rs +++ b/crates/biliup/src/downloader.rs @@ -1,15 +1,5 @@ -use crate::downloader::httpflv::Connection; -use flv_parser::header; -use nom::Err; -use reqwest::header::{HeaderMap, HeaderName, HeaderValue}; -use std::collections::HashMap; -use tracing::{debug, error, info}; -use crate::downloader::util::{LifecycleFile, Segmentable}; -use crate::client::StatelessClient; -use crate::downloader::extractor::CallbackFn; -use std::str::FromStr; pub mod error; pub mod extractor; diff --git a/crates/biliup/src/downloader/extractor/douyu.rs b/crates/biliup/src/downloader/extractor/douyu.rs index 50b9c8b..f89975a 100644 --- a/crates/biliup/src/downloader/extractor/douyu.rs +++ b/crates/biliup/src/downloader/extractor/douyu.rs @@ -72,8 +72,8 @@ impl SiteDefinition for DouyuLive { .await? .json() .await?; - if result["error"] == 0 { - if let Some(key) = regex::Regex::new(r"(\d{1,8}[0-9a-zA-Z]+)_?\d{0,4}(/playlist|.m3u8)") + if result["error"] == 0 + && 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()) { @@ -90,7 +90,6 @@ impl SiteDefinition for DouyuLive { client, }); } - } Err(Error::Custom(result.to_string())) } diff --git a/crates/biliup/src/downloader/httpflv.rs b/crates/biliup/src/downloader/httpflv.rs index 5b32c5e..fc61c04 100644 --- a/crates/biliup/src/downloader/httpflv.rs +++ b/crates/biliup/src/downloader/httpflv.rs @@ -129,11 +129,11 @@ pub(crate) async fn parse_flv( } on_meta_data = Some((tag_header, bytes.clone(), previous_tag_size.clone())); - let flv_tag = FlvTag { + + FlvTag { header: tag_header, data: TagDataHeader::Script(tag_data), - }; - flv_tag + } } }; match &flv_tag { diff --git a/crates/stream-gears/src/server.rs b/crates/stream-gears/src/server.rs index 1bf18ec..5556ae8 100644 --- a/crates/stream-gears/src/server.rs +++ b/crates/stream-gears/src/server.rs @@ -7,43 +7,23 @@ pub mod infrastructure; mod router; mod util; -use anyhow::Context; -use async_channel::{Receiver, SendError, Sender, bounded}; -use async_trait::async_trait; -use axum::http::StatusCode; -use axum::response::IntoResponse; -use error_stack::{Report, ResultExt}; -use fancy_regex::Regex; -use futures::{FutureExt, TryFutureExt}; +use error_stack::ResultExt; +use futures::TryFutureExt; use pyo3::exceptions::PyRuntimeError; -use pyo3::prelude::{PyAnyMethods, PyListMethods, PyModule, PyTypeMethods}; -use pyo3::types::{PyList, PyType}; -use pyo3::{Bound, Py, PyAny, PyObject, PyResult, Python, pyfunction}; -use serde::{Deserialize, Serialize}; -use std::collections::HashMap; +use pyo3::prelude::PyListMethods; +use pyo3::{PyResult, Python, pyfunction}; use std::net::ToSocketAddrs; -use std::path::PathBuf; -use std::sync::{Arc, Mutex, RwLock}; -use std::time::Duration; -use thiserror::Error; -use tokio::signal; -use tokio::sync::oneshot; -use tokio::task::{AbortHandle, JoinHandle}; // use tokio::sync::mpsc::Receiver; use crate::server::app::ApplicationController; -use crate::server::config::{Config, StreamerConfig}; use crate::server::errors::{AppError, AppResult}; use crate::server::infrastructure::connection_pool::ConnectionManager; -use crate::server::infrastructure::context::Worker; use crate::server::infrastructure::repositories; use crate::server::infrastructure::service_register::ServiceRegister; -use biliup::downloader::httpflv::download; -use core::download_manager::DownloadManager; -use tracing::{error, info}; +use tracing::info; #[pyfunction] pub fn main_loop(py: Python<'_>) -> PyResult<()> { - py.detach(|| _main()) + py.detach(_main) .map_err(|e| PyRuntimeError::new_err(format!("{e:?}"))) // for item in plugin_list.iter() { diff --git a/crates/stream-gears/src/server/api/auth.rs b/crates/stream-gears/src/server/api/auth.rs index a5f0403..1e543f2 100644 --- a/crates/stream-gears/src/server/api/auth.rs +++ b/crates/stream-gears/src/server/api/auth.rs @@ -1,8 +1,7 @@ use axum::{ - Form, Router, - extract::Query, + Router, http::StatusCode, - response::{Html, IntoResponse, Redirect}, + response::{IntoResponse, Redirect}, routing::{get, post}, }; use serde::Deserialize; @@ -93,7 +92,7 @@ mod post { mod get { use super::*; - use crate::server::errors::AppResult; + use tracing::error; // pub async fn login( @@ -119,7 +118,7 @@ mod get { Ok(false) => StatusCode::NOT_FOUND.into_response(), Err(e) => { error!(error = ?e, "Error checking existing user"); - return StatusCode::INTERNAL_SERVER_ERROR.into_response(); + StatusCode::INTERNAL_SERVER_ERROR.into_response() } } } diff --git a/crates/stream-gears/src/server/api/bilibili_endpoints.rs b/crates/stream-gears/src/server/api/bilibili_endpoints.rs index b173883..b5b8d18 100644 --- a/crates/stream-gears/src/server/api/bilibili_endpoints.rs +++ b/crates/stream-gears/src/server/api/bilibili_endpoints.rs @@ -1,4 +1,4 @@ -use crate::server::errors::{AppError, AppResult, report_to_response}; +use crate::server::errors::{AppError, report_to_response}; use crate::server::infrastructure::connection_pool::ConnectionPool; use crate::server::infrastructure::models::Configuration; use axum::extract::{Query, State}; @@ -58,7 +58,7 @@ pub async fn get_proxy_endpoint( ) -> Result { info!(params = ¶ms["url"]); // let bili = login_by_cookies(¶ms["user"]).await?; - Ok(client + client .client .get(¶ms["url"]) .send() @@ -68,5 +68,5 @@ pub async fn get_proxy_endpoint( .bytes() .await .change_context(AppError::Unknown) - .map_err(report_to_response)?) + .map_err(report_to_response) } diff --git a/crates/stream-gears/src/server/api/endpoints.rs b/crates/stream-gears/src/server/api/endpoints.rs index 7b3e489..9269b83 100644 --- a/crates/stream-gears/src/server/api/endpoints.rs +++ b/crates/stream-gears/src/server/api/endpoints.rs @@ -1,4 +1,4 @@ -use crate::server::errors::{AppError, AppResult, report_to_response}; +use crate::server::errors::{AppError, report_to_response}; use std::sync::{Arc, RwLock}; use crate::server::config::Config; @@ -9,14 +9,13 @@ use crate::server::infrastructure::models::{ Configuration, FileItem, InsertLiveStreamer, InsertUploadStreamer, LiveStreamer, StreamerInfo, UploadStreamer, }; -use crate::server::infrastructure::repositories::{del_streamer, get_all_streamer, get_config}; +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 error_stack::{ResultExt, bail}; -use ormlite::query_builder::OnConflict; +use error_stack::ResultExt; use ormlite::{Insert, Model}; use serde_json::json; use tracing::info; @@ -40,7 +39,7 @@ pub async fn get_streamers_endpoint( .iter() .find_map(|worker| { if worker.id == x.id { - Some(worker.downloader_status.read().unwrap().clone()) + Some(*worker.downloader_status.read().unwrap()) } else { None } @@ -114,7 +113,7 @@ pub async fn delete_streamers_endpoint( State(workers): State>>>>, Path(id): Path, ) -> Result, Response> { - let _ = service_register + service_register .del_room(id) .await .map_err(report_to_response)?; diff --git a/crates/stream-gears/src/server/app.rs b/crates/stream-gears/src/server/app.rs index 8bf25b5..dd14ae4 100644 --- a/crates/stream-gears/src/server/app.rs +++ b/crates/stream-gears/src/server/app.rs @@ -1,19 +1,9 @@ use biliup::client::StatelessClient; -use anyhow::Context; -use axum::routing::{delete, get, post, put}; -use axum::{Extension, Router, http}; +use axum::{Extension, http}; use crate::server; use crate::server::api::auth; -use crate::server::api::bilibili_endpoints::{ - archive_pre_endpoint, get_myinfo_endpoint, get_proxy_endpoint, -}; -use crate::server::api::endpoints::{ - delete_streamers_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::spa::static_handler; use crate::server::errors::{AppError, AppResult}; use crate::server::infrastructure::service_register::ServiceRegister; diff --git a/crates/stream-gears/src/server/config.rs b/crates/stream-gears/src/server/config.rs index 26bd1cf..436af3f 100644 --- a/crates/stream-gears/src/server/config.rs +++ b/crates/stream-gears/src/server/config.rs @@ -3,13 +3,13 @@ use crate::server::infrastructure::models::HookStep; use anyhow::{Context, bail}; use pyo3::prelude::PyDictMethods; use pyo3::sync::OnceLockExt; -use pyo3::types::{PyAnyMethods, PyDict, PyMapping, PyString}; -use pyo3::{Bound, FromPyObject, Py, PyAny, PyResult, Python, pyclass, pyfunction, pymethods}; +use pyo3::types::{PyAnyMethods, PyDict}; +use pyo3::{Bound, FromPyObject, PyAny, PyResult, Python, pyclass, pyfunction, pymethods}; use pythonize::pythonize; use serde::{Deserialize, Serialize}; use std::ops::Deref; use std::sync::{Arc, OnceLock, RwLock}; -use std::{collections::HashMap, fs, path::Path, path::PathBuf}; +use std::{collections::HashMap, path::Path, path::PathBuf}; #[derive(bon::Builder, Debug, Clone, Serialize, Deserialize)] pub struct Config { diff --git a/crates/stream-gears/src/server/core/download_manager.rs b/crates/stream-gears/src/server/core/download_manager.rs index 4174274..2c9e592 100644 --- a/crates/stream-gears/src/server/core/download_manager.rs +++ b/crates/stream-gears/src/server/core/download_manager.rs @@ -2,15 +2,11 @@ use crate::server::core::downloader::SegmentEvent; use crate::server::core::monitor::{Monitor, RoomsHandle}; use crate::server::core::plugin::{DownloadPlugin, StreamInfo}; use crate::server::errors::{AppError, AppResult}; -use crate::server::infrastructure::connection_pool::ConnectionPool; use crate::server::infrastructure::context::{Worker, WorkerStatus}; -use crate::uploader::UploadLine; -use anyhow::Context; use async_channel::{Receiver, Sender, bounded}; -use biliup::bilibili::{BiliBili, Credit, Studio, Video}; +use biliup::bilibili::{BiliBili, Studio, Video}; use biliup::client::StatelessClient; use biliup::credential::login_by_cookies; -use biliup::error::Kind; use biliup::uploader::line::{Line, Probe}; use biliup::uploader::util::SubmitOption; use biliup::uploader::{VideoFile, line}; @@ -112,7 +108,7 @@ impl UActor { async fn run(&mut self) { while let Ok(msg) = self.receiver.recv().await { - self.handle_message(msg); + self.handle_message(msg).await; } } @@ -148,7 +144,7 @@ impl UActor { config.threads as usize, &client, &line, - &*event.file_path, + &event.file_path, ) .await { @@ -160,7 +156,7 @@ impl UActor { config.threads as usize, &client, &line, - &*event.file_path, + &event.file_path, ) .await { @@ -202,13 +198,11 @@ impl UActor { ) .build(); - if !studio.cover.is_empty() { - if let Ok(c) = &std::fs::read(&studio.cover).map_err(|e| error!(e=?e)) { - if let Ok(url) = bilibili.cover_up(c).await.map_err(|e| error!(e=?e)) { + 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 submit = match config.submit_api { Some(submit) => SubmitOption::from_str(&submit).unwrap_or(SubmitOption::App), @@ -241,11 +235,11 @@ impl UActor { .to_str() ); info!("{line:?}"); - let video_file = VideoFile::new(&video_path).change_context(AppError::Unknown)?; + let video_file = VideoFile::new(video_path).change_context(AppError::Unknown)?; let total_size = video_file.total_size; let file_name = video_file.file_name.clone(); let uploader = line - .pre_upload(&bilibili, video_file) + .pre_upload(bilibili, video_file) .await .change_context(AppError::Unknown)?; diff --git a/crates/stream-gears/src/server/core/downloader.rs b/crates/stream-gears/src/server/core/downloader.rs index 72d8db6..83f7842 100644 --- a/crates/stream-gears/src/server/core/downloader.rs +++ b/crates/stream-gears/src/server/core/downloader.rs @@ -5,7 +5,6 @@ use crate::server::core::download_manager::UploaderMessage; use crate::server::infrastructure::context::Worker; use async_channel::Sender; use async_trait::async_trait; -use ffmpeg_downloader::FfmpegDownloader; use serde::{Deserialize, Serialize}; use std::collections::HashMap; use std::path::PathBuf; diff --git a/crates/stream-gears/src/server/core/downloader/ffmpeg_downloader.rs b/crates/stream-gears/src/server/core/downloader/ffmpeg_downloader.rs index 6277548..7d9ae68 100644 --- a/crates/stream-gears/src/server/core/downloader/ffmpeg_downloader.rs +++ b/crates/stream-gears/src/server/core/downloader/ffmpeg_downloader.rs @@ -4,7 +4,7 @@ use crate::server::core::downloader::{ DownloadConfig, DownloadStatus, Downloader, DownloaderType, SegmentEvent, }; use crate::server::infrastructure::context::Worker; -use async_channel::{SendError, Sender, bounded}; +use async_channel::{Sender, bounded}; use async_trait::async_trait; use std::path::PathBuf; use std::process::Stdio; @@ -181,7 +181,7 @@ impl FfmpegDownloader { .stderr(Stdio::piped()) .kill_on_drop(true); - let mut child = cmd.spawn()?; + let child = cmd.spawn()?; // 保存进程句柄 { @@ -199,9 +199,9 @@ impl FfmpegDownloader { } }; // let (tx, rx) = bounded(16); - let mut i = 0; + let i = 0; // 分段回调 - let mut segment_callback = |event: SegmentEvent| { + let segment_callback = |event: SegmentEvent| { println!("New segment: {:?}", event.file_path); // if i == 0 { // match sender.force_send(UploaderMessage::SegmentEvent(event)) { diff --git a/crates/stream-gears/src/server/core/downloader/stream_gears.rs b/crates/stream-gears/src/server/core/downloader/stream_gears.rs index 28f3c33..25d8901 100644 --- a/crates/stream-gears/src/server/core/downloader/stream_gears.rs +++ b/crates/stream-gears/src/server/core/downloader/stream_gears.rs @@ -1,17 +1,15 @@ use crate::server::core::download_manager::UploaderMessage; use crate::server::core::downloader::{DownloadStatus, Downloader, SegmentEvent}; use crate::server::infrastructure::context::Worker; -use async_channel::{SendError, Sender, bounded}; +use async_channel::{Sender, bounded}; use async_trait::async_trait; use axum::http::HeaderMap; use biliup::client::StatelessClient; -use biliup::downloader::extractor::CallbackFn; use biliup::downloader::flv_parser::header; use biliup::downloader::httpflv::Connection; use biliup::downloader::util::{LifecycleFile, Segmentable}; use biliup::downloader::{hls, httpflv}; use nom::Err; -use std::collections::HashMap; use std::path::PathBuf; use std::sync::{Arc, RwLock}; use std::time::SystemTime; @@ -51,7 +49,7 @@ impl StreamGears { Self { url: url.into(), header_map, - file_name: file_name.into(), + file_name: file_name, segment, proxy, status: Arc::new(RwLock::new(DownloadStatus::Downloading)), diff --git a/crates/stream-gears/src/server/core/monitor.rs b/crates/stream-gears/src/server/core/monitor.rs index 2f0bc56..213fecf 100644 --- a/crates/stream-gears/src/server/core/monitor.rs +++ b/crates/stream-gears/src/server/core/monitor.rs @@ -2,16 +2,11 @@ use crate::server::core::download_manager::{ActorHandle, DownloaderMessage}; use crate::server::core::plugin::{DownloadPlugin, StreamStatus}; use crate::server::infrastructure::context::{Worker, WorkerStatus}; use async_channel::{Receiver, Sender, bounded}; -use pyo3::Py; -use pyo3::types::PyType; -use std::collections::HashMap; -use std::error::Error; -use std::path::PathBuf; use std::sync::Arc; use std::time::Duration; use tokio::sync::oneshot; use tokio::task::JoinHandle; -use tracing::{error, info}; +use tracing::info; async fn start_client( rooms_handle: Arc, diff --git a/crates/stream-gears/src/server/core/plugin.rs b/crates/stream-gears/src/server/core/plugin.rs index ba43cc6..f71ce82 100644 --- a/crates/stream-gears/src/server/core/plugin.rs +++ b/crates/stream-gears/src/server/core/plugin.rs @@ -2,11 +2,10 @@ use crate::construct_headers; 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, SegmentEvent}; +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_channel::{Receiver, Sender, bounded}; use async_trait::async_trait; use biliup::downloader::util::Segmentable; use error_stack::{Report, ResultExt}; @@ -17,11 +16,8 @@ use pyo3::{Bound, Py, PyAny, PyResult, Python}; use serde::{Deserialize, Serialize}; use std::collections::HashMap; use std::path::{Path, PathBuf}; -use std::sync::{Arc, Mutex}; -use std::time::Duration; +use std::sync::Arc; use time::OffsetDateTime; -use tokio::sync::oneshot; -use tokio::task::JoinHandle; use tracing::{debug, info}; // Stream information structures diff --git a/crates/stream-gears/src/server/errors.rs b/crates/stream-gears/src/server/errors.rs index 91065d9..dd2d34c 100644 --- a/crates/stream-gears/src/server/errors.rs +++ b/crates/stream-gears/src/server/errors.rs @@ -4,7 +4,6 @@ use axum::response::{IntoResponse, Response}; use error_stack::Report; use serde::{Deserialize, Serialize}; use std::collections::HashMap; -use std::error::Error; use thiserror::Error; use tracing::log::error; @@ -18,7 +17,7 @@ pub enum AppError { } pub fn report_to_response(report: impl Into>) -> Response { - let mut report = report.into(); + let report = report.into(); let (status, error_message) = match report.downcast_ref::() { Some(AppError::Unknown) => (StatusCode::INTERNAL_SERVER_ERROR, report.to_string()), Some(AppError::Custom(msg)) => (StatusCode::INTERNAL_SERVER_ERROR, msg.into()), diff --git a/crates/stream-gears/src/server/infrastructure/context.rs b/crates/stream-gears/src/server/infrastructure/context.rs index 63c3db0..2c0a08d 100644 --- a/crates/stream-gears/src/server/infrastructure/context.rs +++ b/crates/stream-gears/src/server/infrastructure/context.rs @@ -6,7 +6,7 @@ use crate::server::infrastructure::repositories::{get_config, get_streamer}; use error_stack::ResultExt; use ormlite::Model; use serde::{Deserialize, Serialize}; -use std::sync::{Arc, RwLock}; +use std::sync::RwLock; use tracing::info; pub struct Context { @@ -62,12 +62,12 @@ impl Worker { return Ok(None); }; - Ok(UploadStreamer::select() + UploadStreamer::select() .where_("id = ?") .bind(id) .fetch_optional(&self.pool) .await - .change_context(AppError::Unknown)?) + .change_context(AppError::Unknown) } pub fn change_status(&self, stage: Stage, status: WorkerStatus) { diff --git a/crates/stream-gears/src/server/infrastructure/repositories.rs b/crates/stream-gears/src/server/infrastructure/repositories.rs index f1c9be9..f03bb80 100644 --- a/crates/stream-gears/src/server/infrastructure/repositories.rs +++ b/crates/stream-gears/src/server/infrastructure/repositories.rs @@ -16,7 +16,7 @@ pub async fn get_streamer(pool: &ConnectionPool, id: i64) -> AppResult AppResult { let streamer = get_streamer(pool, id).await?; - let _ = streamer + streamer .clone() .delete(pool) .await @@ -24,10 +24,10 @@ pub async fn del_streamer(pool: &ConnectionPool, id: i64) -> AppResult AppResult> { - Ok(LiveStreamer::select() + LiveStreamer::select() .fetch_all(pool) .await - .change_context(AppError::Unknown)?) + .change_context(AppError::Unknown) } pub async fn get_config(pool: &ConnectionPool) -> AppResult { @@ -56,8 +56,8 @@ pub async fn get_config(pool: &ConnectionPool) -> AppResult { } pub async fn get_all_uploader(pool: &ConnectionPool) -> AppResult> { - Ok(UploadStreamer::select() + UploadStreamer::select() .fetch_all(pool) .await - .change_context(AppError::Unknown)?) + .change_context(AppError::Unknown) } diff --git a/crates/stream-gears/src/server/infrastructure/service_register.rs b/crates/stream-gears/src/server/infrastructure/service_register.rs index 269ebe4..f8a4dfd 100644 --- a/crates/stream-gears/src/server/infrastructure/service_register.rs +++ b/crates/stream-gears/src/server/infrastructure/service_register.rs @@ -3,18 +3,11 @@ use crate::server::config::Config; use crate::server::core::download_manager::{ActorHandle, DownloadManager}; use crate::server::core::monitor::Monitor; use crate::server::core::plugin; -use crate::server::errors::{AppError, AppResult, report_to_response}; +use crate::server::errors::{AppError, AppResult}; use crate::server::infrastructure::connection_pool::ConnectionPool; use crate::server::infrastructure::context::Worker; -use crate::server::infrastructure::models::{ - Configuration, LiveStreamer, StreamerInfo, UploadStreamer, -}; -use crate::server::infrastructure::repositories; -use crate::server::infrastructure::repositories::del_streamer; use axum::extract::FromRef; -use axum::http::StatusCode; -use axum::response::IntoResponse; -use error_stack::{ResultExt, bail}; +use error_stack::bail; use ormlite::Model; use std::sync::{Arc, RwLock}; use tracing::info; @@ -65,12 +58,7 @@ impl ServiceRegister { } pub fn get_manager(&self, url: &str) -> Option<&DownloadManager> { - for manager in self.managers.iter() { - if manager.matches(url) { - return Some(manager); - } - } - None + self.managers.iter().find(|&manager| manager.matches(url)).map(|v| v as _) } pub async fn add_room( diff --git a/crates/stream-gears/src/server/infrastructure/users.rs b/crates/stream-gears/src/server/infrastructure/users.rs index 1563c79..eb65119 100644 --- a/crates/stream-gears/src/server/infrastructure/users.rs +++ b/crates/stream-gears/src/server/infrastructure/users.rs @@ -1,12 +1,11 @@ use crate::server::errors::{AppError, AppResult}; use crate::server::infrastructure::connection_pool::ConnectionPool; -use crate::server::infrastructure::models::Configuration; use axum_login::{AuthUser, AuthnBackend, UserId}; use error_stack::FutureExt; -use error_stack::{Report, ResultExt, bail}; +use error_stack::{ResultExt, bail}; use password_auth::{generate_hash, verify_password}; use serde::{Deserialize, Serialize}; -use sqlx::{FromRow, SqlitePool}; +use sqlx::FromRow; use tokio::task; #[derive(Clone, Serialize, Deserialize, FromRow)] diff --git a/crates/stream-gears/src/server/router.rs b/crates/stream-gears/src/server/router.rs index 8e8fd1f..35308c0 100644 --- a/crates/stream-gears/src/server/router.rs +++ b/crates/stream-gears/src/server/router.rs @@ -7,12 +7,9 @@ use crate::server::api::endpoints::{ get_upload_streamer_endpoint, get_upload_streamers_endpoint, get_users_endpoint, post_streamers_endpoint, put_streamers_endpoint, }; -use crate::server::api::spa::static_handler; use crate::server::infrastructure::service_register::ServiceRegister; -use axum::http::HeaderValue; use axum::routing::{delete, get, post}; -use axum::{Extension, Router, http}; -use tower_http::cors::{AllowMethods, CorsLayer}; +use axum::Router; pub fn router(service_register: ServiceRegister) -> Router<()> { Router::new() diff --git a/crates/stream-gears/src/server/util.rs b/crates/stream-gears/src/server/util.rs index 1b2ddd4..7a0929a 100644 --- a/crates/stream-gears/src/server/util.rs +++ b/crates/stream-gears/src/server/util.rs @@ -49,7 +49,7 @@ impl Recorder { if !self.exists_with_suffix(&base) { return base; } - t = t + Duration::seconds(1); + t += Duration::seconds(1); } } @@ -88,7 +88,7 @@ pub fn media_ext_from_url(input: &str) -> Option { let token = val .trim() .trim_start_matches('.') // .mp4 -> mp4 - .split(|c: char| matches!(c, '/' | ';' | ',' | '?' | '&' | '#')) // video/mp4;codecs=... + .split(['/', ';', ',', '?', '&', '#']) // video/mp4;codecs=... .next() .map(str::trim) .unwrap_or(""); @@ -102,13 +102,11 @@ pub fn media_ext_from_url(input: &str) -> Option { // 1) 先尝试按 URL 解析 if let Ok(url) = Url::parse(input) { // a) 从最后一个 path segment 取扩展名 - if let Some(seg) = url.path_segments().and_then(|s| s.last()) { - if let Some((_, ext)) = seg.rsplit_once('.') { - if let Some(ext) = clean_ext(ext) { + 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); } - } - } // b) 常见 query 参数中找一次（不重复多轮扫描），忽略大小写 let keys = ["format", "type", "ext", "filetype", "fmt"];