Compare commits

..
5 Commits
Author SHA1 Message Date
Bluemangoo 8d4b6e79f3 fix export path 2026-05-04 15:26:22 +08:00
Bluemangoo cc4e6f0615 fix temp create&remove 2026-04-29 21:15:44 +08:00
Bluemangoo cdf617b5c3 fix task clear 2026-04-24 13:12:03 +08:00
Bluemangoo 46cf5cdf4b tcp proxy 2026-04-24 12:50:40 +08:00
Bluemangoo adc3afc7d5 fix ram 2026-04-24 12:00:03 +08:00
12 changed files with 206 additions and 92 deletions
Generated
+26
View File
@@ -126,6 +126,17 @@ dependencies = [
"yaml_serde",
]
[[package]]
name = "async-http-proxy"
version = "1.2.5"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "29faa5d4d308266048bd7505ba55484315a890102f9345b9ff4b87de64201592"
dependencies = [
"httparse",
"thiserror 1.0.69",
"tokio",
]
[[package]]
name = "async-lock"
version = "3.4.1"
@@ -444,6 +455,7 @@ name = "communicator"
version = "0.1.0"
dependencies = [
"anyhow",
"async-http-proxy",
"bytes",
"futures",
"futures-util",
@@ -455,7 +467,9 @@ dependencies = [
"serde_json",
"tokio",
"tokio-rustls",
"tokio-socks",
"tokio-util",
"url",
"webpki-roots",
]
@@ -2715,6 +2729,18 @@ dependencies = [
"tokio",
]
[[package]]
name = "tokio-socks"
version = "0.5.2"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "0d4770b8024672c1101b3f6733eab95b18007dbe0847a8afe341fcf79e06043f"
dependencies = [
"either",
"futures-util",
"thiserror 1.0.69",
"tokio",
]
[[package]]
name = "tokio-util"
version = "0.7.18"
+5 -1
View File
@@ -34,4 +34,8 @@ lazy_static = "1.5.0"
structopt = "0.3.26"
tokio-util = "0.7.18"
futures-util = "0.3.32"
twox-hash = "2.1.2"
twox-hash = "2.1.2"
futures = "0.3.32"
tokio-socks = "0.5.2"
url = "2.5.8"
async-http-proxy = "1.2.5"
+35 -61
View File
@@ -59,6 +59,17 @@ pub fn get_hex_index(input: &str) -> String {
format!("{:032x}", hash_val)
}
pub fn empty_dir(base: PathBuf, name: String) -> PathBuf {
let mut dir = base.join(&name);
let mut cnt = 1;
while dir.exists() {
dir = base.join(format!("{}_{}", &name, cnt));
cnt += 1;
}
std::fs::create_dir_all(&dir).unwrap();
dir
}
pub async fn extract_unity_asset_bundle(
app_config: &AppConfig,
sync_context: &SyncContext,
@@ -69,11 +80,13 @@ pub async fn extract_unity_asset_bundle(
category: &str,
) -> Result<(PathBuf, bool), ExportPipelineError> {
let hash = get_hex_index(export_path);
let output_dir = std::env::temp_dir()
.join("sekai-updater")
.join("extract")
.join(&sync_context.region)
.join(hash);
let output_dir = empty_dir(
std::env::temp_dir()
.join("sekai-updater")
.join("extract")
.join(&sync_context.region),
hash,
);
let Some(asset_studio_cli_path) = app_config.tools.asset_studio_cli_path.as_deref() else {
return Ok((asset_bundle_file.parent().unwrap().to_path_buf(), true));
};
@@ -166,7 +179,6 @@ pub async fn post_process_exported_files(
handle_usm_files(
export_path,
sync_context,
region,
&app_config.tools.ffmpeg_path,
&app_config.execution.retry,
)
@@ -189,7 +201,6 @@ pub async fn post_process_exported_files(
async fn handle_usm_files(
export_path: &Path,
sync_context: &SyncContext,
region: &RegionConfig,
ffmpeg_path: &str,
retry: &crate::core::config::RetryConfig,
) -> Result<Vec<PathBuf>, ExportPipelineError> {
@@ -198,28 +209,17 @@ async fn handle_usm_files(
return Ok(Vec::new());
}
let usm_input = if usm_files.len() == 1 {
usm_files[0].clone()
} else {
merge_usm_files(export_path, &usm_files)?
};
let mut out: Vec<PathBuf> = vec![];
process_usm_file(
&usm_input,
export_path,
sync_context,
region,
ffmpeg_path,
retry,
)
.await
for f in usm_files {
out.append(&mut process_usm_file(&f, sync_context, ffmpeg_path, retry).await?);
}
Ok(out)
}
async fn process_usm_file(
usm_file: &Path,
export_path: &Path,
sync_context: &SyncContext,
_: &RegionConfig,
ffmpeg_path: &str,
retry: &crate::core::config::RetryConfig,
) -> Result<Vec<PathBuf>, ExportPipelineError> {
@@ -235,7 +235,10 @@ async fn process_usm_file(
if sync_context.export.video.convert_to_mp4
&& sync_context.export.video.direct_usm_to_mp4_with_ffmpeg
{
let mp4 = export_path.join(format!("{output_name}.mp4"));
let mp4 = usm_file
.parent()
.unwrap()
.join(format!("{output_name}.mp4"));
convert_usm_to_mp4(usm_file, &mp4, ffmpeg_path, retry).await?;
remove_file_if_exists(usm_file)?;
return Ok(vec![mp4]);
@@ -247,7 +250,7 @@ async fn process_usm_file(
.and_then(|metadata| metadata.video_frame_rate())
.filter(|(_, denominator)| *denominator > 0)
.map(FrameRate::from_tuple);
let extracted = codec::export_usm(usm_file, export_path)?;
let extracted = codec::export_usm(usm_file, usm_file.parent().unwrap())?;
let mut generated = extracted.clone();
if sync_context.export.video.convert_to_mp4 {
@@ -258,7 +261,10 @@ async fn process_usm_file(
.map(|ext| ext.eq_ignore_ascii_case("m2v"))
.unwrap_or(false)
{
let mp4 = export_path.join(format!("{output_name}.mp4"));
let mp4 = usm_file
.parent()
.unwrap()
.join(format!("{output_name}.mp4"));
convert_m2v_to_mp4(
&extracted_file,
&mp4,
@@ -314,7 +320,7 @@ async fn handle_acb_files(
fn process_acb_file(
acb_file: &Path,
output_dir: &Path,
_output_dir: &Path,
sync_context: &SyncContext,
region: &RegionConfig,
ffmpeg_path: &str,
@@ -368,7 +374,7 @@ fn process_acb_file(
&retry,
)
})?;
let final_outputs = move_result_files(output_dir, &generated)?;
let final_outputs = move_result_files(acb_file.parent().unwrap(), &generated)?;
remove_file_if_exists(acb_file)?;
Ok(final_outputs)
@@ -416,7 +422,7 @@ fn process_hca_file(
generated.clear();
}
let final_outputs = move_result_files(output_dir, &generated)?;
let final_outputs = move_result_files(hca_file.parent().unwrap(), &generated)?;
Ok(final_outputs)
}
@@ -697,38 +703,6 @@ fn panic_message(panic: Box<dyn std::any::Any + Send>) -> String {
}
}
fn merge_usm_files(dir: &Path, usm_files: &[PathBuf]) -> Result<PathBuf, ExportPipelineError> {
let dir_name = dir
.file_name()
.and_then(|name| name.to_str())
.unwrap_or("merged");
let merged_file = dir.join(format!("{dir_name}.usm"));
let mut target =
std::fs::File::create(&merged_file).map_err(|source| ExportPipelineError::Io {
path: merged_file.clone(),
source,
})?;
for source_path in usm_files {
if *source_path == merged_file {
continue;
}
let mut source =
std::fs::File::open(source_path).map_err(|source| ExportPipelineError::Io {
path: source_path.clone(),
source,
})?;
std::io::copy(&mut source, &mut target).map_err(|source| ExportPipelineError::Io {
path: source_path.clone(),
source,
})?;
drop(source);
remove_file_if_exists(source_path)?;
}
Ok(merged_file)
}
pub fn find_files(dir: &Path) -> Result<Vec<PathBuf>, ExportPipelineError> {
let mut files = Vec::new();
walk(dir, &mut |path| {
+4 -5
View File
@@ -1,7 +1,7 @@
use crate::config::{ClientConfig, Profile};
use crate::queue::SharedQueue;
use crate::signal::Signal;
use crate::task::{AtomicCounters, AutoSaveManifest, post_run, run_main, run_side};
use crate::task::{AtomicCounters, AutoSaveManifest, pre_run, run_main, run_side};
use common::strings::REGION_NOT_FOUND;
use common::updater::DownloadTask;
use communicator::{ClientManager, Identity, TunnelEndpoint, TunnelListener, connect_tunnel};
@@ -183,7 +183,6 @@ async fn main() -> anyhow::Result<()> {
let liveness_tx = liveness_tx.clone();
join_set.spawn(async move {
let _guard = liveness_tx;
let mut inner_set = JoinSet::new();
loop {
if cancel_token.is_cancelled() {
return;
@@ -205,7 +204,7 @@ async fn main() -> anyhow::Result<()> {
let local_manifest = local_manifest.clone();
let signal = signal.clone();
inner_set.spawn(async move {
tokio::spawn(async move {
loop {
if client.get_client().await.is_err() {
return;
@@ -234,7 +233,7 @@ async fn main() -> anyhow::Result<()> {
if cancel_token.is_cancelled() {
return;
}
let sync_id = post_run(client.clone(), profile.clone(), tasks.clone(), cnt.clone()).await;
let sync_id = pre_run(client.clone(), profile.clone(), tasks.clone(), cnt.clone()).await;
let sig = signal.pick().await;
let result = match sync_id {
Ok(Some(id)) => {
@@ -329,7 +328,7 @@ async fn main() -> anyhow::Result<()> {
if cancel_token.is_cancelled() {
return;
}
let sync_id = post_run(client.clone(), profile.clone(), tasks.clone(), cnt.clone()).await;
let sync_id = pre_run(client.clone(), profile.clone(), tasks.clone(), cnt.clone()).await;
let sig = signal.pick().await;
let result = match sync_id {
Ok(Some(id)) => {
+17
View File
@@ -103,6 +103,23 @@ impl<T> SharedQueue<T> {
})
}
pub fn clear(&self) {
let mut queue = self.inner.data.lock().unwrap();
let removed_count = queue.len();
if removed_count > 0 {
queue.clear();
let prev_pending = self.inner.pending.fetch_sub(removed_count, Ordering::SeqCst);
if prev_pending == removed_count {
self.inner.done_cond.notify_all();
}
}
}
/// 阻塞当前线程,直到所有在途任务(pending == 0)全部处理完
pub fn wait_until_all_consumed(&self) {
let mut _queue_lock = self.inner.data.lock().unwrap();
+16 -6
View File
@@ -15,7 +15,7 @@ use tokio::sync::{RwLock, Semaphore};
use tokio::task::JoinSet;
use tokio_util::sync::CancellationToken;
pub async fn post_run(
pub async fn pre_run(
client: Arc<ClientManager>,
profile: Arc<(String, Arc<RwLock<Profile>>)>,
queue: SharedQueue<DownloadTask>,
@@ -55,6 +55,7 @@ pub async fn post_run(
return Ok(None);
}
cnt.reset();
queue.clear();
queue.push_all(tasks);
Ok(Some(id))
}
@@ -121,11 +122,6 @@ pub async fn run_main(
match r {
Ok(Ok(())) => cnt.inc_success(),
Ok(Err(e)) => {
if e.to_string()
.contains("Session did not reconnect within 15s")
{
return Err(anyhow!(e));
}
error!("{}", e);
cnt.inc_failure()
}
@@ -163,12 +159,21 @@ pub async fn run_side(
let n = p1.concurrent.unwrap_or(5);
let semaphore = Arc::new(Semaphore::new(n));
let mut join_set = JoinSet::new();
let cancel_token = CancellationToken::new();
while let Some(task) = queue.try_pop() {
if cancel_token.is_cancelled() {
break;
}
let permit = semaphore.clone().acquire_owned().await?;
if cancel_token.is_cancelled() {
break;
}
let client = client.clone();
let id = id.clone();
let local_manifest = manifest.clone();
let p1 = p1.clone();
let cancel_token = cancel_token.clone();
join_set.spawn(async move {
let guard = task;
@@ -185,6 +190,11 @@ pub async fn run_side(
let mut retry_conn = client.get_client().await?;
result = download(&mut retry_conn, &req, &p1).await;
}
if let Err(e) = &result
&& e.downcast_ref::<h2::Error>().is_some()
{
cancel_token.cancel();
}
result?;
local_manifest
+4 -1
View File
@@ -17,4 +17,7 @@ webpki-roots = { workspace = true }
anyhow = { workspace = true }
tokio-util = { workspace = true }
futures-util = { workspace = true }
futures = "0.3.32"
futures = { workspace = true }
tokio-socks = { workspace = true }
url = {workspace = true}
async-http-proxy = {workspace = true, features = ["tokio", "runtime-tokio"]}
+3 -1
View File
@@ -34,6 +34,7 @@ pub struct TcpClientTunnelConfig {
pub host: Option<String>,
pub url: String,
pub token: String,
pub proxy: Option<String>,
}
impl TcpClientTunnelConfig {
@@ -43,6 +44,7 @@ impl TcpClientTunnelConfig {
host: self.host,
url: self.url,
token: self.token,
proxy: self.proxy,
}
}
}
}
+91 -12
View File
@@ -1,10 +1,11 @@
use anyhow::anyhow;
use async_http_proxy::http_connect_tokio;
use bytes::Bytes;
use h2::{RecvStream, client, server};
use http::Request;
use log::{debug, error, info};
use serde::{Deserialize, Serialize};
use std::collections::{HashMap, HashSet};
use std::collections::HashMap;
use std::error::Error;
use std::fs::File;
use std::io::BufReader;
@@ -17,6 +18,8 @@ use tokio::time::{Duration, sleep, timeout};
use tokio_rustls::rustls;
use tokio_rustls::rustls::pki_types::{CertificateDer, ServerName};
use tokio_rustls::{TlsAcceptor, TlsConnector};
use tokio_socks::tcp::Socks5Stream;
use url::Url;
#[derive(Debug, Copy, Clone, PartialEq, Eq, Serialize, Deserialize)]
pub enum Identity {
@@ -39,6 +42,7 @@ pub struct ClientTunnelConfig {
pub url: String,
pub token: String,
pub identity: Identity,
pub proxy: Option<String>,
}
impl Default for ClientTunnelConfig {
@@ -48,6 +52,7 @@ impl Default for ClientTunnelConfig {
url: "127.0.0.1:3333".to_string(),
token: "super_secret_magic_token".to_string(),
identity: Identity::Client,
proxy: None,
}
}
}
@@ -95,6 +100,30 @@ pub enum TunnelEndpoint {
Server(Arc<ServerManager>),
}
#[derive(Clone)]
pub enum WeakTunnelEndpoint {
Client(std::sync::Weak<ClientManager>),
Server(std::sync::Weak<ServerManager>),
}
impl TunnelEndpoint {
pub fn downgrade(&self) -> WeakTunnelEndpoint {
match self {
Self::Client(c) => WeakTunnelEndpoint::Client(Arc::downgrade(c)),
Self::Server(s) => WeakTunnelEndpoint::Server(Arc::downgrade(s)),
}
}
}
impl WeakTunnelEndpoint {
pub fn upgrade(&self) -> Option<TunnelEndpoint> {
match self {
Self::Client(c) => c.upgrade().map(TunnelEndpoint::Client),
Self::Server(s) => s.upgrade().map(TunnelEndpoint::Server),
}
}
}
pub struct ClientManager {
pub session_id: AtomicU64,
pub current_client: Mutex<Option<client::SendRequest<Bytes>>>,
@@ -231,9 +260,9 @@ enum ResumeResult {
pub struct TunnelListener {
listener: TcpListener,
config: ServerTunnelConfig,
pending_plain_sessions: Mutex<HashSet<u64>>,
pending_plain_sessions: Mutex<HashMap<u64, std::time::Instant>>,
next_session_id: AtomicU64,
active_sessions: Mutex<HashMap<u64, TunnelEndpoint>>,
active_sessions: Mutex<HashMap<u64, WeakTunnelEndpoint>>,
}
impl TunnelListener {
@@ -243,7 +272,7 @@ impl TunnelListener {
Ok(Self {
listener,
config,
pending_plain_sessions: Mutex::new(HashSet::new()),
pending_plain_sessions: Mutex::new(HashMap::new()),
next_session_id: AtomicU64::new(1),
active_sessions: Mutex::new(HashMap::new()),
})
@@ -264,7 +293,9 @@ impl TunnelListener {
let mut stream = match self.try_resume_plain_session(stream, peer_addr).await? {
ResumeResult::NewSession(ep_raw, sid) => {
let ep = wrap_raw_endpoint(sid, ep_raw, None);
self.active_sessions.lock().await.insert(sid, ep.clone());
let mut sessions = self.active_sessions.lock().await;
sessions.retain(|_, weak_ep| weak_ep.upgrade().is_some());
sessions.insert(sid, ep.downgrade());
return Ok(ep);
}
ResumeResult::ResumedExisting => continue,
@@ -290,7 +321,9 @@ impl TunnelListener {
let sid = self.next_session_id.fetch_add(1, Ordering::Relaxed);
let ep = wrap_raw_endpoint(sid, ep_raw, None);
self.active_sessions.lock().await.insert(sid, ep.clone());
let mut sessions = self.active_sessions.lock().await;
sessions.retain(|_, weak_ep| weak_ep.upgrade().is_some());
sessions.insert(sid, ep.downgrade());
return Ok(ep);
}
}
@@ -320,7 +353,8 @@ impl TunnelListener {
let session_id = self.next_session_id.fetch_add(1, Ordering::Relaxed);
{
let mut pending = self.pending_plain_sessions.lock().await;
pending.insert(session_id);
pending.retain(|_, time| time.elapsed() < Duration::from_secs(30));
pending.insert(session_id, std::time::Instant::now());
}
tls_stream.write_all(TLS_BOOTSTRAP_MAGIC).await?;
@@ -351,7 +385,7 @@ impl TunnelListener {
let is_pending = {
let mut pending = self.pending_plain_sessions.lock().await;
pending.remove(&session_id)
pending.remove(&session_id).is_some()
};
if is_pending {
@@ -366,7 +400,11 @@ impl TunnelListener {
let ep_raw = upgrade_to_h2_raw(stream, self.config.identity)
.await
.map_err(|e| anyhow::anyhow!("{}", e))?;
update_endpoint(&ep, ep_raw).await;
update_endpoint(
&ep.upgrade().ok_or(anyhow!("Connection is cleared"))?,
ep_raw,
)
.await;
info!(
"[{}] Successfully resumed existing session {}",
peer_addr, session_id
@@ -542,6 +580,47 @@ pub async fn connect_tunnel(config: ClientTunnelConfig) -> Result<TunnelEndpoint
Ok(wrap_raw_endpoint(sid, ep_raw, Some(config)))
}
pub async fn connect_with_auto_proxy(config: &ClientTunnelConfig) -> anyhow::Result<TcpStream> {
let Some(proxy) = &config.proxy else {
return Ok(TcpStream::connect(&config.url).await?);
};
let parsed_proxy = Url::parse(proxy)?;
match parsed_proxy.scheme() {
"socks5" => {
let host = parsed_proxy.host_str().unwrap();
let port = parsed_proxy.port().unwrap_or(1080);
let stream = Socks5Stream::connect((host, port), config.url.clone())
.await?
.into_inner();
Ok(stream)
}
"http" | "https" => {
let proxy_addr = format!(
"{}:{}",
parsed_proxy.host_str().unwrap(),
parsed_proxy.port().unwrap_or(80)
);
let mut stream = TcpStream::connect(proxy_addr).await?;
let target_url = Url::parse(&format!("tcp://{}", config.url))?;
http_connect_tokio(
&mut stream,
target_url.host_str().unwrap(),
target_url.port().unwrap_or(80),
)
.await?;
Ok(stream)
}
_ => Err(anyhow::anyhow!(
"Unsupported proxy scheme: {}",
parsed_proxy.scheme()
)),
}
}
async fn do_client_reconnect(
config: &ClientTunnelConfig,
current_sid: &mut u64,
@@ -563,7 +642,7 @@ async fn do_client_reconnect(
*current_sid = sid;
resume_tunnel_client(config, sid).await
} else {
let mut stream = TcpStream::connect(&config.url).await?;
let mut stream = connect_with_auto_proxy(config).await?;
perform_client_handshake(&mut stream, &config.token, config.identity).await?;
let raw = upgrade_to_h2_raw(stream, config.identity).await?;
*current_sid = 0;
@@ -576,7 +655,7 @@ async fn bootstrap_tls_and_get_sid(
host: &str,
) -> anyhow::Result<(u64, ())> {
let connector = build_client_tls_connector();
let tcp = TcpStream::connect(&config.url).await?;
let tcp = connect_with_auto_proxy(config).await?;
let server_name = ServerName::try_from(host.to_string())
.map_err(|_| anyhow!("Invalid TLS host: {}", host))?
.to_owned();
@@ -604,7 +683,7 @@ async fn resume_tunnel_client(
config: &ClientTunnelConfig,
session_id: u64,
) -> anyhow::Result<TunnelEndpointRaw> {
let mut plain_stream = TcpStream::connect(&config.url).await?;
let mut plain_stream = connect_with_auto_proxy(config).await?;
plain_stream.write_all(RESUME_MAGIC).await?;
plain_stream.write_all(&session_id.to_be_bytes()).await?;
upgrade_to_h2_raw(plain_stream, config.identity).await
+3 -3
View File
@@ -82,7 +82,7 @@ profiles:
# 并发下载数(单个资源包的同时下载线程数)
concurrent: 50
# 精确匹配单文件包
# 展开单文件包
exact_single_file_bundle: true
# 导出基础路径
@@ -93,7 +93,7 @@ profiles:
# 启动应用时必须下载的资源类型
start_app:
- "thumbnail" # 缩略图
- "stamp" # 邮票/表情
- "stamp" # 表情
# - "area" # 地图
# - "home" # 主页背景
@@ -158,7 +158,7 @@ profiles:
# 并发下载数
concurrent: 50
# 精确匹配单文件包
# 展开单文件包
exact_single_file_bundle: true
# 导出路径
+1 -1
View File
@@ -108,7 +108,7 @@ concurrency:
# ACB 音频包解析的并发数
acb: 8
# USM 视频解析的并发数
# USM 视频解析的并发数(好像没用到)
usm: 4
# HCA 音频解码的并发数(通常可设置较大值)
+1 -1
View File
@@ -86,7 +86,7 @@ pub async fn download(
let _ = server_send_files(send_stream, &files).await;
}
let _ = std::fs::remove_file(dir);
let _ = std::fs::remove_dir_all(dir);
Ok(())
}