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"];