fix export path

Cargo.lock

@@ -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"

Cargo.toml

@@ -35,3 +35,7 @@ structopt = "0.3.26"
 tokio-util = "0.7.18"
 futures-util = "0.3.32"
 twox-hash = "2.1.2"
+futures = "0.3.32"
+tokio-socks = "0.5.2"
+url = "2.5.8"
+async-http-proxy = "1.2.5"

assets-updater/src/core/export_pipeline.rs

@@ -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()
+    let output_dir = empty_dir(
+        std::env::temp_dir()
             .join("sekai-updater")
             .join("extract")
-        .join(&sync_context.region)
+            .join(&sync_context.region),
-        .join(hash);
+        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 {
+    let mut out: Vec<PathBuf> = vec![];
-        usm_files[0].clone()
-    } else {
-        merge_usm_files(export_path, &usm_files)?
-    };
 
-    process_usm_file(
+    for f in usm_files {
-        &usm_input,
+        out.append(&mut process_usm_file(&f, sync_context, ffmpeg_path, retry).await?);
-        export_path,
+    }
-        sync_context,
+    Ok(out)
-        region,
-        ffmpeg_path,
-        retry,
-    )
-    .await
 }
 
 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| {

client/src/main.rs

@@ -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)) => {

client/src/queue.rs

@@ -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();

client/src/task.rs

@@ -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

communicator/Cargo.toml

@@ -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"]}

communicator/src/config.rs

@@ -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,
         }
     }
 }

communicator/src/stream.rs

@@ -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

sekai-unpacker-client.example.yaml

@@ -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
 
     # 导出路径

sekai-unpacker-server.example.yaml

@@ -108,7 +108,7 @@ concurrency:
   # ACB 音频包解析的并发数
   acb: 8
 
-  # USM 视频解析的并发数
+  # USM 视频解析的并发数（好像没用到）
   usm: 4
 
   # HCA 音频解码的并发数（通常可设置较大值）

server/src/router/download.rs

@@ -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(())
 }