diff --git a/server-rs/Cargo.lock b/server-rs/Cargo.lock index 89c3e8184..b09dbb6f8 100644 --- a/server-rs/Cargo.lock +++ b/server-rs/Cargo.lock @@ -4159,6 +4159,8 @@ dependencies = [ name = "platform-tripo" version = "0.1.0" dependencies = [ + "bytes", + "reqwest", "serde_json", "shared-contracts", "tokio", diff --git a/server-rs/crates/platform-tripo/Cargo.toml b/server-rs/crates/platform-tripo/Cargo.toml index 5aa6ac0bd..4196df708 100644 --- a/server-rs/crates/platform-tripo/Cargo.toml +++ b/server-rs/crates/platform-tripo/Cargo.toml @@ -8,7 +8,10 @@ license.workspace = true shared-contracts = { workspace = true } serde_json = { workspace = true } tripo3d-sdk = { workspace = true } +bytes = { workspace = true } +reqwest = { workspace = true, features = ["rustls-tls", "stream"] } +tokio = { workspace = true, features = ["time"] } url = { workspace = true } [dev-dependencies] -tokio = { workspace = true, features = ["macros", "rt-multi-thread"] } +tokio = { workspace = true, features = ["fs", "io-util", "macros", "rt-multi-thread"] } diff --git a/server-rs/crates/platform-tripo/examples/tripo_generation_smoke.rs b/server-rs/crates/platform-tripo/examples/tripo_generation_smoke.rs index a353c5da5..d9127b85a 100644 --- a/server-rs/crates/platform-tripo/examples/tripo_generation_smoke.rs +++ b/server-rs/crates/platform-tripo/examples/tripo_generation_smoke.rs @@ -12,7 +12,7 @@ //! output, and writes the downloaded model into the current directory. The //! SDK poller is used only here and is not exposed by `platform-tripo`. -use std::{env, fs, time::Duration}; +use std::{env, time::Duration}; use platform_tripo::TripoProviderClient; use shared_contracts::model3d::{ @@ -21,6 +21,7 @@ use shared_contracts::model3d::{ multiview_to_model::{Model3dMultiviewInputs, Model3dMultiviewToModelRequest}, text_to_model::Model3dTextToModelRequest, }; +use tokio::{fs::File, io::AsyncWriteExt}; use tripo3d_sdk::{ClientOptions, TripoClient, WaitOptions}; const SAMPLE_IMAGE_URL: &str = "https://www.rustacean.net/assets/rustacean-flat-happy.png"; @@ -164,11 +165,17 @@ async fn main() -> Result<(), Box> { let downloaded = client.download_model(&snapshot).await?; let filename = downloaded.filename(name); - fs::write(&filename, &downloaded.data)?; + let mut file = File::create(&filename).await?; + let mut bytes = 0u64; + let mut downloaded = downloaded; + while let Some(chunk) = downloaded.next_chunk().await? { + bytes += chunk.len() as u64; + file.write_all(&chunk).await?; + } println!( "{name} artifact={} bytes={} url={}", filename, - downloaded.data.len(), + bytes, downloaded.url.redacted() ); } diff --git a/server-rs/crates/platform-tripo/src/common/client.rs b/server-rs/crates/platform-tripo/src/common/client.rs index bec0f5eb0..e4d03bf75 100644 --- a/server-rs/crates/platform-tripo/src/common/client.rs +++ b/server-rs/crates/platform-tripo/src/common/client.rs @@ -1,3 +1,5 @@ +use std::time::Duration; + use tripo3d_sdk::TripoClient; use super::{ @@ -7,12 +9,24 @@ use super::{ pub struct TripoProviderClient { pub(crate) client: TripoClient, + artifact_client: reqwest::Client, + artifact_retries: u32, } impl TripoProviderClient { pub fn new(settings: TripoSettings) -> Result { + let artifact_client = reqwest::Client::builder() + .user_agent(settings.user_agent.clone()) + .timeout(settings.request_timeout) + .build() + .map_err(|error| TripoError::Request { + message: format!("failed to build artifact download client: {error}"), + status: None, + })?; Ok(Self { client: TripoClient::new(settings.client_options()).map_err(TripoError::from)?, + artifact_client, + artifact_retries: settings.retries, }) } @@ -35,33 +49,91 @@ impl TripoProviderClient { ) -> Result { let handle = &task.handle; validate_task_id(&handle.task_id)?; - // TODO SDK 只公开 download_model(&Task)，必须按 handle 再取一次 task；若将来支持按 URL 下载， - // 可直接使用 snapshot.output 里的模型 URL，省掉这次请求。 - let sdk_task = self - .client - .get_task(&handle.task_id) - .await - .map_err(TripoError::from)?; - let downloaded = self - .client - .download_model(&sdk_task) - .await - .map_err(TripoError::from)? + let output = task + .output + .as_ref() .ok_or_else(|| TripoError::OutputSchema { task_id: handle.task_id.clone(), - message: "completed task has no model URL".into(), + message: "task has no completed model output".into(), })?; - let url = TripoUrl::parse(&downloaded.url).map_err(|error| match error { - TripoError::OutputSchema { message, .. } => TripoError::OutputSchema { - task_id: handle.task_id.clone(), - message, - }, - other => other, - })?; - Ok(TripoDownloadedModel { - url, - content_type: downloaded.content_type, - data: downloaded.data, - }) + // TODO SDK upstream: expose a streaming artifact API; then replace this + // provider-side reqwest client with the SDK stream and remove the duplicate downloader. + self.download_artifact(&handle.task_id, output.model_url()) + .await + } + + async fn download_artifact( + &self, + task_id: &str, + url: &TripoUrl, + ) -> Result { + let total_attempts = self.artifact_retries.saturating_add(1); + let mut last_error = None; + + for attempt in 1..=total_attempts { + match self.download_artifact_once(task_id, url).await { + Ok(downloaded) => return Ok(downloaded), + Err(error) if attempt < total_attempts && error.is_retryable() => { + last_error = Some(error); + tokio::time::sleep(download_backoff(attempt)).await; + } + Err(error) => return Err(error), + } + } + + Err(last_error.expect("artifact download has at least one attempt")) + } + + async fn download_artifact_once( + &self, + task_id: &str, + url: &TripoUrl, + ) -> Result { + let response = self + .artifact_client + .get(url.as_str()) + .send() + .await + .map_err(|error| TripoError::Request { + message: format!( + "artifact download transport failure for task {task_id}: {}", + transport_error_kind(&error) + ), + status: None, + })?; + let status = response.status(); + if !status.is_success() { + return Err(TripoError::Request { + message: format!("artifact download failed for task {task_id}"), + status: Some(status.as_u16()), + }); + } + + let content_type = response + .headers() + .get(reqwest::header::CONTENT_TYPE) + .and_then(|value| value.to_str().ok()) + .map(str::to_owned); + let content_length = response.content_length(); + Ok(TripoDownloadedModel::new( + url.clone(), + content_type, + content_length, + response, + )) + } +} + +fn download_backoff(attempt: u32) -> Duration { + Duration::from_millis(250u64.saturating_mul(2u64.saturating_pow(attempt.min(6)))) +} + +fn transport_error_kind(error: &reqwest::Error) -> &'static str { + if error.is_timeout() { + "timeout" + } else if error.is_connect() { + "connectivity error" + } else { + "request error" } } diff --git a/server-rs/crates/platform-tripo/src/common/types.rs b/server-rs/crates/platform-tripo/src/common/types.rs index aef0b6e57..c639b8717 100644 --- a/server-rs/crates/platform-tripo/src/common/types.rs +++ b/server-rs/crates/platform-tripo/src/common/types.rs @@ -1,5 +1,8 @@ use std::fmt; +use bytes::Bytes; +use reqwest::Response; + use url::Url; use shared_contracts::model3d::common::Model3dTaskStatus; @@ -74,6 +77,16 @@ pub enum TripoTaskOutput { MultiviewToModel(TripoMultiviewToModelResult), } +impl TripoTaskOutput { + pub(crate) fn model_url(&self) -> &TripoUrl { + match self { + Self::TextToModel(result) => &result.model_url, + Self::ImageToModel(result) => &result.model_url, + Self::MultiviewToModel(result) => &result.model_url, + } + } +} + #[derive(Clone, Debug, Eq, PartialEq)] pub struct TripoTaskSnapshot { pub handle: TripoTaskHandle, @@ -86,14 +99,63 @@ pub struct TripoTaskSnapshot { pub completed_at: Option, } -#[derive(Clone, Debug, Eq, PartialEq)] pub struct TripoDownloadedModel { pub url: TripoUrl, pub content_type: Option, - pub data: Vec, + pub content_length: Option, + response: Response, + status: u16, + received: u64, } impl TripoDownloadedModel { + pub(crate) fn new( + url: TripoUrl, + content_type: Option, + content_length: Option, + response: Response, + ) -> Self { + Self { + url, + content_type, + content_length, + status: response.status().as_u16(), + response, + received: 0, + } + } + + pub async fn next_chunk(&mut self) -> Result, TripoError> { + let chunk = self + .response + .chunk() + .await + .map_err(|error| TripoError::Request { + message: format!("failed to read artifact response body: {error}"), + status: Some(self.status), + })?; + match chunk { + Some(chunk) => { + self.received += chunk.len() as u64; + Ok(Some(chunk)) + } + None => { + if let Some(expected) = self.content_length { + if self.received != expected { + return Err(TripoError::Request { + message: format!( + "artifact length mismatch: expected {expected} bytes, received {}", + self.received + ), + status: Some(self.status), + }); + } + } + Ok(None) + } + } + } + pub fn filename(&self, name: &str) -> String { let extension = self .url