image-gen: file upload client, enqueue/snapshot endpoints

This commit is contained in:
Kasisnu committed 2025-04-03 20:16:18 +05:30
1 parent e622dc17e3
commit f3b4ce1204
23 files changed
+1014 -18

No files matched your search

Generated
+2
View File
@@ -9722,6 +9722,7 @@ dependencies = [
"js-sys",
"log",
"mime",
"mime_guess",
"native-tls",
"once_cell",
"percent-encoding",
@@ -11293,6 +11294,7 @@ dependencies = [
"mysql_queries",
"native-tls",
"once_cell",
"openai_sora_client",
"password",
"pin-project-lite",
"primitives",
@@ -24,12 +24,12 @@ subprocess_common = { path = "../../subprocess_common" }
# External
anyhow = { workspace = true }
log = { workspace = true }
reqwest = { workspace = true }
reqwest = { workspace = true, features = ["multipart"] }
serde = { workspace = true }
serde_derive = { workspace = true }
serde_json = { workspace = true }
tokio = { version = "1.38.0", features = ["macros"] }
[dev-dependencies]
tokio = { version = "1.38.0", features = ["macros"] }
testing = { path = "../../testing" }
filesys = { path = "../../files/filesys" }
@@ -1,5 +1,5 @@
use reqwest::RequestBuilder;
use errors::AnyhowResult;
// NB: It appears that the sentinel may require a matching user agent.
const USER_AGENT : &str = "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/127.0.0.0 Safari/537.36";
@@ -7,7 +7,7 @@ const USER_AGENT : &str = "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit
pub struct SoraCredentials {
pub bearer_token: String,
pub cookie: String,
pub sentinel: String,
pub sentinel: Option<String>,
}
impl SoraCredentials {
@@ -18,19 +18,30 @@ impl SoraCredentials {
}
}
pub fn from_env() -> AnyhowResult<Self> {
let bearer = std::env::var("SORA_BEARER_TOKEN")?;
let cookie = std::env::var("SORA_COOKIE")?;
let sentinel = std::env::var("SORA_SENTINEL").ok();
Ok(Self { bearer_token: bearer, cookie, sentinel })
}
pub fn add_credential_headers_to_request(&self, request: RequestBuilder) -> RequestBuilder {
let bearer_header = self.authorization_header_value();
println!(">>> BEARER HEADER = {}", bearer_header);
//println!(">>> COOKIE HEADER = {}", self.cookie);
println!(">>> SENTINEL HEADER = {}", self.sentinel);
println!(">>> SENTINEL HEADER = {}", self.sentinel.clone().unwrap_or("None".to_string()));
let request = request
.header("OpenAI-Sentinel-Token", &self.sentinel)
let mut request = request
.header("User-Agent", USER_AGENT)
.header("Cookie", &self.cookie)
.header("Authorization", bearer_header);
if let Some(sentinel) = &self.sentinel {
request = request.header("OpenAI-Sentinel-Token", sentinel);
}
/*
Without sentinel token:
status: "{\n \"error\": {\n \"message\": \"Hmmm something didn't look right with your request. Please try again later or visit https://help.openai.com if this issue persists.\",\n \"type\": \"invalid_request_error\",\n \"param\": null,\n \"code\": \"sentinel_block\"\n }\n}"
@@ -106,13 +106,16 @@ pub (crate) async fn call_sora_image_gen(sora_request: RawSoraImageGenRequest, c
//let request = credentials.add_credential_headers_to_request(request);
let request= client.post(URL)
.header("OpenAI-Sentinel-Token", &credentials.sentinel)
let mut request = client.post(URL)
.header("User-Agent", USER_AGENT)
.header("Cookie", &credentials.cookie)
.header("Authorization", credentials.authorization_header_value())
.header("Content-Type", "application/json");
if let Some(sentinel) = &credentials.sentinel {
request = request.header("OpenAI-Sentinel-Token", sentinel);
}
let request = request.json(&sora_request).build()?;
println!("\n\n >>> raw request = {:?}\n\n", request);
@@ -54,7 +54,7 @@ mod tests {
let creds = SoraCredentials {
bearer_token: bearer,
cookie,
sentinel,
sentinel: Some(sentinel),
};
let response = sora_image_gen_simple(SoraImageGenSimpleRequest {
@@ -52,7 +52,7 @@ mod tests {
let creds = SoraCredentials {
bearer_token: bearer,
cookie,
sentinel,
sentinel: Some(sentinel),
};
let task_id = "task_01jqwwrkvgeqp8jsf5mqk1jceh";
@@ -1,4 +1,4 @@
mod credentials;
pub mod credentials;
pub mod image_gen;
pub mod job;
pub mod upload;
@@ -1 +1,2 @@
pub mod sora_image_upload;
pub mod sora_image_upload;
pub mod sora_media_upload;
@@ -0,0 +1,116 @@
use errors::AnyhowResult;
use std::path::Path;
use reqwest::multipart::{Form, Part};
use reqwest::Client;
use serde::Deserialize;
use tokio::fs::File;
use tokio::io::AsyncReadExt;
use crate::credentials::SoraCredentials;
const URL: &str = "https://sora.com/backend/uploads";
#[derive(Debug, Deserialize)]
pub struct SoraMediaUploadResponse {
pub id: String,
pub r#type: String,
pub created_at: String,
pub filename: String,
pub extension: String,
pub mime_type: String,
pub url: String,
pub width: Option<u32>,
pub height: Option<u32>,
pub duration_sec: Option<f64>,
pub n_frames: Option<u32>,
pub size_bytes: u64,
pub thumbnail_url: Option<String>,
}
pub struct SoraMediaUploadRequest<'a> {
pub file_path: String,
pub credentials: &'a SoraCredentials,
}
pub async fn sora_media_upload(request: SoraMediaUploadRequest<'_>) -> AnyhowResult<SoraMediaUploadResponse> {
let client = Client::new();
let file_path = Path::new(&request.file_path);
let filename = file_path.file_name()
.ok_or_else(|| anyhow::anyhow!("Invalid file path"))?
.to_string_lossy()
.to_string();
// Read file data
let mut file = File::open(&file_path).await?;
let mut buffer = Vec::new();
file.read_to_end(&mut buffer).await?;
let mime_type = match file_path.extension().and_then(|e| e.to_str()) {
Some("jpg") | Some("jpeg") => "image/jpeg",
Some("png") => "image/png",
// Some("webp") => "image/webp",
// Some("gif") => "image/gif",
// Some("mp4") => "video/mp4",
// Some("mov") => "video/quicktime",
// Some("webm") => "video/webm",
_ => "application/octet-stream",
};
// Create multipart form
let part = Part::bytes(buffer)
.file_name(filename)
.mime_str(mime_type)?;
let form = Form::new().part("file", part);
// Make API request
let request_builder = client.post(URL).multipart(form);
let request_builder = request.credentials.add_credential_headers_to_request(request_builder);
let response = request_builder.send().await?;
// Check response status
if !response.status().is_success() {
let status = response.status();
let text = response.text().await?;
return Err(anyhow::anyhow!("Upload failed with status {}: {}", status, text));
}
// Parse response
let upload_response = response.json::<SoraMediaUploadResponse>().await?;
Ok(upload_response)
}
#[cfg(test)]
mod tests {
use std::fs::read_to_string;
use errors::AnyhowResult;
use testing::test_file_path::test_file_path;
use crate::credentials::SoraCredentials;
use crate::upload::sora_media_upload::{sora_media_upload, SoraMediaUploadRequest};
// #[ignore]
#[tokio::test]
pub async fn test() -> AnyhowResult<()> {
let cookie = read_to_string(test_file_path("test_data/test_data/temp/cookie.txt")?)?;
let cookie = cookie.trim().to_string();
let bearer = read_to_string(test_file_path("test_data/test_data/temp/bearer.txt")?)?;
let bearer = bearer.trim().to_string();
let creds = SoraCredentials {
bearer_token: bearer,
cookie,
sentinel: None,
};
let response = sora_media_upload(SoraMediaUploadRequest {
file_path: "/Users/kasisnu/Downloads/puppies-test-image.png".to_string(),
credentials: &creds,
}).await?;
println!("media_id: {}", response.id);
println!("media_url: {}", response.url);
assert!(response.id.starts_with("media_"));
Ok(())
}
}
@@ -1,7 +1,7 @@
use crate::payloads::generic_inference_args::inner_payloads::f5_tts_payload::F5TTSPayload;
use crate::payloads::generic_inference_args::inner_payloads::face_fusion_payload::FaceFusionPayload;
use crate::payloads::generic_inference_args::inner_payloads::gptsovits_payload::GptSovitsPayload;
use crate::payloads::generic_inference_args::inner_payloads::image_generation_payload::StableDiffusionArgs;
use crate::payloads::generic_inference_args::inner_payloads::image_generation_payload::{SoraImageGenArgs, StableDiffusionArgs};
use crate::payloads::generic_inference_args::inner_payloads::lipsync_payload::LipsyncArgs;
use crate::payloads::generic_inference_args::inner_payloads::live_portrait_payload::LivePortraitPayload;
use crate::payloads::generic_inference_args::inner_payloads::mocap_payload::MocapArgs;
@@ -111,7 +111,7 @@ pub enum PolymorphicInferenceArgs {
Tts(TTSArgs),
// No arguments yet.
// It might be best to just not include this when not used.
/// Voice conversion. (Short name to save space when serializing.)
Vc {
/// Argument for so-vits-svc
@@ -143,6 +143,10 @@ pub enum PolymorphicInferenceArgs {
/// Image generation. (Short name to save space when serializing.)
Ig(StableDiffusionArgs),
/// Sora Image Generation. (Short name to save space when serializing.)
Sg(SoraImageGenArgs),
/// Mocap (Short name to save space when serializing.)
Mc(MocapArgs),
@@ -1,3 +1,4 @@
use tokens::tokens::media_files::MediaFileToken;
use tokens::tokens::model_weights::ModelWeightToken;
/// Video sources can be one of several:
@@ -52,7 +53,7 @@ pub struct StableDiffusionArgs {
#[serde(rename = "se")]
#[serde(skip_serializing_if = "Option::is_none")]
pub maybe_seed: Option<i64>,
#[serde(rename = "mu")]
#[serde(skip_serializing_if = "Option::is_none")]
pub maybe_upload_path: Option<String>,
@@ -60,7 +61,7 @@ pub struct StableDiffusionArgs {
#[serde(rename = "cf")]
#[serde(skip_serializing_if = "Option::is_none")]
pub maybe_cfg_scale: Option<u32>,
#[serde(rename = "lu")]
#[serde(skip_serializing_if = "Option::is_none")]
pub maybe_lora_upload_path: Option<String>,
@@ -90,3 +91,32 @@ pub struct StableDiffusionArgs {
}
#[derive(Clone, Debug, Default, Serialize, Deserialize, PartialEq, Eq)]
pub struct SoraImageGenArgs {
#[serde(rename = "p")]
#[serde(skip_serializing_if = "Option::is_none")]
pub prompt: Option<String>,
#[serde(rename = "sm")]
pub scene_media_token: Option<MediaFileToken>,
#[serde(rename = "mf")]
#[serde(skip_serializing_if = "Option::is_none")]
pub maybe_additional_media_file_tokens: Option<Vec<MediaFileToken>>,
#[serde(rename = "sa")]
#[serde(skip_serializing_if = "Option::is_none")]
pub maybe_number_of_samples: Option<u32>,
#[serde(rename = "su")]
#[serde(skip_serializing_if = "Option::is_none")]
pub maybe_sora_media_upload_tokens: Option<Vec<String>>,
#[serde(rename = "ss")]
#[serde(skip_serializing_if = "Option::is_none")]
pub maybe_sora_scene_media_token: Option<String>,
}
+2
View File
@@ -17,7 +17,9 @@ pub fn test_file_path<P: AsRef<Path>>(path_from_repo_root: P) -> AnyhowResult<Pa
.to_str()
.ok_or(anyhow!("path cannot convert to string"))?;
path.push(format!("../../../{}", file));
println!(" >>> path = {:?}", path);
Ok(path.canonicalize()?)
}
#[cfg(test)]
@@ -69,6 +69,7 @@ tts_common = { path = "../../../lib/tts_common" }
url_config = { path = "../../../lib/url_config" }
user_input_common = { path = "../../../lib/user_input_common" }
videos = { path = "../../../lib/files/videos" }
openai_sora_client = { path = "../../../lib/api_clients/openai_sora_client" }
# Internal Components
billing_component = { path = "../../plugins/billing" }
@@ -507,6 +507,9 @@ fn validate_request(
Ok(())
}
fn inference_mode_from_http_request(http_request: &HttpRequest) -> Option<TypeOfInference> {
inference_mode_from_url_path(http_request.path())
}
@@ -0,0 +1,2 @@
pub mod upload;
pub mod prompt;
@@ -0,0 +1,489 @@
#![forbid(unused_mut)]
use std::fmt::Debug;
use std::fmt::{Display, Formatter};
use std::path::{Path, PathBuf};
use std::sync::Arc;
use actix_web::error::ResponseError;
use actix_web::http::StatusCode;
use actix_web::web::Json;
use actix_web::{web, HttpRequest, HttpResponse};
use bucket_paths::legacy::typified_paths::public::media_files::bucket_file_path::MediaFileBucketPath;
use cloud_storage::bucket_client::BucketClient;
use log::error;
use log::warn;
use mysql_queries::queries::media_files::get::get_media_file::get_media_file;
use openai_sora_client::upload::sora_media_upload::{sora_media_upload, SoraMediaUploadRequest};
use openai_sora_client::credentials::SoraCredentials;
use serde::Deserialize;
use serde::Serialize;
use sqlx::MySqlPool;
use utoipa::ToSchema;
use web::Data;
use enums::by_table::generic_inference_jobs::inference_category::InferenceCategory;
use enums::by_table::generic_inference_jobs::inference_job_type::InferenceJobType;
use enums::by_table::generic_inference_jobs::inference_model_type::InferenceModelType;
use http_server_common::request::get_request_header_optional::get_request_header_optional;
use http_server_common::request::get_request_ip::get_request_ip;
use mysql_queries::payloads::generic_inference_args::generic_inference_args::{
GenericInferenceArgs,
InferenceCategoryAbbreviated,
PolymorphicInferenceArgs,
};
use mysql_queries::payloads::generic_inference_args::inner_payloads::image_generation_payload::SoraImageGenArgs;
use mysql_queries::queries::generic_inference::web::insert_generic_inference_job::{
insert_generic_inference_job,
InsertGenericInferenceArgs,
};
use mysql_queries::queries::idepotency_tokens::insert_idempotency_token::insert_idempotency_token;
use tokens::tokens::generic_inference_jobs::InferenceJobToken;
use tokens::tokens::media_files::MediaFileToken;
use tokens::tokens::model_weights::ModelWeightToken;
use tokens::tokens::users::UserToken;
use crate::configs::plans::get_correct_plan_for_session::get_correct_plan_for_session;
use crate::http_server::validations::validate_idempotency_token_format::validate_idempotency_token_format;
use crate::http_server::web_utils::response_error_helpers::to_simple_json_error;
use crate::state::server_state::ServerState;
use crate::util::allowed_studio_access::allowed_studio_access;
/// This is the number of images (batch size) to generate for each request.
/// We should allow all users to have multiple images generated at once as this
/// is what other providers do.
const MINIMUM_IMAGE_COUNT : u32 = 1;
/// The maximum number of images (batch size) to generate for each request.
const MAXIMUM_IMAGE_COUNT : u32 = 2;
/// Debug requests can get routed to special "debug-only" workers, which can
/// be used to trial new code, run debugging, etc.
const DEBUG_HEADER_NAME: &str = "enable-debug-mode";
/// The routing tag header can send workloads to particular k8s hosts.
/// This is useful for catching the live logs or intercepting the job.
const ROUTING_TAG_HEADER_NAME: &str = "routing-tag";
#[derive(Deserialize, ToSchema)]
pub struct EnqueueStudioImageGenRequest {
pub uuid_idempotency_token: String,
pub scene_media_token: MediaFileToken,
pub prompt: String,
pub maybe_additional_images: Option<Vec<MediaFileToken>>,
pub maybe_number_of_samples: Option<u32>,
}
#[derive(Serialize, ToSchema)]
pub struct EnqueueImageGenRequestSuccessResponse {
pub success: bool,
pub inference_job_token: InferenceJobToken,
}
#[derive(Debug, ToSchema)]
pub enum EnqueueImageGenRequestError {
BadInput(String),
NotAuthorized,
ServerError,
RateLimited,
}
impl ResponseError for EnqueueImageGenRequestError {
fn status_code(&self) -> StatusCode {
match *self {
EnqueueImageGenRequestError::BadInput(_) => StatusCode::BAD_REQUEST,
EnqueueImageGenRequestError::NotAuthorized => StatusCode::UNAUTHORIZED,
EnqueueImageGenRequestError::ServerError => StatusCode::INTERNAL_SERVER_ERROR,
EnqueueImageGenRequestError::RateLimited => StatusCode::TOO_MANY_REQUESTS,
}
}
fn error_response(&self) -> HttpResponse {
let error_reason = match self {
EnqueueImageGenRequestError::BadInput(reason) => reason.to_string(),
EnqueueImageGenRequestError::NotAuthorized => "unauthorized".to_string(),
EnqueueImageGenRequestError::ServerError => "server error".to_string(),
EnqueueImageGenRequestError::RateLimited => "rate limited".to_string(),
};
to_simple_json_error(&error_reason, self.status_code())
}
}
impl Display for EnqueueImageGenRequestError {
fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result {
write!(f, "{:?}", self)
}
}
// Implementation for enqueuing a TTS request
// Reference enqueue_infer_tts_handler.rs for checks: rate limiting / user sessions
// insert generic inference job.rs
// Need to convert it to generic inference job.
#[utoipa::path(
post,
path = "/v1/image_studio/prompt",
responses(
(
status = 200,
description = "Enqueue TTS generically",
body = EnqueueImageGenRequestSuccessResponse,
),
(status = 400, description = "Bad input", body = EnqueueImageGenRequestError),
(status = 401, description = "Not authorized", body = EnqueueImageGenRequestError),
(status = 429, description = "Rate limited", body = EnqueueImageGenRequestError),
(status = 500, description = "Server error", body = EnqueueImageGenRequestError)
),
params(("request" = EnqueueStudioImageGenRequest, description = "Payload for Image Generation Request"))
)]
pub async fn enqueue_studio_image_generation_request(
http_request: HttpRequest,
request: Json<EnqueueStudioImageGenRequest>,
server_state: Data<Arc<ServerState>>
) -> Result<HttpResponse, EnqueueImageGenRequestError> {
validate_request(&request)?;
let mut maybe_user_token: Option<UserToken> = None;
let visbility = enums::common::visibility::Visibility::Public;
let mut mysql_connection = server_state.mysql_pool.acquire().await.map_err(|err| {
warn!("MySql pool error: {:?}", err);
EnqueueImageGenRequestError::ServerError
})?;
// ==================== USER SESSION ==================== //
let maybe_user_session = server_state.session_checker
.maybe_get_user_session_extended_from_connection(&http_request, &mut mysql_connection).await
.map_err(|e| {
warn!("Session checker error: {:?}", e);
EnqueueImageGenRequestError::ServerError
})?;
if let Some(user_session) = maybe_user_session.as_ref() {
maybe_user_token = Some(UserToken::new_from_str(&user_session.user_token));
}
// ==================== PLANS ==================== //
// Plan should handle "first anonymous use" and "investor" cases.
let plan = get_correct_plan_for_session(
server_state.server_environment_old,
maybe_user_session.as_ref()
);
// Separate priority for animation.
let priority_level = plan.web_vc_base_priority_level();
// ==================== DEBUG MODE + ROUTING TAG ==================== //
let is_debug_request = get_request_header_optional(&http_request, DEBUG_HEADER_NAME).is_some();
let maybe_routing_tag = get_request_header_optional(&http_request, ROUTING_TAG_HEADER_NAME).map(
|routing_tag| routing_tag.trim().to_string()
);
// ==================== BANNED USERS ==================== //
if let Some(ref user) = maybe_user_session {
if user.role.is_banned {
return Err(EnqueueImageGenRequestError::NotAuthorized);
}
}
// DETECT premium user and queue
// ==================== RATE LIMIT ==================== //
let rate_limiter = match maybe_user_session {
None => &server_state.redis_rate_limiters.logged_out,
Some(ref _user) => &server_state.redis_rate_limiters.logged_in,
};
if let Err(_err) = rate_limiter.rate_limit_request(&http_request) {
return Err(EnqueueImageGenRequestError::RateLimited);
}
// Get up IP address
let ip_address = get_request_ip(&http_request);
// ==================== HANDLE IDEMPOTENCY ==================== //
if let Err(reason) = validate_idempotency_token_format(&request.uuid_idempotency_token) {
return Err(EnqueueImageGenRequestError::BadInput(reason));
}
insert_idempotency_token(&request.uuid_idempotency_token, &mut *mysql_connection)
.await
.map_err(|err| {
error!("Error inserting idempotency token: {:?}", err);
EnqueueImageGenRequestError::BadInput("invalid idempotency token".to_string())
})?;
// ==================== INFERENCE ARGS ==================== //
let work_temp_dir = server_state.temp_dir_creator.new_tempdir("image_studio").unwrap();
let public_bucket_client = server_state.public_bucket_client.clone();
let private_bucket_client = server_state.private_bucket_client.clone();
let number_of_samples = match request.maybe_number_of_samples {
Some(val) => if val > MAXIMUM_IMAGE_COUNT {
MAXIMUM_IMAGE_COUNT
} else if val < MINIMUM_IMAGE_COUNT {
MINIMUM_IMAGE_COUNT
} else {
val
},
None => 1,
};
let additional_images = request.maybe_additional_images.clone().unwrap_or_else(|| vec![]);
let prompt = request.prompt.clone();
let sora_credentials = match SoraCredentials::from_env() {
Ok(creds) => creds,
Err(err) => {
error!("Failed to load Sora credentials from environment: {:?}", err);
return Err(EnqueueImageGenRequestError::ServerError);
}
};
let scene_media_token = request.scene_media_token.clone();
let scene_media_name = scene_media_token.to_string();
let scene_media_path = work_temp_dir.path().join(scene_media_name.clone());
let scene_media_path = do_download_media_file(
&scene_media_name,
&scene_media_token,
&scene_media_path,
&public_bucket_client,
&server_state.mysql_pool
).await.map_err(|err| {
error!("Failed to download scene media file: {:?}", err);
EnqueueImageGenRequestError::ServerError
})?;
let scene_media_upload_request = SoraMediaUploadRequest {
file_path: scene_media_path.to_string_lossy().to_string(),
credentials: &sora_credentials,
};
let scene_media_upload_response = sora_media_upload(scene_media_upload_request).await.map_err(|err| {
error!("Failed to upload scene media to Sora: {:?}", err);
EnqueueImageGenRequestError::ServerError
})?;
let sora_scene_media_id = scene_media_upload_response.id;
log::info!("Uploaded scene media to Sora with ID: {}", sora_scene_media_id);
let mut sora_additional_media_ids: Vec<String> = vec![];
for media_file_token in &additional_images {
let media_file_name = media_file_token.to_string();
let media_file_path = work_temp_dir.path().join(media_file_name.clone());
let media_file_path = do_download_media_file(
&media_file_name,
&media_file_token,
&media_file_path,
&public_bucket_client,
&server_state.mysql_pool
).await.map_err(|err| {
error!("Failed to download additional media file: {:?}", err);
EnqueueImageGenRequestError::ServerError
})?;
let sora_media_upload_request = SoraMediaUploadRequest {
file_path: media_file_path.to_string_lossy().to_string(),
credentials: &sora_credentials,
};
let sora_media_upload_response = sora_media_upload(sora_media_upload_request).await.map_err(|err| {
error!("Failed to upload additional media to Sora: {:?}", err);
EnqueueImageGenRequestError::ServerError
})?;
let sora_media_id = sora_media_upload_response.id;
log::info!("Uploaded additional media to Sora with ID: {}", sora_media_id);
sora_additional_media_ids.push(sora_media_id);
}
// Store the actual inference args that will go to the database
let inference_args = SoraImageGenArgs {
prompt: Some(prompt),
scene_media_token: Some(request.scene_media_token.clone()),
maybe_additional_media_file_tokens: Some(additional_images),
maybe_number_of_samples: Some(number_of_samples),
maybe_sora_scene_media_token: Some(sora_scene_media_id.clone()),
maybe_sora_media_upload_tokens: Some(sora_additional_media_ids),
};
// The Sora API IDs will be used later by the worker from environment variables
log::info!("Sora scene media ID: {}", sora_scene_media_id);
// create the inference args here
let maybe_avt_token = server_state.avt_cookie_manager.get_avt_token_from_request(&http_request);
// create the job record here!
let query_result = insert_generic_inference_job(InsertGenericInferenceArgs {
uuid_idempotency_token: &request.uuid_idempotency_token,
job_type: InferenceJobType::StableDiffusion,
maybe_product_category: None, // This is not a product anymore
inference_category: InferenceCategory::ImageGeneration,
maybe_model_type: Some(InferenceModelType::StableDiffusion), // NB: Model is static during inference
maybe_model_token: None, // NB: Model is static during inference
maybe_input_source_token: None,
maybe_input_source_token_type: None,
maybe_download_url: None,
maybe_cover_image_media_file_token: None,
maybe_raw_inference_text: None,
maybe_max_duration_seconds: None,
maybe_inference_args: Some(GenericInferenceArgs {
inference_category: Some(InferenceCategoryAbbreviated::ImageGeneration),
args: Some(PolymorphicInferenceArgs::Sg(inference_args)),
}),
maybe_creator_user_token: maybe_user_token.as_ref(),
maybe_avt_token: maybe_avt_token.as_ref(),
creator_ip_address: &ip_address,
creator_set_visibility: visbility,
priority_level,
requires_keepalive: false, //reverse ... TODO fix this. we set it base on account is premium or not ...
is_debug_request,
maybe_routing_tag: maybe_routing_tag.as_deref(),
mysql_pool: &server_state.mysql_pool,
}).await;
let job_token = match query_result {
Ok((job_token, _id)) => job_token,
Err(err) => {
warn!("New generic inference job creation DB error: {:?}", err);
if err.had_duplicate_idempotency_token() {
return Err(EnqueueImageGenRequestError::BadInput("Duplicate idempotency token".to_string()));
}
return Err(EnqueueImageGenRequestError::ServerError);
}
};
let response: EnqueueImageGenRequestSuccessResponse = EnqueueImageGenRequestSuccessResponse {
success: true,
inference_job_token: job_token,
};
let body = serde_json
::to_string(&response)
.map_err(|_e| EnqueueImageGenRequestError::ServerError)?;
// Error handling 101 rust result type returned like so.
Ok(HttpResponse::Ok().content_type("application/json").body(body))
}
fn validate_request(
request: &Json<EnqueueStudioImageGenRequest>,
) -> Result<(), EnqueueImageGenRequestError> {
Ok(())
}
#[derive(Debug)]
pub enum DownloadMediaFileError {
IoError(std::io::Error),
MediaFileNotFound,
Other(anyhow::Error),
}
impl From<GetMediaFileError> for DownloadMediaFileError {
fn from(err: GetMediaFileError) -> Self {
match err {
GetMediaFileError::Other(e) => DownloadMediaFileError::Other(e),
GetMediaFileError::NotFound => DownloadMediaFileError::MediaFileNotFound,
}
}
}
impl From<anyhow::Error> for DownloadMediaFileError {
fn from(err: anyhow::Error) -> Self {
DownloadMediaFileError::Other(err)
}
}
pub async fn do_download_media_file(
media_file_name: &str,
media_file_token: &MediaFileToken,
download_directory: &Path,
bucket_client: &BucketClient,
mysql_pool: &MySqlPool,
) -> Result<(PathBuf), DownloadMediaFileError> {
let final_download_path = download_directory.join(format!("{}", media_file_name));
if final_download_path.exists() {
log::debug!("Deleting existing file at {:?}", final_download_path.display().to_string());
std::fs::remove_file(&final_download_path)
.map_err(|err| DownloadMediaFileError::IoError(err))?;
}
let bucket_object_path = get_media_file_details(&media_file_token, mysql_pool).await?;
log::info!("Downloading media from bucket path: {:?}", &bucket_object_path);
bucket_client.download_file_to_disk(
&bucket_object_path,
&final_download_path,
).await?;
Ok(final_download_path)
}
#[derive(Debug)]
enum GetMediaFileError {
NotFound,
Other(anyhow::Error),
}
async fn get_media_file_details(media_file_token: &MediaFileToken, mysql_pool: &MySqlPool) -> Result<PathBuf, GetMediaFileError> {
let media_file_result = get_media_file(
&media_file_token,
false,
mysql_pool
).await;
let media_file_result = match media_file_result {
Ok(Some(result)) => result,
Ok(None) => {
warn!("could not find media file: {:?}", media_file_token);
return Err(GetMediaFileError::NotFound)
}
Err(e) => {
error!("could not query media file: {:?}", e);
return Err(GetMediaFileError::Other(e))
}
};
let media_file_bucket_path = MediaFileBucketPath::from_object_hash(
&media_file_result.public_bucket_directory_hash,
media_file_result.maybe_public_bucket_prefix.as_deref(),
media_file_result.maybe_public_bucket_extension.as_deref());
Ok(media_file_bucket_path.to_full_object_pathbuf())
}
#[cfg(test)]
mod tests {
}
@@ -0,0 +1 @@
pub mod enqueue_studio_image_generation;
@@ -0,0 +1 @@
pub mod upload_image_media_file_handler;
@@ -0,0 +1,297 @@
use std::collections::HashSet;
use std::io::{BufReader, Bytes, Cursor, Read};
use std::path::PathBuf;
use std::sync::Arc;
use actix_multipart::form::tempfile::TempFile;
use actix_multipart::form::text::Text;
use actix_multipart::form::MultipartForm;
use actix_multipart::Multipart;
use actix_web::web::Json;
use actix_web::{web, HttpRequest, HttpResponse};
use log::{error, info, warn};
use once_cell::sync::Lazy;
use stripe::CreatePaymentLinkShippingAddressCollectionAllowedCountries::Mf;
use utoipa::ToSchema;
use bucket_paths::legacy::typified_paths::public::media_files::bucket_file_path::MediaFileBucketPath;
use enums::by_table::media_files::media_file_class::MediaFileClass;
use enums::by_table::media_files::media_file_engine_category::MediaFileEngineCategory;
use enums::by_table::media_files::media_file_type::MediaFileType;
use enums::common::visibility::Visibility;
use enums::no_table::style_transfer::style_transfer_name::StyleTransferName;
use hashing::sha256::sha256_hash_bytes::sha256_hash_bytes;
use http_server_common::request::get_request_ip::get_request_ip;
use media::decode_basic_audio_info::decode_basic_audio_bytes_info;
use mimetypes::mimetype_for_bytes::get_mimetype_for_bytes;
use mimetypes::mimetype_to_extension::mimetype_to_extension;
use mysql_queries::queries::idepotency_tokens::insert_idempotency_token::insert_idempotency_token;
use mysql_queries::queries::media_files::create::specialized_insert::insert_media_file_from_file_upload::{insert_media_file_from_file_upload, InsertMediaFileFromUploadArgs, UploadType};
use mysql_queries::queries::media_files::get::get_media_file::get_media_file;
use tokens::tokens::media_files::MediaFileToken;
use crate::http_server::endpoints::media_files::upload::upload_error::MediaFileUploadError;
use crate::http_server::validations::validate_idempotency_token_format::validate_idempotency_token_format;
use crate::state::server_state::ServerState;
use crate::util::check_creator_tokens::{check_creator_tokens, CheckCreatorTokenArgs, CheckCreatorTokenResult};
/// Form-multipart request fields.
///
/// IF VIEWING DOCS, PLEASE SEE BOTTOM OF PAGE `UploadImageMediaFileForm` (Under "Schema") FOR DETAILS ON FIELDS AND NULLABILITY.
#[derive(MultipartForm, ToSchema)]
#[multipart(duplicate_field = "deny")]
pub struct UploadImageMediaFileForm {
/// UUID for request idempotency
#[multipart(limit = "2 KiB")]
#[schema(value_type = String, format = Binary)]
uuid_idempotency_token: Text<String>,
/// The uploaded screenshot
#[multipart(limit = "512 MiB")]
#[schema(value_type = Vec<u8>, format = Binary)]
screenshot: TempFile,
/// Optional: Title (name) of the scene
#[multipart(limit = "2 KiB")]
#[schema(value_type = Option<String>, format = Binary)]
maybe_title: Option<Text<String>>,
/// Optional: Visibility of the scene
#[multipart(limit = "2 KiB")]
#[schema(value_type = Option<Visibility>, format = Binary)]
maybe_visibility: Option<Text<Visibility>>,
/// Optional: If an engine scene was used to create this snapshot, provide it here to create a link
#[multipart(limit = "2 KiB")]
#[schema(value_type = Option<MediaFileToken>, format = Binary)]
maybe_scene_source_media_file_token: Option<Text<MediaFileToken>>,
}
// Unlike the "upload" endpoints, which are pure inserts, these endpoints are *upserts*.
#[derive(Serialize, ToSchema)]
pub struct UploadImageMediaFileSuccessResponse {
pub success: bool,
pub media_file_token: MediaFileToken,
}
static ALLOWED_MIME_TYPES : Lazy<HashSet<&'static str>> = Lazy::new(|| {
HashSet::from([
"image/jpeg",
"image/png",
"image/gif",
"image/webp",
])
});
/// This endpoint is for uploading image files.
#[utoipa::path(
post,
tag = "Media Files (Upload)",
path = "/v1/image_studio/scene_snapshot",
responses(
(status = 200, description = "Success Update", body = UploadImageMediaFileSuccessResponse),
(status = 400, description = "Bad input", body = MediaFileUploadError),
(status = 401, description = "Not authorized", body = MediaFileUploadError),
(status = 429, description = "Too many requests", body = MediaFileUploadError),
(status = 500, description = "Server error", body = MediaFileUploadError),
),
params(
(
"request" = UploadImageMediaFileForm,
description = "IF VIEWING DOCS, PLEASE SEE BOTTOM OF PAGE `UploadImageMediaFileForm` (Under 'Schema') FOR DETAILS ON FIELDS AND NULLABILITY."
),
)
)]
pub async fn upload_studio_scene_snapshot_handler(
http_request: HttpRequest,
server_state: web::Data<Arc<ServerState>>,
MultipartForm(mut form): MultipartForm<UploadImageMediaFileForm>,
) -> Result<Json<UploadImageMediaFileSuccessResponse>, MediaFileUploadError> {
let mut mysql_connection = server_state.mysql_pool
.acquire()
.await
.map_err(|err| {
error!("MySql pool error: {:?}", err);
MediaFileUploadError::ServerError
})?;
// ==================== READ SESSION ==================== //
let maybe_user_session = server_state
.session_checker
.maybe_get_user_session_from_connection(&http_request, &mut mysql_connection)
.await
.map_err(|e| {
error!("Session checker error: {:?}", e);
MediaFileUploadError::ServerError
})?;
let maybe_user_token = maybe_user_session
.as_ref()
.map(|session| session.get_strongly_typed_user_token());
let maybe_avt_token = server_state
.avt_cookie_manager
.get_avt_token_from_request(&http_request);
// ==================== BANNED USERS ==================== //
if let Some(ref user) = maybe_user_session {
if user.is_banned {
return Err(MediaFileUploadError::NotAuthorizedVerbose("user is banned".to_string()));
}
}
// ==================== RATE LIMIT ==================== //
let rate_limiter = match maybe_user_session {
None => &server_state.redis_rate_limiters.file_upload_logged_out,
Some(ref _session) => &server_state.redis_rate_limiters.file_upload_logged_in,
};
if let Err(_err) = rate_limiter.rate_limit_request(&http_request) {
return Err(MediaFileUploadError::RateLimited);
}
// ==================== HANDLE IDEMPOTENCY ==================== //
// TODO(bt, 2024-02-26): This should be a transaction.
let uuid_idempotency_token = form.uuid_idempotency_token.as_ref();
if let Err(reason) = validate_idempotency_token_format(uuid_idempotency_token) {
return Err(MediaFileUploadError::BadInput(reason));
}
insert_idempotency_token(uuid_idempotency_token, &mut *mysql_connection)
.await
.map_err(|err| {
error!("Error inserting idempotency token: {:?}", err);
MediaFileUploadError::BadInput("invalid idempotency token".to_string())
})?;
// ==================== UPLOAD METADATA ==================== //
let maybe_title = form.maybe_title.map(|title| title.to_string());
let creator_set_visibility = form.maybe_visibility
.map(|visibility| visibility.0)
.or_else(|| {
maybe_user_session
.as_ref()
.map(|user_session| user_session.preferred_tts_result_visibility)
})
.unwrap_or(Visibility::default());
// ==================== USER DATA ==================== //
let ip_address = get_request_ip(&http_request);
// ==================== FILE VALIDATION ==================== //
let mut file_bytes = Vec::new();
form.screenshot.file.read_to_end(&mut file_bytes)
.map_err(|e| {
error!("Problem reading file: {:?}", e);
MediaFileUploadError::ServerError
})?;
let mimetype = get_mimetype_for_bytes(file_bytes.as_ref())
.map(|mimetype| mimetype.to_string())
.ok_or_else(|| {
warn!("Could not determine mimetype for file");
MediaFileUploadError::BadInput("Could not determine mimetype for file".to_string())
})?;
if !ALLOWED_MIME_TYPES.contains(mimetype.as_str()) {
// NB: Don't let our error message inject malicious strings
let filtered_mimetype = mimetype
.chars()
.filter(|c| c.is_ascii())
.filter(|c| c.is_alphanumeric() || *c == '/')
.collect::<String>();
return Err(MediaFileUploadError::BadInput(format!("unpermitted mime type: {}", &filtered_mimetype)));
}
// ==================== OTHER FILE METADATA ==================== //
let maybe_filename = form.screenshot.file_name.as_deref()
.as_deref()
.map(|filename| PathBuf::from(filename));
let extension = mimetype_to_extension(&mimetype)
.or_else(|| {
maybe_filename
.as_ref()
.and_then(|filename| filename.extension()) // TODO needs dot prefix
.and_then(|ext| ext.to_str())
})
.ok_or_else(|| {
warn!("Could not determine file extension for mimetype: {}", &mimetype);
MediaFileUploadError::ServerError
})?;
let extension = format!(".{extension}"); // NB: needs dot prefix
let file_size_bytes = file_bytes.len();
let hash = sha256_hash_bytes(&file_bytes)
.map_err(|io_error| {
error!("Problem hashing bytes: {:?}", io_error);
MediaFileUploadError::ServerError
})?;
// ==================== UPLOAD AND SAVE ==================== //
const PREFIX : Option<&str> = Some("image_");
let public_upload_path = MediaFileBucketPath::generate_new(PREFIX, Some(&extension));
info!("Uploading media to bucket path: {}", public_upload_path.get_full_object_path_str());
server_state.public_bucket_client.upload_file_with_content_type(
public_upload_path.get_full_object_path_str(),
file_bytes.as_ref(),
&mimetype)
.await
.map_err(|e| {
warn!("Upload media bytes to bucket error: {:?}", e);
MediaFileUploadError::ServerError
})?;
let (token, record_id) = insert_media_file_from_file_upload(InsertMediaFileFromUploadArgs {
maybe_media_class: Some(MediaFileClass::Image),
media_file_type: MediaFileType::Image,
maybe_creator_user_token: maybe_user_token.as_ref(),
maybe_creator_anonymous_visitor_token: maybe_avt_token.as_ref(),
creator_ip_address: &ip_address,
creator_set_visibility,
upload_type: UploadType::Filesystem, // TODO(bt,2024-05-02): This should be a parameter and a well-known enum.
maybe_engine_category: None,
maybe_animation_type: None,
maybe_mime_type: Some(&mimetype),
file_size_bytes: file_size_bytes as u64,
maybe_duration_millis: None,
sha256_checksum: &hash,
maybe_title: maybe_title.as_deref(),
maybe_scene_source_media_file_token: None,
is_intermediate_system_file: true,
public_bucket_directory_hash: public_upload_path.get_object_hash(),
maybe_public_bucket_prefix: PREFIX,
maybe_public_bucket_extension: Some(&extension),
pool: &server_state.mysql_pool,
})
.await
.map_err(|err| {
warn!("New file creation DB error: {:?}", err);
MediaFileUploadError::ServerError
})?;
info!("new media file id: {} token: {:?}", record_id, &token);
Ok(Json(UploadImageMediaFileSuccessResponse {
success: true,
media_file_token: token,
}))
}
@@ -40,3 +40,4 @@ pub mod voice_designer;
pub mod w2l;
pub mod weights;
pub mod workflows;
pub mod image_studio;
@@ -0,0 +1,29 @@
use crate::http_server::endpoints::image_studio::prompt::enqueue_studio_image_generation::enqueue_studio_image_generation_request;
use crate::http_server::endpoints::image_studio::upload::upload_image_media_file_handler::upload_studio_scene_snapshot_handler;
use actix_http::body::MessageBody;
use actix_service::ServiceFactory;
use actix_web::dev::{ServiceRequest, ServiceResponse};
use actix_web::{web, App, Error, HttpResponse};
pub fn add_image_studio_routes<T, B> (app: App<T>) -> App<T>
where
B: MessageBody,
T: ServiceFactory<
ServiceRequest,
Config = (),
Response = ServiceResponse<B>,
Error = Error,
InitError = (),
>,
{
app.service(web::scope("/v1/image_studio")
.service(web::resource("/scene_snapshot")
.route(web::post().to(upload_studio_scene_snapshot_handler))
.route(web::head().to(|| HttpResponse::Ok()))
)
.service(web::resource("/prompt")
.route(web::post().to(enqueue_studio_image_generation_request))
.route(web::head().to(|| HttpResponse::Ok()))
)
)
}
@@ -123,6 +123,7 @@ use crate::http_server::endpoints::w2l::get_w2l_template_use_count::get_w2l_temp
use crate::http_server::endpoints::w2l::get_w2l_upload_template_job_status::get_w2l_upload_template_job_status_handler;
use crate::http_server::endpoints::w2l::list_w2l_templates::list_w2l_templates_handler;
use crate::http_server::endpoints::w2l::set_w2l_template_mod_approval::set_w2l_template_mod_approval_handler;
use crate::http_server::routes::add_image_studio_routes::add_image_studio_routes;
use crate::http_server::routes::add_studio_gen2_routes::add_studio_gen2_routes;
use crate::http_server::routes::beta_key_routes::add_beta_key_routes;
use crate::http_server::routes::job_routes::add_job_routes;
@@ -171,6 +172,7 @@ pub fn add_routes<T, B> (app: App<T>, server_environment: ServerEnvironment) ->
app = add_model_download_routes(app);
app = add_workflow_routes(app);
app = add_studio_gen2_routes(app);
app = add_image_studio_routes(app);
app = add_job_routes(app);
app = add_engine_routes(app); /* /v1/engine/... */
@@ -8,4 +8,5 @@ pub mod moderation_routes;
pub mod user_routes;
pub mod weights_routes;
pub mod workflow_routes;
mod tag_routes;
mod tag_routes;
mod add_image_studio_routes;