okooooooo/biliup
1
0 Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Refactor upload logic and add WebSocket log endpoint

Browse Source 
Refactored upload logic to decouple upload and submission, moving UploadLine and related logic to biliup-cli crate root. Added a new WebSocket log endpoint in a dedicated ws.rs module and routed it accordingly. Updated database migration to set certain fields to null instead of empty strings. Adjusted context and download/upload manager logic to pass StreamInfo and Context as needed. Added static file serving route and improved filename generation for uploads. Bumped version to 1.1.8 and enabled 'fs' feature for tower-http.
This commit is contained in:
ForgQi committed 2025-09-27 01:13:18 +08:00
1 parent 9c2c892c39
commit e41acef883
24 files changed
+596 -440
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

Cargo.lock
Generated
+18 -2
View File
@@ -395,7 +395,7 @@ dependencies = [
[[package]]
name = "biliup-cli"
version = "1.1.7"
version = "1.1.8"
dependencies = [
"async-channel",
"async-trait",
@@ -1464,6 +1464,12 @@ dependencies = [
"pin-project-lite",
]
[[package]]
name = "http-range-header"
version = "0.4.2"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "9171a2ea8a68358193d15dd5d70c1c10a2afc3e7e4c5bc92bc9f025cebd7359c"
[[package]]
name = "httparse"
version = "1.10.1"
@@ -3635,7 +3641,7 @@ checksum = "a8f112729512f8e442d81f95a8a7ddf2b7c6b8a1a6f509a95864142b30cab2d3"
[[package]]
name = "stream-gears"
version = "1.1.7"
version = "1.1.8"
dependencies = [
"async-channel",
"async-trait",
@@ -4090,14 +4096,24 @@ checksum = "adc82fd73de2a9722ac5da747f12383d2bfdb93591ee6c58486e0097890f05f2"
dependencies = [
"bitflags 2.9.1",
"bytes",
"futures-core",
"futures-util",
"http",
"http-body",
"http-body-util",
"http-range-header",
"httpdate",
"iri-string",
"mime",
"mime_guess",
"percent-encoding",
"pin-project-lite",
"tokio",
"tokio-util",
"tower",
"tower-layer",
"tower-service",
"tracing",
]
[[package]]
Cargo.toml
+1 -1
View File
@@ -7,7 +7,7 @@ members = [
resolver = "3"
[workspace.package]
version = "1.1.7"
version = "1.1.8"
[workspace.dependencies]
serde = { version = "1.0", features = ["derive"] }
crates/biliup-cli/Cargo.toml
+1 -1
View File
@@ -34,7 +34,7 @@ indexmap = { version = "2.8.0" }
async-trait = { workspace = true }
humantime = "2.2.0"
axum = { version = "0.8", features = ["macros", "ws"] }
tower-http = { version = "0.6.2", features = ["cors"] }
tower-http = { version = "0.6.2", features = ["fs", "cors"] }
reqwest = { workspace = true }
sqlx = { version = "0.8", features = [ "runtime-tokio-rustls", "sqlite" ] }
regex = { workspace = true }
crates/biliup-cli/migrations/2_fix_preprocessor_null.sql
+16 -16
View File
@@ -3,34 +3,34 @@
-- 修复 livestreamers 表的 JSON 字段
UPDATE livestreamers
SET preprocessor = '[]'
WHERE preprocessor IS NULL OR preprocessor = '' OR preprocessor = 'null';
SET preprocessor = null
WHERE preprocessor = '' OR preprocessor = 'null';
UPDATE livestreamers
SET segment_processor = '[]'
WHERE segment_processor IS NULL OR segment_processor = '' OR segment_processor = 'null';
SET segment_processor = null
WHERE segment_processor = '' OR segment_processor = 'null';
UPDATE livestreamers
SET downloaded_processor = '[]'
WHERE downloaded_processor IS NULL OR downloaded_processor = '' OR downloaded_processor = 'null';
SET downloaded_processor = null
WHERE downloaded_processor = '' OR downloaded_processor = 'null';
UPDATE livestreamers
SET postprocessor = '[]'
WHERE postprocessor IS NULL OR postprocessor = '' OR postprocessor = 'null';
SET postprocessor = null
WHERE postprocessor = '' OR postprocessor = 'null';
UPDATE livestreamers
SET override = ''
WHERE override IS NULL OR override = 'null';
SET override = null
WHERE override = '' OR override = 'null';
UPDATE livestreamers
SET excluded_keywords = ''
WHERE excluded_keywords IS NULL OR excluded_keywords = 'null';
SET excluded_keywords = null
WHERE excluded_keywords = '' OR excluded_keywords = 'null';
UPDATE livestreamers
SET opt_args = ''
WHERE opt_args IS NULL OR opt_args = 'null';
SET opt_args = null
WHERE opt_args = '' OR opt_args = 'null';
-- 修复 uploadstreamers 表的 JSON 字段
UPDATE uploadstreamers
SET tags = '[]'
WHERE tags IS NULL OR tags = '' OR tags = 'null';
SET tags = null
WHERE tags = '' OR tags = 'null';
crates/biliup-cli/src/cli.rs
+2 -20
View File
@@ -1,7 +1,8 @@
use biliup::uploader::bilibili::{Studio, Vid};
use biliup::uploader::util::SubmitOption;
use clap::{Parser, Subcommand, ValueEnum};
use clap::{Parser, Subcommand};
use biliup_cli::UploadLine;
use std::path::PathBuf;
#[derive(Parser)]
@@ -148,25 +149,6 @@ pub enum Commands {
},
}
#[derive(Clone, PartialEq, Eq, PartialOrd, Ord, ValueEnum)]
pub enum UploadLine {
Bldsa,
Cnbldsa,
Andsa,
Atdsa,
Bda2,
Cnbd,
Anbd,
Atbd,
Tx,
Cntx,
Antx,
Attx,
Bda,
Txa,
Alia,
}
fn human_size(s: &str) -> Result<u64, String> {
let ret = match s.as_bytes() {
[init @ .., b'K'] => parse_u8(init)? * 1000.0,
crates/biliup-cli/src/lib.rs
+20
View File
@@ -7,6 +7,7 @@ use crate::server::errors::{AppError, AppResult};
use crate::server::infrastructure::connection_pool::ConnectionManager;
use crate::server::infrastructure::repositories;
use crate::server::infrastructure::service_register::ServiceRegister;
use clap::ValueEnum;
use error_stack::ResultExt;
use std::net::ToSocketAddrs;
use std::sync::{Arc, RwLock};
@@ -41,3 +42,22 @@ pub async fn run(addr: (&str, u16)) -> AppResult<()> {
.attach("could not initialize application routes")?;
Ok(())
}
#[derive(Clone, PartialEq, Eq, PartialOrd, Ord, ValueEnum)]
pub enum UploadLine {
Bldsa,
Cnbldsa,
Andsa,
Atdsa,
Bda2,
Cnbd,
Anbd,
Atbd,
Tx,
Cntx,
Antx,
Attx,
Bda,
Txa,
Alia,
}
crates/biliup-cli/src/server/api/endpoints.rs
+47 -235
View File
@@ -1,3 +1,6 @@
use crate::UploadLine;
use crate::server::common::upload::{build_studio, submit_to_bilibili, upload};
use crate::server::common::util::Recorder;
use crate::server::config::Config;
use crate::server::core::download_manager::ActorHandle;
use crate::server::errors::{AppError, report_to_response};
@@ -18,6 +21,7 @@ use axum::http::StatusCode;
use axum::response::{IntoResponse, Response};
use axum::{Json, debug_handler};
use biliup::credential::Credential;
use clap::ValueEnum;
use error_stack::{Report, ResultExt};
use ormlite::{Insert, Model};
use serde::Deserialize;
@@ -91,7 +95,7 @@ pub async fn post_streamers_endpoint(
info!(workers=?service_register.workers, "successfully inserted new live streamers");
Ok(Json(live_streamers))
}
#[debug_handler(state = ServiceRegister)]
pub async fn put_streamers_endpoint(
State(service_register): State<ServiceRegister>,
State(pool): State<ConnectionPool>,
@@ -479,244 +483,52 @@ pub async fn get_status(
})))
}
static ALLOWED_FILES: &[&str] = &["ds_update.log", "download.log", "upload.log"];
#[derive(Debug, Deserialize, Clone)]
pub struct LogsQuery {
file: Option<String>,
}
pub async fn ws_logs(
ws: WebSocketUpgrade,
Query(query): Query<LogsQuery>,
) -> axum::response::Response {
ws.on_upgrade(move |socket| websocket_logs(socket, query))
#[derive(Deserialize)]
pub struct PostUploads {
files: Vec<PathBuf>,
params: UploadStreamer,
}
async fn websocket_logs(mut ws: WebSocket, query: LogsQuery) {
// 参数获取与校验
let file_param = query.file.unwrap_or_else(|| "ds_update.log".to_string());
if !ALLOWED_FILES.contains(&file_param.as_str()) {
let _ = ws
.send(Message::Text(
format!("不允许访问请求的文件: {}", file_param).into(),
))
.await;
let _ = ws.send(Message::Close(None)).await;
return;
}
let log_file = PathBuf::from(&file_param);
// 发送初始内容（最后50行）并获取当前大小
let mut file_size = match send_last_lines(&mut ws, &log_file, 50).await {
Ok(size) => size,
Err(e) => {
match e.kind() {
ErrorKind::NotFound => {
let _ = ws
.send(Message::Text(
format!("日志文件 {} 不存在", log_file.display()).into(),
))
.await;
}
_ => {
let _ = ws
.send(Message::Text(format!("读取日志文件错误: {}", e).into()))
.await;
error!("读取日志文件错误: {}", e);
}
}
let _ = ws.send(Message::Close(None)).await;
return;
}
#[debug_handler]
pub async fn post_uploads(
State(config): State<Arc<RwLock<Config>>>,
Json(json_data): Json<PostUploads>,
) -> Result<Json<serde_json::Value>, Response> {
let upload_config = json_data.params;
let (line, limit, submit_api) = {
let config = config.read().unwrap();
let line = UploadLine::from_str(&config.lines, true).ok();
let limit = config.threads;
let submit_api = config.submit_api.clone();
(line, limit, submit_api)
};
// 心跳/轮询间隔
let mut tick = interval(Duration::from_millis(500));
tick.set_missed_tick_behavior(MissedTickBehavior::Skip);
// 主循环：同时处理客户端消息和文件更新
loop {
tokio::select! {
maybe_msg = ws.recv() => {
match maybe_msg {
Some(Ok(Message::Close(_))) => {
let _ = ws.send(Message::Close(None)).await;
break;
}
Some(Ok(Message::Ping(payload))) => {
// 回应 PONG
let _ = ws.send(Message::Pong(payload)).await;
}
Some(Ok(_)) => {
// 其他消息不处理（Text/Binary等）
}
Some(Err(e)) => {
error!("WebSocket连接错误: {}", e);
break;
}
None => {
info!("WebSocket连接已关闭");
break;
}
}
}
_ = tick.tick() => {
// 文件是否存在
let meta = match fs::metadata(&log_file).await {
Ok(m) => m,
Err(e) if e.kind() == ErrorKind::NotFound => {
let _ = ws.send(Message::Text(format!(
"日志文件 {} 不再存在",
log_file.display()
).into())).await;
break;
}
Err(e) => {
let _ = ws.send(Message::Text(format!("监控日志文件错误: {}", e).into())).await;
error!("websocket_logs错误: {}", e);
break;
}
};
let current_size = meta.len();
// 文件被截断
if current_size < file_size {
let _ = ws.send(Message::Text(Utf8Bytes::from("日志文件被截断，重新加载...".to_string()))).await;
match send_last_lines(&mut ws, &log_file, 50).await {
Ok(size) => file_size = size,
Err(e) => {
let _ = ws.send(Message::Text(format!("读取日志文件错误: {}", e).into())).await;
error!("读取日志文件错误: {}", e);
break;
}
}
continue;
}
// 文件新增内容
if current_size > file_size {
if let Err(e) = send_new_lines_from_offset(&mut ws, &log_file, file_size).await {
let _ = ws.send(Message::Text(format!("监控日志文件错误: {}", e).into())).await;
error!("websocket_logs错误: {}", e);
break;
}
file_size = current_size;
}
}
}
}
let _ = ws.send(Message::Close(None)).await;
debug!("WebSocket日志会话结束: {}", file_param);
}
// 发送最后 n 行，并返回当前文件大小
async fn send_last_lines(
ws: &mut WebSocket,
path: &std::path::Path,
n: usize,
) -> std::io::Result<u64> {
let meta = fs::metadata(path).await?;
let file_size = meta.len();
let file = fs::File::open(path).await?;
let mut reader = BufReader::new(file);
let mut lines = reader.lines();
let mut buf: VecDeque<String> = VecDeque::with_capacity(n);
while let Some(line) = lines.next_line().await? {
if buf.len() == n {
buf.pop_front();
}
buf.push_back(line);
}
for line in buf {
ws.send(Message::Text(Utf8Bytes::from(line)))
info!("通过页面开始上传");
let (bilibili, videos) = upload(
upload_config
.user_cookie
.as_deref()
.unwrap_or("cookies.json"),
None,
line,
&json_data.files,
limit as usize,
)
.await
.map_err(report_to_response)?;
if !videos.is_empty() {
let recorder = Recorder::new(
upload_config.title.clone(),
&upload_config.template_name,
"stream_title",
"",
);
let studio = build_studio(&upload_config, &bilibili, videos, recorder)
.await
.map_err(|e| {
std::io::Error::new(
ErrorKind::ConnectionAborted,
format!("发送WebSocket消息失败: {}", e),
)
})?;
}
Ok(file_size)
}
// 从偏移量开始读取新增内容，并逐行发送
async fn send_new_lines_from_offset(
ws: &mut WebSocket,
path: &std::path::Path,
offset: u64,
) -> std::io::Result<()> {
let mut file = fs::File::open(path).await?;
file.seek(std::io::SeekFrom::Start(offset)).await?;
// 直接读到字符串（UTF-8），若遇到非UTF-8可换成读bytes+lossy
let mut s = String::new();
if let Err(e) = file.read_to_string(&mut s).await {
// 如果遇到非UTF-8数据，降级为 lossy
let mut bytes = Vec::new();
file.seek(std::io::SeekFrom::Start(offset)).await?;
file.read_to_end(&mut bytes).await?;
s = String::from_utf8_lossy(&bytes).into_owned();
if e.kind() != ErrorKind::InvalidData {
// 非编码错误也要汇报
error!("读取日志文件新内容失败: {}", e);
}
}
for line in s.lines() {
ws.send(Message::Text(Utf8Bytes::from(line.to_string())))
.map_err(report_to_response)?;
let response_data = submit_to_bilibili(&bilibili, &studio, submit_api.as_deref())
.await
.map_err(|e| {
std::io::Error::new(
ErrorKind::ConnectionAborted,
format!("发送WebSocket消息失败: {}", e),
)
})?;
.map_err(report_to_response)?;
info!("通过页面上传成功 {:?}", response_data);
}
Ok(())
}
async fn resolve_latest_log_path(
dir: &std::path::Path,
prefix: &str,
suffix: &str,
) -> io::Result<PathBuf> {
// 1) 活跃文件 prefix.log
let active = dir.join(format!("{}.{}", prefix, suffix));
if fs::metadata(&active).await.is_ok() {
return Ok(active);
}
// 2) 回退到归档文件 prefix.*.log 中最新的一个
let mut rd = fs::read_dir(dir).await?;
let pre = format!("{}.", prefix);
let suf = format!(".{}", suffix);
let mut candidates: Vec<(String, PathBuf)> = Vec::new();
while let Some(ent) = rd.next_entry().await? {
let path = ent.path();
if !path.is_file() {
continue;
}
let Some(name) = path.file_name().and_then(|s| s.to_str()) else {
continue;
};
if name.starts_with(&pre) && name.ends_with(&suf) {
candidates.push((name.to_string(), path));
}
}
if candidates.is_empty() {
return Err(io::Error::new(ErrorKind::NotFound, "no log file found"));
}
// tracing-appender 的归档名使用 YYYY-MM-DD，中间按字符串排序即时间顺序
candidates.sort_by(|a, b| a.0.cmp(&b.0));
Ok(candidates.last().unwrap().1.clone())
Ok(Json(serde_json::json!({})))
}
crates/biliup-cli/src/server/api/mod.rs
+1
View File
@@ -6,3 +6,4 @@ pub mod bilibili_endpoints;
pub mod endpoints;
/// 单页应用静态文件处理
pub mod spa;
pub mod ws;
crates/biliup-cli/src/server/api/ws.rs
+255
View File
@@ -0,0 +1,255 @@
use axum::extract::ws::{Message, Utf8Bytes, WebSocket};
use axum::extract::{Query, WebSocketUpgrade};
use serde::Deserialize;
use std::collections::VecDeque;
use std::io;
use std::io::ErrorKind;
use std::path::PathBuf;
use std::time::Duration;
use tokio::fs;
use tokio::io::{AsyncBufReadExt, AsyncReadExt, AsyncSeekExt, BufReader};
use tokio::time::{MissedTickBehavior, interval};
use tracing::{debug, error, info};
static ALLOWED_FILES: &[&str] = &["ds_update.log", "download.log", "upload.log"];
#[derive(Debug, Deserialize, Clone)]
pub struct LogsQuery {
file: Option<String>,
}
pub async fn ws_logs(
ws: WebSocketUpgrade,
Query(query): Query<LogsQuery>,
) -> axum::response::Response {
ws.on_upgrade(move |socket| websocket_logs(socket, query))
}
async fn websocket_logs(mut ws: WebSocket, query: LogsQuery) {
// 参数获取与校验
let file_param = query.file.unwrap_or_else(|| "ds_update.log".to_string());
if !ALLOWED_FILES.contains(&file_param.as_str()) {
let _ = ws
.send(Message::Text(
format!("不允许访问请求的文件: {}", file_param).into(),
))
.await;
let _ = ws.send(Message::Close(None)).await;
return;
}
let log_file = PathBuf::from(&file_param);
// 发送初始内容（最后50行）并获取当前大小
let mut file_size = match send_last_lines(&mut ws, &log_file, 50).await {
Ok(size) => size,
Err(e) => {
match e.kind() {
ErrorKind::NotFound => {
let _ = ws
.send(Message::Text(
format!("日志文件 {} 不存在", log_file.display()).into(),
))
.await;
}
_ => {
let _ = ws
.send(Message::Text(format!("读取日志文件错误: {}", e).into()))
.await;
error!("读取日志文件错误: {}", e);
}
}
let _ = ws.send(Message::Close(None)).await;
return;
}
};
// 心跳/轮询间隔
let mut tick = interval(Duration::from_millis(500));
tick.set_missed_tick_behavior(MissedTickBehavior::Skip);
// 主循环：同时处理客户端消息和文件更新
loop {
tokio::select! {
maybe_msg = ws.recv() => {
match maybe_msg {
Some(Ok(Message::Close(_))) => {
let _ = ws.send(Message::Close(None)).await;
break;
}
Some(Ok(Message::Ping(payload))) => {
// 回应 PONG
let _ = ws.send(Message::Pong(payload)).await;
}
Some(Ok(_)) => {
// 其他消息不处理（Text/Binary等）
}
Some(Err(e)) => {
error!("WebSocket连接错误: {}", e);
break;
}
None => {
info!("WebSocket连接已关闭");
break;
}
}
}
_ = tick.tick() => {
// 文件是否存在
let meta = match fs::metadata(&log_file).await {
Ok(m) => m,
Err(e) if e.kind() == ErrorKind::NotFound => {
let _ = ws.send(Message::Text(format!(
"日志文件 {} 不再存在",
log_file.display()
).into())).await;
break;
}
Err(e) => {
let _ = ws.send(Message::Text(format!("监控日志文件错误: {}", e).into())).await;
error!("websocket_logs错误: {}", e);
break;
}
};
let current_size = meta.len();
// 文件被截断
if current_size < file_size {
let _ = ws.send(Message::Text(Utf8Bytes::from("日志文件被截断，重新加载...".to_string()))).await;
match send_last_lines(&mut ws, &log_file, 50).await {
Ok(size) => file_size = size,
Err(e) => {
let _ = ws.send(Message::Text(format!("读取日志文件错误: {}", e).into())).await;
error!("读取日志文件错误: {}", e);
break;
}
}
continue;
}
// 文件新增内容
if current_size > file_size {
if let Err(e) = send_new_lines_from_offset(&mut ws, &log_file, file_size).await {
let _ = ws.send(Message::Text(format!("监控日志文件错误: {}", e).into())).await;
error!("websocket_logs错误: {}", e);
break;
}
file_size = current_size;
}
}
}
}
let _ = ws.send(Message::Close(None)).await;
debug!("WebSocket日志会话结束: {}", file_param);
}
// 发送最后 n 行，并返回当前文件大小
async fn send_last_lines(
ws: &mut WebSocket,
path: &std::path::Path,
n: usize,
) -> std::io::Result<u64> {
let meta = fs::metadata(path).await?;
let file_size = meta.len();
let file = fs::File::open(path).await?;
let mut reader = BufReader::new(file);
let mut lines = reader.lines();
let mut buf: VecDeque<String> = VecDeque::with_capacity(n);
while let Some(line) = lines.next_line().await? {
if buf.len() == n {
buf.pop_front();
}
buf.push_back(line);
}
for line in buf {
ws.send(Message::Text(Utf8Bytes::from(line)))
.await
.map_err(|e| {
std::io::Error::new(
ErrorKind::ConnectionAborted,
format!("发送WebSocket消息失败: {}", e),
)
})?;
}
Ok(file_size)
}
// 从偏移量开始读取新增内容，并逐行发送
async fn send_new_lines_from_offset(
ws: &mut WebSocket,
path: &std::path::Path,
offset: u64,
) -> std::io::Result<()> {
let mut file = fs::File::open(path).await?;
file.seek(std::io::SeekFrom::Start(offset)).await?;
// 直接读到字符串（UTF-8），若遇到非UTF-8可换成读bytes+lossy
let mut s = String::new();
if let Err(e) = file.read_to_string(&mut s).await {
// 如果遇到非UTF-8数据，降级为 lossy
let mut bytes = Vec::new();
file.seek(std::io::SeekFrom::Start(offset)).await?;
file.read_to_end(&mut bytes).await?;
s = String::from_utf8_lossy(&bytes).into_owned();
if e.kind() != ErrorKind::InvalidData {
// 非编码错误也要汇报
error!("读取日志文件新内容失败: {}", e);
}
}
for line in s.lines() {
ws.send(Message::Text(Utf8Bytes::from(line.to_string())))
.await
.map_err(|e| {
std::io::Error::new(
ErrorKind::ConnectionAborted,
format!("发送WebSocket消息失败: {}", e),
)
})?;
}
Ok(())
}
async fn resolve_latest_log_path(
dir: &std::path::Path,
prefix: &str,
suffix: &str,
) -> io::Result<PathBuf> {
// 1) 活跃文件 prefix.log
let active = dir.join(format!("{}.{}", prefix, suffix));
if fs::metadata(&active).await.is_ok() {
return Ok(active);
}
// 2) 回退到归档文件 prefix.*.log 中最新的一个
let mut rd = fs::read_dir(dir).await?;
let pre = format!("{}.", prefix);
let suf = format!(".{}", suffix);
let mut candidates: Vec<(String, PathBuf)> = Vec::new();
while let Some(ent) = rd.next_entry().await? {
let path = ent.path();
if !path.is_file() {
continue;
}
let Some(name) = path.file_name().and_then(|s| s.to_str()) else {
continue;
};
if name.starts_with(&pre) && name.ends_with(&suf) {
candidates.push((name.to_string(), path));
}
}
if candidates.is_empty() {
return Err(io::Error::new(ErrorKind::NotFound, "no log file found"));
}
// tracing-appender 的归档名使用 YYYY-MM-DD，中间按字符串排序即时间顺序
candidates.sort_by(|a, b| a.0.cmp(&b.0));
Ok(candidates.last().unwrap().1.clone())
}
crates/biliup-cli/src/server/app.rs
+3 -3
View File
@@ -2,8 +2,8 @@ use axum::http;
use crate::server;
use crate::server::api::auth;
use crate::server::api::endpoints::ws_logs;
use crate::server::api::spa::static_handler;
use crate::server::api::ws::ws_logs;
use crate::server::errors::{AppError, AppResult};
use crate::server::infrastructure::service_register::ServiceRegister;
use crate::server::infrastructure::users::Backend;
@@ -48,8 +48,8 @@ impl ApplicationController {
// 配置会话管理层
let session_layer = SessionManagerLayer::new(session_store)
.with_secure(false)
.with_expiry(Expiry::OnInactivity(Duration::days(1)))
.with_signed(key);
.with_expiry(Expiry::OnInactivity(Duration::days(7)));
// .with_signed(key);
// 认证服务配置
// 将会话层与后端结合，建立认证服务，将认证会话作为请求扩展提供
crates/biliup-cli/src/server/common/download.rs
+18 -13
View File
@@ -144,13 +144,13 @@ pub struct SegmentEventProcessor {
tx: Sender<SegmentInfo>,
rx: Receiver<SegmentInfo>,
uploader: Sender<UploaderMessage>,
worker: Arc<Worker>,
ctx: Context,
file_validator: FileValidator,
}
impl SegmentEventProcessor {
/// 创建处理器
pub fn new(uploader: Sender<UploaderMessage>, worker: Arc<Worker>) -> Self {
pub fn new(uploader: Sender<UploaderMessage>, ctx: Context) -> Self {
let (tx, rx) = async_channel::bounded(32); // Use tokio channel for async
Self {
@@ -158,10 +158,17 @@ impl SegmentEventProcessor {
rx,
uploader,
file_validator: FileValidator::new(
worker.clone().config.read().unwrap().filtering_threshold * 1000 * 1000,
ctx.worker
.clone()
.config
.read()
.unwrap()
.filtering_threshold
* 1000
* 1000,
true,
),
worker,
ctx,
}
}
@@ -175,7 +182,7 @@ impl SegmentEventProcessor {
.uploader
.force_send(UploaderMessage::SegmentEvent(
self.rx.clone(),
self.worker.clone(),
self.ctx.clone(),
))
.change_context(AppError::Custom("Failed to send to uploader".to_string()))?;
if let Some(prev) = res {
@@ -239,7 +246,6 @@ impl SegmentEventProcessor {
/// 下载任务
pub struct DownloadTask {
plugin: Arc<dyn DownloadPlugin + Send + Sync>,
stream_info: StreamInfo,
ctx: Context,
rooms_handle: Arc<RoomsHandle>,
}
@@ -252,13 +258,11 @@ struct DownloadComponents {
impl DownloadTask {
pub fn new(
plugin: Arc<dyn DownloadPlugin + Send + Sync>,
stream_info: StreamInfo,
ctx: Context,
rooms_handle: Arc<RoomsHandle>,
) -> Self {
Self {
plugin,
stream_info,
ctx,
rooms_handle,
}
@@ -277,7 +281,7 @@ impl DownloadTask {
);
// 创建事件处理器
let processor = SegmentEventProcessor::new(uploader.clone(), self.ctx.worker.clone());
let processor = SegmentEventProcessor::new(uploader.clone(), self.ctx.clone());
// 启动弹幕客户端
if let Some(ref client) = components.danmaku_client {
@@ -307,15 +311,16 @@ impl DownloadTask {
// 获取配置和主播信息
let config = self.ctx.worker.get_config();
let streamer = self.ctx.worker.get_streamer();
let stream_info = &self.ctx.stream_info;
// 确定文件格式后缀
let suffix = streamer
.format
.unwrap_or_else(|| self.stream_info.suffix.clone());
.unwrap_or_else(|| stream_info.suffix.clone());
// 创建录制器
let recorder = Recorder::new(
streamer.filename_prefix,
&streamer.remark,
&self.stream_info.title,
&stream_info.title,
&suffix,
);
// 可选的弹幕客户端
@@ -329,7 +334,7 @@ impl DownloadTask {
// 创建下载器实例
let downloader = self
.plugin
.create_downloader(&self.stream_info, config, recorder)
.create_downloader(stream_info, config, recorder)
.await;
Ok(DownloadComponents {
@@ -342,7 +347,7 @@ impl DownloadTask {
// 启动弹幕下载逻辑
println!(
"Starting danmaku client for stream: {}",
self.stream_info.url
self.ctx.stream_info.url
);
client.download(Box::new(|_| {})).await?;
Ok(())
crates/biliup-cli/src/server/common/upload.rs
+123 -29
View File
@@ -1,16 +1,20 @@
use crate::UploadLine;
use crate::server::common::util::Recorder;
use crate::server::core::download_manager::UActor;
use crate::server::core::downloader::{SegmentEvent, SegmentInfo};
use crate::server::errors::{AppError, AppResult};
use crate::server::infrastructure::context::Worker;
use crate::server::infrastructure::context::{Context, Worker};
use crate::server::infrastructure::models::UploadStreamer;
use crate::server::infrastructure::models::hook_step::process_video;
use async_channel::Receiver;
use biliup::bilibili::{BiliBili, ResponseData, Studio, Video};
use biliup::bilibili::{BiliBili, Credit, ResponseData, 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};
use chrono::Local;
use error_stack::ResultExt;
use futures::StreamExt;
use std::path::{Path, PathBuf};
@@ -35,24 +39,38 @@ struct UploadedVideos {
pub async fn process_with_upload(
rx: Receiver<SegmentInfo>,
worker: &Worker,
ctx: &Context,
upload_config: UploadStreamer,
) -> AppResult<()> {
info!(upload_config=?upload_config, "Starting process with upload");
// 1. 初始化上传环境
let upload_context = initialize_upload_context(&worker, upload_config).await?;
let upload_context = initialize_upload_context(&ctx.worker, upload_config).await?;
// 2. 流水线处理视频上传
let uploaded_videos = pipeline_upload_videos(rx, &upload_context).await?;
// 3. 提交到B站
if !uploaded_videos.videos.is_empty() {
submit_to_bilibili(&upload_context, uploaded_videos.videos, &worker).await?;
let recorder = Recorder::new(
upload_context.upload_config.title.clone(),
&ctx.worker.live_streamer.remark,
&ctx.stream_info.title,
"",
);
let studio = build_studio(
&upload_context.upload_config,
&upload_context.bilibili,
uploaded_videos.videos,
recorder,
)
.await?;
let submit_api = ctx.worker.config.read().unwrap().submit_api.clone();
submit_to_bilibili(&upload_context.bilibili, &studio, submit_api.as_deref()).await?;
}
// 4. 执行后处理
if !uploaded_videos.paths.is_empty() {
execute_postprocessor(uploaded_videos.paths, &worker).await?;
execute_postprocessor(uploaded_videos.paths, &ctx.worker).await?;
}
Ok(())
@@ -102,7 +120,7 @@ async fn pipeline_upload_videos(
context: &UploadContext,
) -> AppResult<UploadedVideos> {
// let mut desc_v2 = Vec::new();
// for credit in desc_v2_credit {
// for credit in context.upload_config.desc_v2_credit {
// desc_v2.push(Credit {
// type_id: credit.type_id,
// raw_text: credit.raw_text,
@@ -167,12 +185,11 @@ async fn upload_single_file(file_path: &Path, context: &UploadContext) -> AppRes
Ok(video)
}
async fn submit_to_bilibili(
context: &UploadContext,
videos: Vec<Video>,
worker: &Worker,
pub async fn submit_to_bilibili(
bilibili: &BiliBili,
studio: &Studio,
submit_api: Option<&str>,
) -> AppResult<ResponseData> {
let studio = build_studio(context, videos).await?;
// let submit = match worker.config.read().unwrap().submit_api {
// Some(submit) => SubmitOption::from_str(&submit).unwrap_or(SubmitOption::App),
// _ => SubmitOption::App,
@@ -185,23 +202,17 @@ async fn submit_to_bilibili(
// _ => bilibili.submit_by_app(&studio, None).await,
// };
let submit_option = worker
.config
.read()
.unwrap()
.submit_api
.as_deref()
.and_then(|s| SubmitOption::from_str(s).ok())
.unwrap_or(SubmitOption::App);
let submit_option = match submit_api {
Some(submit) => SubmitOption::from_str(submit).unwrap_or(SubmitOption::App),
_ => SubmitOption::App,
};
let result = match submit_option {
SubmitOption::BCutAndroid => context
.bilibili
SubmitOption::BCutAndroid => bilibili
.submit_by_bcut_android(&studio, None)
.await
.change_context(AppError::Unknown)?,
_ => context
.bilibili
_ => bilibili
.submit_by_app(&studio, None)
.await
.change_context(AppError::Unknown)?,
@@ -210,8 +221,12 @@ async fn submit_to_bilibili(
Ok(result)
}
async fn build_studio(context: &UploadContext, videos: Vec<Video>) -> AppResult<Studio> {
let upload_config = &context.upload_config;
pub(crate) async fn build_studio(
upload_config: &UploadStreamer,
bilibili: &BiliBili,
videos: Vec<Video>,
recorder: Recorder,
) -> AppResult<Studio> {
// 使用 Builder 模式简化构建
let mut studio: Studio = Studio::builder()
.desc(upload_config.description.clone().unwrap_or_default())
@@ -222,7 +237,7 @@ async fn build_studio(context: &UploadContext, videos: Vec<Video>) -> AppResult<
.source(upload_config.copyright_source.clone().unwrap_or_default())
.tag(upload_config.tags.join(","))
.maybe_tid(upload_config.tid)
.title(upload_config.title.clone().unwrap_or_default())
.title(recorder.format_filename(Local::now()))
.videos(videos)
.dolby(upload_config.dolby.unwrap_or_default())
// .lossless_music(upload_config.)
@@ -239,8 +254,8 @@ async fn build_studio(context: &UploadContext, videos: Vec<Video>) -> AppResult<
.build();
// 处理封面上传
if !studio.cover.is_empty()
&& let Ok(c) = &std::fs::read(&studio.cover).map_err(|e| error!(e=?e))
&& let Ok(url) = context.bilibili.cover_up(c).await.map_err(|e| error!(e=?e))
&& let Ok(c) = &std::fs::read(&studio.cover).inspect_err(|e| error!(e=?e))
&& let Ok(url) = bilibili.cover_up(c).await.inspect_err(|e| error!(e=?e))
{
studio.cover = url;
};
@@ -255,3 +270,82 @@ pub async fn execute_postprocessor(video_paths: Vec<PathBuf>, worker: &Worker) -
}
Ok(())
}
pub async fn upload(
cookie_file: impl AsRef<Path>,
proxy: Option<&str>,
line: Option<UploadLine>,
video_paths: &[PathBuf],
limit: usize,
) -> AppResult<(BiliBili, Vec<Video>)> {
let bilibili = login_by_cookies(&cookie_file, proxy).await;
let bilibili = match bilibili {
Err(Kind::IO(_)) => bilibili.change_context_lazy(|| {
AppError::Custom(format!(
"open cookies file: {}",
&cookie_file.as_ref().to_string_lossy()
))
})?,
_ => bilibili.change_context_lazy(|| AppError::Unknown)?,
};
let client = StatelessClient::default();
let mut videos = Vec::new();
let line = match line {
Some(UploadLine::Bldsa) => line::bldsa(),
Some(UploadLine::Cnbldsa) => line::cnbldsa(),
Some(UploadLine::Andsa) => line::andsa(),
Some(UploadLine::Atdsa) => line::atdsa(),
Some(UploadLine::Bda2) => line::bda2(),
Some(UploadLine::Cnbd) => line::cnbd(),
Some(UploadLine::Anbd) => line::anbd(),
Some(UploadLine::Atbd) => line::atbd(),
Some(UploadLine::Tx) => line::tx(),
Some(UploadLine::Cntx) => line::cntx(),
Some(UploadLine::Antx) => line::antx(),
Some(UploadLine::Attx) => line::attx(),
// Some(UploadLine::Bda) => line::bda(),
Some(UploadLine::Txa) => line::txa(),
Some(UploadLine::Alia) => line::alia(),
_ => Probe::probe(&client.client).await.unwrap_or_default(),
};
for video_path in video_paths {
println!(
"{:?}",
video_path
.canonicalize()
.change_context_lazy(|| AppError::Unknown)?
.to_str()
);
info!("{line:?}");
let video_file = VideoFile::new(&video_path).change_context_lazy(|| 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)
.await
.change_context_lazy(|| AppError::Unknown)?;
let instant = Instant::now();
let video = uploader
.upload(client.clone(), limit, |vs| {
vs.map(|vs| {
let chunk = vs?;
let len = chunk.len();
Ok((chunk, len))
})
})
.await
.change_context_lazy(|| AppError::Unknown)?;
let t = instant.elapsed().as_millis();
info!(
"Upload completed: {file_name} => cost {:.2}s, {:.2} MB/s.",
t as f64 / 1000.,
total_size as f64 / 1000. / t as f64
);
videos.push(video);
}
Ok((bilibili, videos))
}
crates/biliup-cli/src/server/common/util.rs
+8 -1
View File
@@ -1,4 +1,4 @@
use chrono::{Duration, Local};
use chrono::{DateTime, Duration, Local};
use serde::{Deserialize, Serialize};
use std::path::{Path, PathBuf};
use url::Url;
@@ -58,6 +58,13 @@ impl Recorder {
}
}
/// 生成“基名”（不带扩展名）
pub fn format_filename(&self, date_time: DateTime<Local>) -> String {
let template = self.filename_template();
date_time.format(&template).to_string()
}
/// 直接生成带扩展名的完整路径（当前目录下）
pub fn generate_path(&self) -> PathBuf {
PathBuf::from(self.generate_filename()).with_extension(&self.suffix)
crates/biliup-cli/src/server/config.rs
+1 -9
View File
@@ -1,6 +1,7 @@
use crate::server::core::downloader::DownloaderType;
use crate::server::errors::{AppError, AppResult};
use crate::server::infrastructure::models::hook_step::HookStep;
use biliup::bilibili::Credit;
use error_stack::bail;
use serde::{Deserialize, Serialize};
use std::{collections::HashMap, path::Path, path::PathBuf};
@@ -296,15 +297,6 @@ pub struct StreamerConfig {
pub override_cfg: Option<HashMap<String, serde_json::Value>>,
}
/// 制作人员信息
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct Credit {
/// 用户名
pub username: String,
/// 用户ID
pub uid: String,
}
/// 用户配置结构体
#[derive(bon::Builder, Debug, Clone, Serialize, Deserialize, Default)]
pub struct UserConfig {
crates/biliup-cli/src/server/core/download_manager.rs
+10 -10
View File
@@ -114,9 +114,9 @@ impl DActor {
/// * `msg` - 要处理的下载消息
async fn handle_message(&mut self, msg: DownloaderMessage) -> AppResult<()> {
match msg {
DownloaderMessage::Start(plugin, stream_info, ctx, rooms_handle) => {
DownloaderMessage::Start(plugin, ctx, rooms_handle) => {
// 创建下载任务
let task = DownloadTask::new(plugin, stream_info, ctx, rooms_handle);
let task = DownloadTask::new(plugin, ctx, rooms_handle);
// 执行下载（使用 Result 链式处理）
task.execute(&self.sender).await?;
@@ -152,17 +152,18 @@ impl UActor {
/// * `msg` - 要处理的上传消息
async fn handle_message(&mut self, msg: UploaderMessage) {
match msg {
UploaderMessage::SegmentEvent(rx, worker) => {
worker.change_status(Stage::Upload, WorkerStatus::Pending);
let result = match worker.get_upload_config() {
Some(config) => process_with_upload(rx, &worker, config).await,
UploaderMessage::SegmentEvent(rx, ctx) => {
ctx.worker
.change_status(Stage::Upload, WorkerStatus::Pending);
let result = match ctx.worker.get_upload_config() {
Some(config) => process_with_upload(rx, &ctx, config).await,
None => {
let mut paths = Vec::new();
while let Ok(event) = rx.recv().await {
paths.push(event.prev_file_path);
}
// 无上传配置时，直接执行后处理
execute_postprocessor(paths, &worker).await
execute_postprocessor(paths, &ctx.worker).await
}
};
@@ -170,7 +171,7 @@ impl UActor {
error!("Process segment event failed: {}", e);
// 可以添加错误通知机制
}
worker.change_status(Stage::Upload, WorkerStatus::Idle);
ctx.worker.change_status(Stage::Upload, WorkerStatus::Idle);
}
}
}
@@ -234,7 +235,7 @@ impl ActorHandle {
#[derive(Debug)]
pub enum UploaderMessage {
/// 分段事件消息，包含事件、接收器和工作器
SegmentEvent(Receiver<SegmentInfo>, Arc<Worker>),
SegmentEvent(Receiver<SegmentInfo>, Context),
}
/// 下载消息枚举
@@ -243,7 +244,6 @@ pub enum DownloaderMessage {
/// 开始下载消息，包含插件、流信息、上下文和房间句柄
Start(
Arc<dyn DownloadPlugin + Send + Sync>,
StreamInfo,
Context,
Arc<RoomsHandle>,
),
crates/biliup-cli/src/server/core/monitor.rs
+2 -3
View File
@@ -29,7 +29,7 @@ async fn start_client(
let ulr = room.get_streamer().url;
interval = room.get_config().event_loop_interval;
info!("[{platform_name}] room: {ulr}");
let mut ctx = Context::new(room.clone());
let mut ctx = Context::new(room.clone(), Default::default());
// 检查直播状态
match plugin.check_status(&mut ctx).await.unwrap() {
StreamStatus::Live { stream_info } => {
@@ -41,8 +41,7 @@ async fn start_client(
.down_sender
.send(DownloaderMessage::Start(
plugin.clone(),
stream_info,
ctx,
Context::new(room.clone(), stream_info),
rooms_handle.clone(),
))
.await
crates/biliup-cli/src/server/core/plugin.rs
+4 -3
View File
@@ -4,6 +4,7 @@ use crate::server::core::downloader::Downloader;
use crate::server::errors::{AppError, AppResult};
use crate::server::infrastructure::context::{Context, Worker};
use async_trait::async_trait;
use chrono::{DateTime, Local};
use error_stack::Report;
use serde::{Deserialize, Serialize};
use std::collections::HashMap;
@@ -13,7 +14,7 @@ use time::OffsetDateTime;
/// 流信息结构
/// 包含直播流的详细信息
#[derive(Debug, Clone, Serialize, Deserialize)]
#[derive(Debug, Clone, Serialize, Deserialize, Default)]
pub struct StreamInfo {
/// 主播名称
pub name: String,
@@ -24,8 +25,8 @@ pub struct StreamInfo {
/// 直播标题
pub title: String,
/// 直播开始时间
#[serde(with = "time::serde::rfc3339")]
pub date: OffsetDateTime,
// #[serde(with = "time::serde::rfc3339")]
pub date: DateTime<Local>,
/// 直播封面路径（可选）
pub live_cover_path: Option<String>,
/// 平台名称
crates/biliup-cli/src/server/infrastructure/context.rs
+4 -1
View File
@@ -1,5 +1,6 @@
use crate::server::config::Config;
use crate::server::core::downloader::Downloader;
use crate::server::core::plugin::StreamInfo;
use crate::server::errors::{AppError, AppResult};
use crate::server::infrastructure::connection_pool::ConnectionPool;
use crate::server::infrastructure::models::{LiveStreamer, UploadStreamer};
@@ -18,6 +19,7 @@ use tracing::info;
pub struct Context {
/// 工作器实例
pub worker: Arc<Worker>,
pub stream_info: StreamInfo,
/// 扩展数据容器
pub extension: Extensions,
}
@@ -27,9 +29,10 @@ impl Context {
///
/// # 参数
/// * `worker` - 工作器实例的Arc引用
pub fn new(worker: Arc<Worker>) -> Self {
pub fn new(worker: Arc<Worker>, stream_info: StreamInfo) -> Self {
Self {
worker,
stream_info,
extension: Default::default(),
}
}
crates/biliup-cli/src/server/infrastructure/models.rs
+5 -2
View File
@@ -102,7 +102,8 @@ pub struct LiveStreamer {
pub format: Option<String>,
/// 覆盖配置（"override"为关键字，字段名避让）
#[ormlite(column = "override")]
pub override_: Option<Value>,
#[serde(rename = "override")]
pub override_cfg: Option<Value>,
/// 预处理器列表（JSON格式）
/// 注意：数据库空与json空有区别，所以这里不能用#[ormlite(json)]
@@ -162,7 +163,9 @@ pub struct InsertLiveStreamer {
pub upload_streamers_id: Option<i64>, // FK，可空
pub format: Option<String>,
#[ormlite(column = "override")]
pub override_: Option<Value>, // "override" 为关键字，字段名避让
// “override” 是字段名，这里改为 override_cfg 避免与保留字混淆
#[serde(rename = "override")]
pub override_cfg: Option<Value>, // "override" 为关键字，字段名避让
// #[ormlite(json)] 数据库空与json空有区别所以这里不能用
// pub preprocessor: Option<Vec<String>>,
crates/biliup-cli/src/server/router.rs
+16 -2
View File
@@ -6,12 +6,16 @@ use crate::server::api::endpoints::{
delete_template_endpoint, delete_user_endpoint, get_configuration, get_qrcode, get_status,
get_streamer_info, get_streamers_endpoint, get_upload_streamer_endpoint,
get_upload_streamers_endpoint, get_users_endpoint, get_videos, login_by_qrcode,
post_streamers_endpoint, put_configuration, put_streamers_endpoint, ws_logs,
post_streamers_endpoint, post_uploads, put_configuration, put_streamers_endpoint,
};
use crate::server::infrastructure::service_register::ServiceRegister;
use axum::Router;
use axum::body::Body;
use axum::http::Request;
use axum::response::IntoResponse;
use axum::routing::{delete, get, post};
use tower::ServiceExt;
use tower_http::services::ServeFile;
/// 创建应用程序路由
pub fn router(service_register: ServiceRegister) -> Router<()> {
Router::new()
@@ -51,5 +55,15 @@ pub fn router(service_register: ServiceRegister) -> Router<()> {
// 视频文件管理路由
.route("/v1/videos", get(get_videos)) // 获取视频列表
.route("/v1/status", get(get_status))
.route("/v1/uploads", post(post_uploads))
.route_service("/static/{path}", get(using_serve_file_from_a_route))
.with_state(service_register) // 注入服务注册器状态
}
async fn using_serve_file_from_a_route(
axum::extract::Path(path): axum::extract::Path<String>,
request: Request<Body>,
) -> impl IntoResponse {
let serve_file = ServeFile::new(path);
serve_file.oneshot(request).await
}
crates/biliup-cli/src/uploader.rs
+1 -1
View File
@@ -1,4 +1,3 @@
use crate::cli::UploadLine;
use biliup::client::StatelessClient;
use biliup::error::Kind;
use biliup::uploader::bilibili::{BiliBili, Studio, Vid, Video};
@@ -6,6 +5,7 @@ use biliup::uploader::credential::{Credential, LoginInfo};
use biliup::uploader::line::Probe;
use biliup::uploader::util::SubmitOption;
use biliup::uploader::{VideoFile, credential, line, load_config};
use biliup_cli::UploadLine;
use biliup_cli::server::errors::{AppError, AppResult};
use bytes::{Buf, Bytes};
use clap::ValueEnum;
crates/biliup/src/uploader/bilibili.rs
+1 -1
View File
@@ -173,7 +173,7 @@ pub struct Subtitle {
lan: String,
}
#[derive(Deserialize, Serialize, Debug)]
#[derive(Deserialize, Serialize, Debug, Clone)]
pub struct Credit {
#[serde(rename(deserialize = "type_id", serialize = "type"))]
pub type_id: i8,
crates/stream-gears/src/server.rs
+3 -1
View File
@@ -15,6 +15,7 @@ use biliup_cli::server::infrastructure::context::{Context, Worker};
use biliup_cli::server::infrastructure::repositories;
use biliup_cli::server::infrastructure::repositories::get_upload_config;
use biliup_cli::server::infrastructure::service_register::ServiceRegister;
use chrono::Local;
use error_stack::{Report, ResultExt};
use fancy_regex::Regex;
use pyo3::exceptions::PyRuntimeError;
@@ -237,7 +238,8 @@ pub fn stream_info_from_py(
};
// date 直接使用传入的 start_time（保留为 Python 对象）
let date = OffsetDateTime::now_utc();
// let date = OffsetDateTime::now_utc();
let date = Local::now();
// end_time: 若传入 None 或“假值”，则使用 time.localtime()
// let end_time_obj: PyObject = match end_time {
// Some(et) if et.is_true()? => et.to_object(py),
crates/stream-gears/src/uploader.rs
+36 -86
View File
@@ -9,6 +9,9 @@ use futures::StreamExt;
use pyo3::prelude::*;
use pyo3::pyclass;
use biliup_cli::server::common;
use biliup_cli::server::common::upload::submit_to_bilibili;
use biliup_cli::server::errors::{AppError, AppResult};
use bon::Builder;
use error_stack::ResultExt;
use std::collections::HashMap;
@@ -17,8 +20,6 @@ use std::str::FromStr;
use std::time::Instant;
use tracing::info;
use biliup_cli::server::errors::{AppError, AppResult};
#[pyclass]
#[derive(Clone, PartialEq, Eq, PartialOrd, Ord)]
pub enum UploadLine {
@@ -39,6 +40,30 @@ pub enum UploadLine {
Alia,
}
impl Into<biliup_cli::UploadLine> for UploadLine {
fn into(self) -> biliup_cli::UploadLine {
use UploadLine as P;
use biliup_cli::UploadLine as C;
match self {
P::Bldsa => C::Bldsa,
P::Cnbldsa => C::Cnbldsa,
P::Andsa => C::Andsa,
P::Atdsa => C::Atdsa,
P::Bda2 => C::Bda2,
P::Cnbd => C::Cnbd,
P::Anbd => C::Anbd,
P::Atbd => C::Atbd,
P::Tx => C::Tx,
P::Cntx => C::Cntx,
P::Antx => C::Antx,
P::Attx => C::Attx,
P::Bda => C::Bda,
P::Txa => C::Txa,
P::Alia => C::Alia,
}
}
}
#[derive(FromPyObject)]
pub struct PyCredit {
#[pyo3(item("type"))]
@@ -112,74 +137,14 @@ pub async fn upload(
extra_fields,
} = studio_pre;
let bilibili = login_by_cookies(&cookie_file, proxy).await;
let bilibili = match bilibili {
Err(Kind::IO(_)) => bilibili.change_context_lazy(|| {
AppError::Custom(format!(
"open cookies file: {}",
&cookie_file.to_string_lossy()
))
})?,
_ => bilibili.change_context_lazy(|| AppError::Unknown)?,
};
let client = StatelessClient::default();
let mut videos = Vec::new();
let line = match line {
Some(UploadLine::Bldsa) => line::bldsa(),
Some(UploadLine::Cnbldsa) => line::cnbldsa(),
Some(UploadLine::Andsa) => line::andsa(),
Some(UploadLine::Atdsa) => line::atdsa(),
Some(UploadLine::Bda2) => line::bda2(),
Some(UploadLine::Cnbd) => line::cnbd(),
Some(UploadLine::Anbd) => line::anbd(),
Some(UploadLine::Atbd) => line::atbd(),
Some(UploadLine::Tx) => line::tx(),
Some(UploadLine::Cntx) => line::cntx(),
Some(UploadLine::Antx) => line::antx(),
Some(UploadLine::Attx) => line::attx(),
// Some(UploadLine::Bda) => line::bda(),
Some(UploadLine::Txa) => line::txa(),
Some(UploadLine::Alia) => line::alia(),
_ => Probe::probe(&client.client).await.unwrap_or_default(),
};
for video_path in video_path {
println!(
"{:?}",
video_path
.canonicalize()
.change_context_lazy(|| AppError::Unknown)?
.to_str()
);
info!("{line:?}");
let video_file = VideoFile::new(&video_path).change_context_lazy(|| 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)
.await
.change_context_lazy(|| AppError::Unknown)?;
let instant = Instant::now();
let video = uploader
.upload(client.clone(), limit, |vs| {
vs.map(|vs| {
let chunk = vs?;
let len = chunk.len();
Ok((chunk, len))
})
})
.await
.change_context_lazy(|| AppError::Unknown)?;
let t = instant.elapsed().as_millis();
info!(
"Upload completed: {file_name} => cost {:.2}s, {:.2} MB/s.",
t as f64 / 1000.,
total_size as f64 / 1000. / t as f64
);
videos.push(video);
}
let (bilibili, videos) = common::upload::upload(
&cookie_file,
proxy,
line.map(Into::into),
video_path.as_slice(),
limit,
)
.await?;
let mut desc_v2 = Vec::new();
for credit in desc_v2_credit {
@@ -225,20 +190,5 @@ pub async fn upload(
studio.cover = url;
}
let submit = match submit {
Some(submit) => SubmitOption::from_str(submit).unwrap_or(SubmitOption::App),
_ => SubmitOption::App,
};
let submit_result = match submit {
SubmitOption::BCutAndroid => bilibili
.submit_by_bcut_android(&studio, proxy)
.await
.change_context_lazy(|| AppError::Unknown)?,
_ => bilibili
.submit_by_app(&studio, proxy)
.await
.change_context_lazy(|| AppError::Unknown)?,
};
Ok(submit_result)
Ok(submit_to_bilibili(&bilibili, &studio, submit).await?)
}