Compare commits

...
Author SHA1 Message Date
Brandon Thomas b3da60aa3a cleanup pathing (still wip) 2024-07-10 20:58:39 -04:00
Brandon Thomas 342ab2302c refactor worker (wip) 2024-07-10 19:11:28 -04:00
Brandon Thomas ab645cacc6 fix more batch queries with empty tokens in where in clause 2024-07-10 19:02:48 -04:00
Brandon Thomas ad4b297b55 stop invalid batch queries 2024-07-10 18:52:11 -04:00
Brandon Thomas b1b7f4e0ee comfy refactoring of pathing 2024-07-10 00:12:48 -04:00
Brandon Thomas 77bebd4833 cleaning up pathing 2024-07-09 23:15:08 -04:00
Brandon Thomas 9383ba42fb more refactor 2024-07-09 22:14:56 -04:00
Brandon Thomas 2e79eb9c3a this refactor might make things clearer or might be horrible 2024-07-09 22:06:02 -04:00
Brandon Thomas 55deba82ae download, process, and pass (depth,normal,outline) map videos 2024-07-09 21:33:58 -04:00
Brandon Thomas e0121dffeb download secondary videos 2024-07-09 18:38:07 -04:00
Brandon Thomas 62d4e260a0 new video inputs to vst 2024-07-09 17:04:18 -04:00
Brandon Thomas ea1435ad6e doc queue stats 2024-07-09 16:32:51 -04:00
Brandon Thomas 34d19adc1e form validation 2024-07-09 15:22:47 -04:00
Brandon Thomas 7e630bf588 validate frame skip in enqueue vst 2024-07-08 23:34:59 -04:00
Brandon Thomas c1544535df consolidate pathinfo type 2024-07-07 23:54:21 -04:00
Brandon Thomas e6a18f7974 new endpoint to modify media file animation type 2024-07-07 23:37:21 -04:00
Brandon Thomas 783d73a48b better organize media file queries 2024-07-07 23:13:43 -04:00
Brandon Thomas 3f51515b17 new endpoint to modify engine category 2024-07-07 22:51:22 -04:00
Brandon Thomas c44efc2a18 update elasticsearch cli search to match update job 2024-07-07 20:07:48 -04:00
Brandon Thomas 7fcaf61170 es-update-job: update media files relevant to the engine 2024-07-07 20:06:47 -04:00
Brandon Thomas f31a1bb12e fix discord username validation 2024-07-07 19:30:47 -04:00
Brandon Thomas 113ed359fd start using Json() for better typing, correctness, testability 2024-07-07 19:23:57 -04:00
Brandon Thomas 17a4cf774d remove lazy_static from storyteller-web 2024-07-07 18:38:42 -04:00
Brandon Thomas 7bfd38a382 merge users component with storyteller-web
This will make adding SSO functions easier without all of the package
indirections.
2024-07-07 18:25:55 -04:00
Brandon Thomas 970e802f14 get rid of top level model module 2024-07-07 17:31:23 -04:00
Brandon Thomas ce186179ce move api docs into module 2024-07-07 17:23:19 -04:00
Brandon Thomas 45792890b7 report new prompt fields in API 2024-07-07 11:13:03 -04:00
Brandon Thomas 776c086470 save frame skip to prompt record 2024-07-07 11:07:23 -04:00
Brandon Thomas b666858b21 save travel prompt 2024-07-06 23:50:24 -04:00
Brandon Thomas 4ca8932591 inference job: add prompt travel and skip frame 2024-07-06 18:37:50 -04:00
Brandon Thomas 004e8c27d4 read frame skip and travel prompt in enqueue VST endpoint 2024-07-06 17:28:20 -04:00
Brandon Thomas 70c2c995ee improve comfy estimates code 2024-07-06 08:43:25 -04:00
Brandon Thomas 54b7e182b0 added tests for job progress 2024-07-05 22:15:26 -04:00
Brandon Thomas 50807429ad bad comparison 2024-07-05 22:04:04 -04:00
Brandon Thomas 1cd3175e13 move memory caching module 2024-07-05 22:02:15 -04:00
Brandon Thomas bbaed857d0 move server_state into module 2024-07-05 20:56:00 -04:00
Brandon Thomas 37ccf0f60d reduce unused code and module subscriptions 2024-07-05 20:51:01 -04:00
Brandon Thomas c95ad4dd55 refactor away more useless modules 2024-07-05 20:45:07 -04:00
Brandon Thomas 7609c040cc move module routes 2024-07-05 20:40:54 -04:00
Brandon Thomas 076a865747 move validations module 2024-07-05 20:38:20 -04:00
Brandon Thomas 972e8199e7 pull routes out of users component, remove unused module 2024-07-05 20:35:19 -04:00
Brandon Thomas 175035000b better doc 2024-07-05 16:04:02 -04:00
Brandon Thomas 807c8fd4f8 add progress estimation to job status endpoints (this is incomplete) (#353)
This adds a progress indicator to the three job status endpoints. 

There's a function to calculate progress, which is mostly left as a stub
for now.

In the future we may even want to do dynamically updating averages.
2024-07-05 15:23:00 -04:00
370 changed files with 3058 additions and 1555 deletions
@@ -0,0 +1,12 @@
{
"db_name": "MySQL",
"query": "\nUPDATE media_files\nSET\n maybe_animation_type = ?\nWHERE\n token = ?\nLIMIT 1\n ",
"describe": {
"columns": [],
"parameters": {
"Right": 2
},
"nullable": []
},
"hash": "930cfd52f78fa14a86bd1a67c5fe36d8c8961079b2bd684a0d4da315bf3ce16e"
}
@@ -0,0 +1,12 @@
{
"db_name": "MySQL",
"query": "\nUPDATE media_files\nSET\n maybe_engine_category = ?\nWHERE\n token = ?\nLIMIT 1\n ",
"describe": {
"columns": [],
"parameters": {
"Right": 2
},
"nullable": []
},
"hash": "b56ee40be840cdec013f3f99eb585e4ac333ccfe563ba3da4c9a80df33dccd57"
}
@@ -1,6 +1,6 @@
{
"db_name": "MySQL",
"query": "\nSELECT\n jobs.token as `job_token: tokens::tokens::generic_inference_jobs::InferenceJobToken`,\n\n jobs.status as `status: enums::common::job_status_plus::JobStatusPlus`,\n jobs.attempt_count,\n\n jobs.maybe_creator_user_token as `maybe_creator_user_token: tokens::tokens::users::UserToken`,\n jobs.maybe_creator_anonymous_visitor_token as `maybe_creator_anonymous_visitor_token: tokens::tokens::anonymous_visitor_tracking::AnonymousVisitorTrackingToken`,\n jobs.creator_ip_address,\n\n jobs.inference_category as `inference_category: enums::by_table::generic_inference_jobs::inference_category::InferenceCategory`,\n jobs.maybe_model_type,\n jobs.maybe_model_token,\n jobs.maybe_raw_inference_text,\n jobs.maybe_inference_args,\n\n jobs.frontend_failure_category as `maybe_frontend_failure_category: enums::by_table::generic_inference_jobs::frontend_failure_category::FrontendFailureCategory`,\n\n jobs.on_success_result_entity_type as maybe_result_entity_type,\n jobs.on_success_result_entity_token as maybe_result_entity_token,\n\n model_weights.title as maybe_model_weights_title,\n tts_models.title as maybe_tts_model_title,\n voice_conversion_models.title as maybe_voice_conversion_model_title,\n\n tts_results.public_bucket_wav_audio_path as maybe_tts_public_bucket_path,\n voice_conversion_results.public_bucket_hash as maybe_voice_conversion_public_bucket_hash,\n\n media_files.public_bucket_directory_hash as maybe_media_file_public_bucket_directory_hash,\n media_files.maybe_public_bucket_prefix as maybe_media_file_public_bucket_prefix,\n media_files.maybe_public_bucket_extension as maybe_media_file_public_bucket_extension,\n\n jobs.assigned_worker as maybe_assigned_worker,\n jobs.assigned_cluster as maybe_assigned_cluster,\n\n jobs.is_keepalive_required,\n\n jobs.created_at,\n jobs.updated_at,\n\n jobs.first_started_at as maybe_first_started_at,\n jobs.successfully_completed_at as maybe_successfully_completed_at\n\nFROM generic_inference_jobs as jobs\n\nLEFT OUTER JOIN model_weights ON jobs.maybe_model_token = model_weights.token\nLEFT OUTER JOIN tts_models ON jobs.maybe_model_token = tts_models.token\nLEFT OUTER JOIN voice_conversion_models ON jobs.maybe_model_token = voice_conversion_models.token\n\nLEFT OUTER JOIN tts_results ON jobs.on_success_result_entity_token = tts_results.token\nLEFT OUTER JOIN voice_conversion_results ON jobs.on_success_result_entity_token = voice_conversion_results.token\nLEFT OUTER JOIN media_files ON jobs.on_success_result_entity_token = media_files.token\n\nWHERE jobs.token = ?\n ",
"query": "\nSELECT\n jobs.token as `job_token: tokens::tokens::generic_inference_jobs::InferenceJobToken`,\n\n jobs.status as `status: enums::common::job_status_plus::JobStatusPlus`,\n jobs.attempt_count,\n\n jobs.maybe_creator_user_token as `maybe_creator_user_token: tokens::tokens::users::UserToken`,\n jobs.maybe_creator_anonymous_visitor_token as `maybe_creator_anonymous_visitor_token: tokens::tokens::anonymous_visitor_tracking::AnonymousVisitorTrackingToken`,\n jobs.creator_ip_address,\n\n jobs.inference_category as `inference_category: enums::by_table::generic_inference_jobs::inference_category::InferenceCategory`,\n jobs.maybe_model_type,\n jobs.maybe_model_token,\n jobs.maybe_raw_inference_text,\n jobs.maybe_inference_args,\n\n jobs.frontend_failure_category as `maybe_frontend_failure_category: enums::by_table::generic_inference_jobs::frontend_failure_category::FrontendFailureCategory`,\n\n jobs.on_success_result_entity_type as maybe_result_entity_type,\n jobs.on_success_result_entity_token as maybe_result_entity_token,\n\n model_weights.title as maybe_model_weights_title,\n tts_models.title as maybe_tts_model_title,\n voice_conversion_models.title as maybe_voice_conversion_model_title,\n\n tts_results.public_bucket_wav_audio_path as maybe_tts_public_bucket_path,\n voice_conversion_results.public_bucket_hash as maybe_voice_conversion_public_bucket_hash,\n\n media_files.public_bucket_directory_hash as maybe_media_file_public_bucket_directory_hash,\n media_files.maybe_public_bucket_prefix as maybe_media_file_public_bucket_prefix,\n media_files.maybe_public_bucket_extension as maybe_media_file_public_bucket_extension,\n\n jobs.assigned_worker as maybe_assigned_worker,\n jobs.assigned_cluster as maybe_assigned_cluster,\n\n jobs.is_keepalive_required,\n\n jobs.created_at,\n jobs.updated_at,\n\n jobs.first_started_at as maybe_first_started_at,\n jobs.successfully_completed_at as maybe_successfully_completed_at,\n\n NOW() as database_clock\n\nFROM generic_inference_jobs as jobs\n\nLEFT OUTER JOIN model_weights ON jobs.maybe_model_token = model_weights.token\nLEFT OUTER JOIN tts_models ON jobs.maybe_model_token = tts_models.token\nLEFT OUTER JOIN voice_conversion_models ON jobs.maybe_model_token = voice_conversion_models.token\n\nLEFT OUTER JOIN tts_results ON jobs.on_success_result_entity_token = tts_results.token\nLEFT OUTER JOIN voice_conversion_results ON jobs.on_success_result_entity_token = voice_conversion_results.token\nLEFT OUTER JOIN media_files ON jobs.on_success_result_entity_token = media_files.token\n\nWHERE jobs.token = ?\n ",
"describe": {
"columns": [
{
@@ -292,6 +292,16 @@
"char_set": 63,
"max_size": 19
}
},
{
"ordinal": 29,
"name": "database_clock",
"type_info": {
"type": "Datetime",
"flags": "NOT_NULL | BINARY",
"char_set": 63,
"max_size": 19
}
}
],
"parameters": {
@@ -326,8 +336,9 @@
false,
false,
true,
true
true,
false
]
},
"hash": "7d5030b221e9b470ee7af4b73539fd2ef477e6f3c98c56422016449c6775222d"
"hash": "e5121d132d93ec43e48197e00b19040a50ff2942224dc70b0faa0ec988aee7cf"
}
Generated
+3 -49
View File
@@ -2375,6 +2375,7 @@ dependencies = [
"elasticsearch_schema",
"enums",
"errors",
"futures 0.3.29",
"hostname 0.4.0",
"log",
"my-workspace-hack",
@@ -6817,6 +6818,7 @@ dependencies = [
"mysql_queries",
"native-tls",
"once_cell",
"password",
"pin-project-lite",
"primitives",
"prompt_engineering",
@@ -6837,6 +6839,7 @@ dependencies = [
"speculoos",
"sqlx",
"strum",
"strum_macros",
"symphonia",
"time",
"tokens",
@@ -6846,7 +6849,6 @@ dependencies = [
"url_config",
"user_input_common",
"user_traits_component",
"users_component",
"utoipa",
"utoipa-swagger-ui",
"uuid",
@@ -7938,54 +7940,6 @@ dependencies = [
"my-workspace-hack",
]
[[package]]
name = "users_component"
version = "0.0.1"
dependencies = [
"actix-web",
"actix_helpers",
"anyhow",
"base64 0.22.1",
"byteorder",
"chrono",
"cookies",
"crockford",
"dotenv",
"email_sender",
"enums",
"errors",
"hmac 0.10.1",
"http_server_common",
"idempotency",
"jwt",
"lazy_static",
"log",
"md-5 0.10.6",
"my-workspace-hack",
"mysql_queries",
"once_cell",
"password",
"r2d2_redis",
"rand 0.8.5",
"redis_caching",
"redis_common",
"regex",
"reusable_types",
"serde",
"serde_derive",
"serde_json",
"server_environment",
"sha2 0.9.9",
"sqlx",
"strum_macros",
"time",
"tokens",
"toml 0.8.14",
"user_input_common",
"user_traits_component",
"utoipa",
]
[[package]]
name = "utf-8"
version = "0.7.6"
-1
View File
@@ -88,7 +88,6 @@ members = [
"crates/service/plugins/billing",
"crates/service/plugins/inference_workloads",
"crates/service/plugins/user_traits_component",
"crates/service/plugins/users",
# Vendored code
"crates/vendor/limitation",
@@ -11,7 +11,7 @@ from media_files as m
join users as u
on m.maybe_creator_user_token = u.token
where u.can_access_studio = true
and m.created_at > (CURDATE() - INTERVAL 1 DAY)
and m.created_at > (CURDATE() - INTERVAL 14 DAY)
and m.is_intermediate_system_file = false
and m.media_class = 'video'
group by u.token
@@ -8,7 +8,7 @@ use sqlx::{MySql, Pool};
use enums::by_table::media_files::media_file_type::MediaFileType;
use enums::common::view_as::ViewAs;
use errors::AnyhowResult;
use mysql_queries::queries::media_files::delete_media_file::delete_media_file_as_mod;
use mysql_queries::queries::media_files::delete::delete_media_file::delete_media_file_as_mod;
use mysql_queries::queries::media_files::list::list_media_files::{list_media_files, ListMediaFilesArgs};
use crate::cli_args::Args;
@@ -6,7 +6,7 @@ use sqlx::{MySql, Pool};
use enums::common::view_as::ViewAs;
use errors::{anyhow, AnyhowResult};
use mysql_queries::queries::media_files::delete_media_file::delete_media_file_as_mod;
use mysql_queries::queries::media_files::delete::delete_media_file::delete_media_file_as_mod;
use mysql_queries::queries::media_files::list::list_media_files_for_user::{list_media_files_for_user, ListMediaFileForUserArgs};
use mysql_queries::queries::users::user_profiles::get_user_profile_by_username::get_user_profile_by_username;
@@ -43,10 +43,15 @@ pub async fn create_dimensional_media_file_documents(
maybe_filter_engine_categories: Some(&HashSet::from_iter(vec![
MediaFileEngineCategory::Animation,
MediaFileEngineCategory::Character,
MediaFileEngineCategory::Creature,
MediaFileEngineCategory::Expression,
MediaFileEngineCategory::ImagePlane,
MediaFileEngineCategory::Location,
MediaFileEngineCategory::Object,
MediaFileEngineCategory::Scene,
MediaFileEngineCategory::SetDressing,
MediaFileEngineCategory::Skybox,
MediaFileEngineCategory::VideoPlane,
])),
maybe_filter_media_types: Some(&HashSet::from_iter(vec![
// Engine types
@@ -11,6 +11,8 @@ pub async fn test_search_media_files(client: &Elasticsearch) -> AnyhowResult<()>
let results = search_media_files(SearchArgs {
search_term: "cac",
is_featured: None,
maybe_creator_user_token: None,
maybe_media_classes: Some(HashSet::from_iter(vec![
MediaFileClass::Dimensional,
MediaFileClass::Image,
@@ -58,6 +58,24 @@ pub struct WorkflowArgs {
#[serde(skip_serializing_if = "Option::is_none")]
pub maybe_input_file: Option<MediaFileToken>,
#[serde(rename = "df")]
#[serde(skip_serializing_if = "Option::is_none")]
pub maybe_input_depth_file: Option<MediaFileToken>,
#[serde(rename = "nf")]
#[serde(skip_serializing_if = "Option::is_none")]
pub maybe_input_normal_file: Option<MediaFileToken>,
#[serde(rename = "of")]
#[serde(skip_serializing_if = "Option::is_none")]
pub maybe_input_outline_file: Option<MediaFileToken>,
// Optional global IP Adapter image media file token
// This will apply to the entire workflow / video
#[serde(rename = "gi")]
#[serde(skip_serializing_if = "Option::is_none")]
pub global_ip_adapter_token: Option<MediaFileToken>,
#[serde(rename = "out")]
#[serde(skip_serializing_if = "Option::is_none")]
pub maybe_output_path: Option<String>,
@@ -123,11 +141,9 @@ pub struct WorkflowArgs {
#[serde(skip_serializing_if = "Option::is_none")]
pub negative_prompt: Option<String>,
// Optional global IP Adapter image media file token
// This will apply to the entire workflow / video
#[serde(rename = "gi")]
#[serde(rename = "tp")]
#[serde(skip_serializing_if = "Option::is_none")]
pub global_ip_adapter_token: Option<MediaFileToken>,
pub travel_prompt: Option<String>,
#[deprecated(note = "Use `lipsync_enabled` instead")]
#[serde(rename = "el")]
@@ -177,4 +193,8 @@ pub struct WorkflowArgs {
#[serde(rename = "s")]
#[serde(skip_serializing_if = "Option::is_none")]
pub strength: Option<f32>,
#[serde(rename = "fs")]
#[serde(skip_serializing_if = "Option::is_none")]
pub frame_skip: Option<u8>,
}
@@ -1,4 +1,5 @@
use chrono::Duration;
use enums::no_table::style_transfer::style_transfer_name::StyleTransferName;
use errors::AnyhowResult;
use tokens::tokens::media_files::MediaFileToken;
@@ -72,6 +73,18 @@ pub struct PromptInnerPayload {
#[serde(alias = "global_ipa_token")]
#[serde(skip_serializing_if = "Option::is_none")]
pub global_ipa_token: Option<MediaFileToken>,
/// Prompt travel prompt
#[serde(rename = "tp")] // NB: DO NOT CHANGE: IT WILL BREAK MYSQL RECORDS. Renamed to consume fewer bytes.
#[serde(alias = "travel_prompt")]
#[serde(skip_serializing_if = "Option::is_none")]
pub travel_prompt: Option<String>,
/// Frame skip
#[serde(rename = "fs")] // NB: DO NOT CHANGE: IT WILL BREAK MYSQL RECORDS. Renamed to consume fewer bytes.
#[serde(alias = "frame_skip")]
#[serde(skip_serializing_if = "Option::is_none")]
pub frame_skip: Option<u8>,
}
pub struct PromptInnerPayloadBuilder {
@@ -87,6 +100,8 @@ pub struct PromptInnerPayloadBuilder {
pub face_detailer_workflow: Option<String>,
pub upscaler_workflow: Option<String>,
pub global_ipa_token: Option<MediaFileToken>,
pub travel_prompt: Option<String>,
pub frame_skip: Option<u8>,
}
impl PromptInnerPayloadBuilder {
@@ -104,6 +119,8 @@ impl PromptInnerPayloadBuilder {
face_detailer_workflow: None,
upscaler_workflow: None,
global_ipa_token: None,
travel_prompt: None,
frame_skip: None,
}
}
@@ -120,6 +137,8 @@ impl PromptInnerPayloadBuilder {
&& self.face_detailer_workflow.is_none()
&& self.upscaler_workflow.is_none()
&& self.global_ipa_token.is_none()
&& self.travel_prompt.is_none()
&& self.frame_skip.is_none()
{
return None;
}
@@ -140,6 +159,8 @@ impl PromptInnerPayloadBuilder {
face_detailer_workflow: self.face_detailer_workflow,
upscaler_workflow: self.upscaler_workflow,
global_ipa_token: self.global_ipa_token,
travel_prompt: self.travel_prompt,
frame_skip: self.frame_skip,
})
}
@@ -210,6 +231,14 @@ impl PromptInnerPayloadBuilder {
pub fn set_global_ipa_token(&mut self, token: Option<MediaFileToken>) {
self.global_ipa_token = token;
}
pub fn set_travel_prompt(&mut self, travel_prompt: Option<String>) {
self.travel_prompt = travel_prompt;
}
pub fn set_frame_skip(&mut self, frame_skip: Option<u8>) {
self.frame_skip = frame_skip;
}
}
impl PromptInnerPayload{
@@ -7,7 +7,6 @@ use sqlx::pool::PoolConnection;
use enums::by_table::generic_inference_jobs::frontend_failure_category::FrontendFailureCategory;
use enums::by_table::generic_inference_jobs::inference_category::InferenceCategory;
use enums::common::job_status_plus::JobStatusPlus;
use enums::no_table::style_transfer::style_transfer_name::StyleTransferName;
use enums::traits::mysql_from_row::MySqlFromRow;
use errors::AnyhowResult;
use tokens::tokens::anonymous_visitor_tracking::AnonymousVisitorTrackingToken;
@@ -17,84 +16,19 @@ use tokens::traits::mysql_token_from_row::MySqlTokenFromRow;
use crate::helpers::boolean_converters::i8_to_bool;
use crate::payloads::generic_inference_args::generic_inference_args::{GenericInferenceArgs, PolymorphicInferenceArgs};
// NB: Serialization is for internal endpoints, not for returning to end-users.
#[derive(Debug, Default, Serialize)]
pub struct GenericInferenceJobStatus {
pub job_token: InferenceJobToken,
pub status: JobStatusPlus,
pub attempt_count: u16,
pub maybe_assigned_worker: Option<String>,
pub maybe_assigned_cluster: Option<String>,
pub maybe_first_started_at: Option<DateTime<Utc>>,
pub maybe_frontend_failure_category: Option<FrontendFailureCategory>,
pub request_details: RequestDetails,
pub maybe_result_details: Option<ResultDetails>,
pub user_details: UserDetails,
pub is_keepalive_required: bool,
pub created_at: DateTime<Utc>,
pub updated_at: DateTime<Utc>,
}
/// NB: Serialize is for internal moderator-only endpoints
/// Details about the user's original inference request
/// (We may want to present it in the "pending" UI.)
#[derive(Debug, Default, Serialize)]
pub struct RequestDetails {
pub inference_category: InferenceCategory,
pub maybe_model_type: Option<String>, // TODO: Strongly type
pub maybe_model_token: Option<String>,
pub maybe_model_title: Option<String>,
/// TTS input. In the future, perhaps voice conversion SST
pub maybe_raw_inference_text: Option<String>,
/// For Comfy / Video Style Transfer jobs, this might include
/// the name of the selected style.
pub maybe_style_name: Option<StyleTransferName>,
}
/// NB: Serialize is for internal moderator-only endpoints
/// Details about the generated result
#[derive(Debug, Default, Serialize)]
pub struct ResultDetails {
pub entity_type: String,
pub entity_token: String,
/// The bucket storage hash (for vc and media_files) or full path (for tts)
pub public_bucket_location_or_hash: String,
pub maybe_media_file_public_bucket_prefix: Option<String>,
pub maybe_media_file_public_bucket_extension: Option<String>,
/// Whether the location is a full path (for tts) or a hash (for vc) that
/// needs to be reconstructed into a path.
pub public_bucket_location_is_hash: bool,
pub maybe_successfully_completed_at: Option<DateTime<Utc>>,
}
/// NB: DO NOT EXPOSE TO FRONTEND.
/// NB: Serialize is for internal moderator-only endpoints
/// This is used to gate access to job termination
#[derive(Debug, Default, Serialize)]
pub struct UserDetails {
pub maybe_creator_user_token: Option<UserToken>,
pub maybe_creator_anonymous_visitor_token: Option<AnonymousVisitorTrackingToken>,
pub creator_ip_address: String,
}
use crate::queries::generic_inference::web::job_status::{GenericInferenceJobStatus, RequestDetails, ResultDetails, UserDetails};
/// Look up job status.
/// Returns Ok(Vec::new()) when the records cannot be found.
pub async fn batch_get_inference_job_status(job_tokens: &[InferenceJobToken], mysql_pool: &MySqlPool)
-> AnyhowResult<Vec<GenericInferenceJobStatus>>
{
if job_tokens.is_empty() {
// NB: We should always eagerly return, but if we don't, the query builder will build an
// invalid query.
return Ok(Vec::new());
}
let mut connection = mysql_pool.acquire().await?;
batch_get_inference_job_status_from_connection(job_tokens, &mut connection).await
}
@@ -187,7 +121,9 @@ SELECT
jobs.updated_at,
jobs.first_started_at as maybe_first_started_at,
jobs.successfully_completed_at as maybe_successfully_completed_at
jobs.successfully_completed_at as maybe_successfully_completed_at,
NOW() as database_clock
FROM generic_inference_jobs as jobs
@@ -302,6 +238,7 @@ fn raw_records_to_public_result(records: Vec<RawGenericInferenceJobStatus>) -> V
maybe_model_token: record.maybe_model_token,
maybe_model_title: maybe_model_title.map(|title| title.to_string()),
maybe_raw_inference_text: record.maybe_raw_inference_text,
maybe_inference_args: record.maybe_inference_args,
maybe_style_name,
},
maybe_result_details,
@@ -313,6 +250,7 @@ fn raw_records_to_public_result(records: Vec<RawGenericInferenceJobStatus>) -> V
is_keepalive_required: i8_to_bool(record.is_keepalive_required),
created_at: record.created_at,
updated_at: record.updated_at,
database_clock: record.database_clock,
}
})
.collect::<Vec<_>>()
@@ -360,6 +298,8 @@ struct RawGenericInferenceJobStatus {
pub maybe_first_started_at: Option<DateTime<Utc>>,
pub maybe_successfully_completed_at: Option<DateTime<Utc>>,
pub database_clock: DateTime<Utc>,
}
// NB(bt,2023-12-05): There's an issue with type hinting in the `as` clauses with QueryBuilder (or
@@ -406,6 +346,7 @@ impl FromRow<'_, MySqlRow> for RawGenericInferenceJobStatus {
maybe_successfully_completed_at: row.try_get("maybe_successfully_completed_at")?,
created_at: row.try_get("created_at")?,
updated_at: row.try_get("updated_at")?,
database_clock: row.try_get("database_clock")?,
})
}
}
@@ -1,5 +1,5 @@
use anyhow::anyhow;
use chrono::{DateTime, Utc};
use chrono::{DateTime, NaiveDateTime, TimeZone, Utc};
use log::warn;
use sqlx::{MySql, MySqlPool};
use sqlx::pool::PoolConnection;
@@ -15,78 +15,7 @@ use tokens::tokens::users::UserToken;
use crate::helpers::boolean_converters::i8_to_bool;
use crate::payloads::generic_inference_args::generic_inference_args::{GenericInferenceArgs, PolymorphicInferenceArgs};
// NB: Serialization is for internal endpoints, not for returning to end-users.
#[derive(Debug, Default, Serialize)]
pub struct GenericInferenceJobStatus {
pub job_token: InferenceJobToken,
pub status: JobStatusPlus,
pub attempt_count: u16,
pub maybe_assigned_worker: Option<String>,
pub maybe_assigned_cluster: Option<String>,
pub maybe_first_started_at: Option<DateTime<Utc>>,
pub maybe_frontend_failure_category: Option<FrontendFailureCategory>,
pub request_details: RequestDetails,
pub maybe_result_details: Option<ResultDetails>,
pub user_details: UserDetails,
pub is_keepalive_required: bool,
pub created_at: DateTime<Utc>,
pub updated_at: DateTime<Utc>,
}
/// NB: Serialize is for internal moderator-only endpoints
/// Details about the user's original inference request
/// (We may want to present it in the "pending" UI.)
#[derive(Debug, Default, Serialize)]
pub struct RequestDetails {
pub inference_category: InferenceCategory,
pub maybe_model_type: Option<String>, // TODO: Strongly type
pub maybe_model_token: Option<String>,
pub maybe_model_title: Option<String>,
/// TTS input. In the future, perhaps voice conversion SST
pub maybe_raw_inference_text: Option<String>,
/// For Comfy / Video Style Transfer jobs, this might include
/// the name of the selected style.
pub maybe_style_name: Option<StyleTransferName>,
}
/// NB: Serialize is for internal moderator-only endpoints
/// Details about the generated result
#[derive(Debug, Default, Serialize)]
pub struct ResultDetails {
pub entity_type: String,
pub entity_token: String,
/// The bucket storage hash (for vc and media_files) or full path (for tts)
pub public_bucket_location_or_hash: String,
pub maybe_media_file_public_bucket_prefix: Option<String>,
pub maybe_media_file_public_bucket_extension: Option<String>,
/// Whether the location is a full path (for tts) or a hash (for vc) that
/// needs to be reconstructed into a path.
pub public_bucket_location_is_hash: bool,
pub maybe_successfully_completed_at: Option<DateTime<Utc>>,
}
/// NB: DO NOT EXPOSE TO FRONTEND.
/// NB: Serialize is for internal moderator-only endpoints
/// This is used to gate access to job termination
#[derive(Debug, Default, Serialize)]
pub struct UserDetails {
pub maybe_creator_user_token: Option<UserToken>,
pub maybe_creator_anonymous_visitor_token: Option<AnonymousVisitorTrackingToken>,
pub creator_ip_address: String,
}
use crate::queries::generic_inference::web::job_status::{GenericInferenceJobStatus, RequestDetails, ResultDetails, UserDetails};
/// Look up job status.
/// Returns Ok(None) when the record cannot be found.
@@ -151,7 +80,9 @@ SELECT
jobs.updated_at,
jobs.first_started_at as maybe_first_started_at,
jobs.successfully_completed_at as maybe_successfully_completed_at
jobs.successfully_completed_at as maybe_successfully_completed_at,
NOW() as database_clock
FROM generic_inference_jobs as jobs
@@ -283,6 +214,7 @@ fn raw_record_to_public_result(record: RawGenericInferenceJobStatus) -> GenericI
maybe_model_token: record.maybe_model_token,
maybe_model_title: maybe_model_title.map(|title| title.to_string()),
maybe_raw_inference_text: record.maybe_raw_inference_text,
maybe_inference_args: record.maybe_inference_args,
maybe_style_name,
},
maybe_result_details,
@@ -294,6 +226,7 @@ fn raw_record_to_public_result(record: RawGenericInferenceJobStatus) -> GenericI
is_keepalive_required: i8_to_bool(record.is_keepalive_required),
created_at: record.created_at,
updated_at: record.updated_at,
database_clock: DateTime::from_naive_utc_and_offset(record.database_clock, Utc),
}
}
@@ -339,6 +272,10 @@ struct RawGenericInferenceJobStatus {
pub maybe_first_started_at: Option<DateTime<Utc>>,
pub maybe_successfully_completed_at: Option<DateTime<Utc>>,
// NB: Typed query can't convert to Utc, so we use NaiveDateTime and do the type conversion ourselves.
// The database server *should* be reporting in UTC.
pub database_clock: NaiveDateTime,
}
#[cfg(test)]
@@ -0,0 +1,88 @@
use chrono::{DateTime, Utc};
use enums::by_table::generic_inference_jobs::frontend_failure_category::FrontendFailureCategory;
use enums::by_table::generic_inference_jobs::inference_category::InferenceCategory;
use enums::common::job_status_plus::JobStatusPlus;
use enums::no_table::style_transfer::style_transfer_name::StyleTransferName;
use tokens::tokens::anonymous_visitor_tracking::AnonymousVisitorTrackingToken;
use tokens::tokens::generic_inference_jobs::InferenceJobToken;
use tokens::tokens::users::UserToken;
/// Shared inference job status struct for several queries (list, get, batch get)
/// NB: Serialization is for internal endpoints, not for returning to end-users.
#[derive(Debug, Default, Serialize)]
pub struct GenericInferenceJobStatus {
pub job_token: InferenceJobToken,
pub status: JobStatusPlus,
pub attempt_count: u16,
pub maybe_assigned_worker: Option<String>,
pub maybe_assigned_cluster: Option<String>,
pub maybe_first_started_at: Option<DateTime<Utc>>,
pub maybe_frontend_failure_category: Option<FrontendFailureCategory>,
pub request_details: RequestDetails,
pub maybe_result_details: Option<ResultDetails>,
pub user_details: UserDetails,
pub is_keepalive_required: bool,
pub created_at: DateTime<Utc>,
pub updated_at: DateTime<Utc>,
/// This is MySQL's NOW(), which will let us do duration math without clock skew.
pub database_clock: DateTime<Utc>,
}
/// NB: Serialize is for internal moderator-only endpoints
/// Details about the user's original inference request
/// (We may want to present it in the "pending" UI.)
#[derive(Debug, Default, Serialize)]
pub struct RequestDetails {
pub inference_category: InferenceCategory,
pub maybe_model_type: Option<String>, // TODO: Strongly type
pub maybe_model_token: Option<String>,
pub maybe_model_title: Option<String>,
/// TTS input. In the future, perhaps voice conversion SST
pub maybe_raw_inference_text: Option<String>,
/// Raw JSON payload of polymorphic inference args
pub maybe_inference_args: Option<String>,
/// For Comfy / Video Style Transfer jobs, this might include
/// the name of the selected style.
pub maybe_style_name: Option<StyleTransferName>,
}
/// NB: Serialize is for internal moderator-only endpoints
/// Details about the generated result
#[derive(Debug, Default, Serialize)]
pub struct ResultDetails {
pub entity_type: String,
pub entity_token: String,
/// The bucket storage hash (for vc and media_files) or full path (for tts)
pub public_bucket_location_or_hash: String,
pub maybe_media_file_public_bucket_prefix: Option<String>,
pub maybe_media_file_public_bucket_extension: Option<String>,
/// Whether the location is a full path (for tts) or a hash (for vc) that
/// needs to be reconstructed into a path.
pub public_bucket_location_is_hash: bool,
pub maybe_successfully_completed_at: Option<DateTime<Utc>>,
}
/// NB: DO NOT EXPOSE TO FRONTEND.
/// NB: Serialize is for internal moderator-only endpoints
/// This is used to gate access to job termination
#[derive(Debug, Default, Serialize)]
pub struct UserDetails {
pub maybe_creator_user_token: Option<UserToken>,
pub maybe_creator_anonymous_visitor_token: Option<AnonymousVisitorTrackingToken>,
pub creator_ip_address: String,
}
@@ -9,7 +9,6 @@ use sqlx::pool::PoolConnection;
use enums::by_table::generic_inference_jobs::frontend_failure_category::FrontendFailureCategory;
use enums::by_table::generic_inference_jobs::inference_category::InferenceCategory;
use enums::common::job_status_plus::JobStatusPlus;
use enums::no_table::style_transfer::style_transfer_name::StyleTransferName;
use enums::traits::mysql_from_row::MySqlFromRow;
use errors::AnyhowResult;
use tokens::tokens::anonymous_visitor_tracking::AnonymousVisitorTrackingToken;
@@ -19,78 +18,7 @@ use tokens::traits::mysql_token_from_row::MySqlTokenFromRow;
use crate::helpers::boolean_converters::i8_to_bool;
use crate::payloads::generic_inference_args::generic_inference_args::{GenericInferenceArgs, PolymorphicInferenceArgs};
// NB: Serialization is for internal endpoints, not for returning to end-users.
#[derive(Debug, Default, Serialize)]
pub struct GenericInferenceJobStatus {
pub job_token: InferenceJobToken,
pub status: JobStatusPlus,
pub attempt_count: u16,
pub maybe_assigned_worker: Option<String>,
pub maybe_assigned_cluster: Option<String>,
pub maybe_first_started_at: Option<DateTime<Utc>>,
pub maybe_frontend_failure_category: Option<FrontendFailureCategory>,
pub request_details: RequestDetails,
pub maybe_result_details: Option<ResultDetails>,
pub user_details: UserDetails,
pub is_keepalive_required: bool,
pub created_at: DateTime<Utc>,
pub updated_at: DateTime<Utc>,
}
/// NB: Serialize is for internal moderator-only endpoints
/// Details about the user's original inference request
/// (We may want to present it in the "pending" UI.)
#[derive(Debug, Default, Serialize)]
pub struct RequestDetails {
pub inference_category: InferenceCategory,
pub maybe_model_type: Option<String>, // TODO: Strongly type
pub maybe_model_token: Option<String>,
pub maybe_model_title: Option<String>,
/// TTS input. In the future, perhaps voice conversion SST
pub maybe_raw_inference_text: Option<String>,
/// For Comfy / Video Style Transfer jobs, this might include
/// the name of the selected style.
pub maybe_style_name: Option<StyleTransferName>,
}
/// NB: Serialize is for internal moderator-only endpoints
/// Details about the generated result
#[derive(Debug, Default, Serialize)]
pub struct ResultDetails {
pub entity_type: String,
pub entity_token: String,
/// The bucket storage hash (for vc and media_files) or full path (for tts)
pub public_bucket_location_or_hash: String,
pub maybe_media_file_public_bucket_prefix: Option<String>,
pub maybe_media_file_public_bucket_extension: Option<String>,
/// Whether the location is a full path (for tts) or a hash (for vc) that
/// needs to be reconstructed into a path.
pub public_bucket_location_is_hash: bool,
pub maybe_successfully_completed_at: Option<DateTime<Utc>>,
}
/// NB: DO NOT EXPOSE TO FRONTEND.
/// NB: Serialize is for internal moderator-only endpoints
/// This is used to gate access to job termination
#[derive(Debug, Default, Serialize)]
pub struct UserDetails {
pub maybe_creator_user_token: Option<UserToken>,
pub maybe_creator_anonymous_visitor_token: Option<AnonymousVisitorTrackingToken>,
pub creator_ip_address: String,
}
use crate::queries::generic_inference::web::job_status::{GenericInferenceJobStatus, RequestDetails, ResultDetails, UserDetails};
pub struct ListSessionJobsForUserArgs<'a> {
// TODO(bt,2024-04-23): Support AnonymousVisitorToken
@@ -239,7 +167,9 @@ SELECT
jobs.updated_at,
jobs.first_started_at as maybe_first_started_at,
jobs.successfully_completed_at as maybe_successfully_completed_at
jobs.successfully_completed_at as maybe_successfully_completed_at,
NOW() as database_clock
FROM generic_inference_jobs as jobs
@@ -354,6 +284,7 @@ fn raw_records_to_public_result(records: Vec<RawGenericInferenceJobStatus>) -> V
maybe_model_token: record.maybe_model_token,
maybe_model_title: maybe_model_title.map(|title| title.to_string()),
maybe_raw_inference_text: record.maybe_raw_inference_text,
maybe_inference_args: record.maybe_inference_args,
maybe_style_name,
},
maybe_result_details,
@@ -365,6 +296,7 @@ fn raw_records_to_public_result(records: Vec<RawGenericInferenceJobStatus>) -> V
is_keepalive_required: i8_to_bool(record.is_keepalive_required),
created_at: record.created_at,
updated_at: record.updated_at,
database_clock: record.database_clock,
}
})
.collect::<Vec<_>>()
@@ -412,6 +344,8 @@ struct RawGenericInferenceJobStatus {
pub maybe_first_started_at: Option<DateTime<Utc>>,
pub maybe_successfully_completed_at: Option<DateTime<Utc>>,
pub database_clock: DateTime<Utc>,
}
// NB(bt,2023-12-05): There's an issue with type hinting in the `as` clauses with QueryBuilder (or
@@ -458,6 +392,7 @@ impl FromRow<'_, MySqlRow> for RawGenericInferenceJobStatus {
maybe_successfully_completed_at: row.try_get("maybe_successfully_completed_at")?,
created_at: row.try_get("created_at")?,
updated_at: row.try_get("updated_at")?,
database_clock: row.try_get("database_clock")?,
})
}
}
@@ -3,6 +3,7 @@ pub mod dismiss_finished_jobs_for_user;
pub mod get_inference_job_status;
pub mod get_pending_inference_job_count;
pub mod insert_generic_inference_job;
pub mod job_status;
pub mod kill_generic_inference_jobs;
pub mod list_session_jobs;
pub mod mark_generic_inference_job_cancelled_by_user;
@@ -0,0 +1 @@
pub mod delete_media_file;
@@ -1,4 +1,7 @@
pub mod rename_media_file;
pub mod set_media_file_cover_image;
pub mod update_media_file_animation_type;
pub mod update_media_file_engine_category;
pub mod update_media_file_stored_cloud_contents;
pub mod update_media_file_visibility;
pub mod update_media_file_visibility_transactional;
@@ -0,0 +1,27 @@
use sqlx::MySqlPool;
use enums::by_table::media_files::media_file_animation_type::MediaFileAnimationType;
use errors::AnyhowResult;
use tokens::tokens::media_files::MediaFileToken;
pub async fn update_media_file_animation_type(
media_file_token: &MediaFileToken,
maybe_animation_type: Option<MediaFileAnimationType>,
mysql_pool: &MySqlPool
) -> AnyhowResult<()> {
let _r = sqlx::query!(
r#"
UPDATE media_files
SET
maybe_animation_type = ?
WHERE
token = ?
LIMIT 1
"#,
maybe_animation_type.map(|t| t.to_str()),
media_file_token,
)
.execute(mysql_pool)
.await?;
Ok(())
}
@@ -0,0 +1,27 @@
use sqlx::MySqlPool;
use enums::by_table::media_files::media_file_engine_category::MediaFileEngineCategory;
use errors::AnyhowResult;
use tokens::tokens::media_files::MediaFileToken;
pub async fn update_media_file_engine_category(
media_file_token: &MediaFileToken,
engine_category: MediaFileEngineCategory,
mysql_pool: &MySqlPool
) -> AnyhowResult<()> {
let _r = sqlx::query!(
r#"
UPDATE media_files
SET
maybe_engine_category = ?
WHERE
token = ?
LIMIT 1
"#,
engine_category.to_str(),
media_file_token,
)
.execute(mysql_pool)
.await?;
Ok(())
}
@@ -16,7 +16,7 @@ use tokens::tokens::users::UserToken;
use crate::payloads::prompt_args::prompt_inner_payload::PromptInnerPayload;
#[derive(Serialize)]
#[derive(Serialize, Clone)]
pub struct MediaFilesByTokensRecord {
pub token: MediaFileToken,
@@ -70,6 +70,12 @@ pub async fn batch_get_media_files_by_tokens(
can_see_deleted: bool
) -> AnyhowResult<Vec<MediaFilesByTokensRecord>> {
if media_file_tokens.is_empty() {
// NB: We should always eagerly return, but if we don't, the query builder will build an
// invalid query.
return Ok(vec![]);
}
let mut connection = mysql_pool.acquire().await?;
let raw_media_files: Vec<RawMediaFileJoinUser> = get_raw_media_files_by_tokens(&mut connection, media_file_tokens, can_see_deleted).await?;
@@ -1,3 +1,4 @@
pub mod batch_get_media_files;
pub mod batch_get_media_files_by_tokens;
pub mod get_media_file;
pub mod get_media_file;
pub mod get_media_file_for_inference;
@@ -1,8 +1,6 @@
pub mod create;
pub mod delete_media_file;
pub mod delete;
pub mod edit;
pub mod get;
pub mod get_media_file_for_inference;
pub mod list;
pub mod rename_media_file;
pub mod upsert;
@@ -64,6 +64,11 @@ pub async fn batch_get_model_weights_for_elastic_search_backfill<'e, 'c, E>(
) -> AnyhowResult<Vec<ModelWeightForElasticsearchRecord>>
where E: 'e + Executor<'c, Database=MySql>
{
if tokens.is_empty() {
// NB: We should always eagerly return, but if we don't, the query builder will build an
// invalid query.
return Ok(Vec::new());
}
let maybe_models
= list_model_weights(mysql_executor, tokens)
@@ -81,6 +81,10 @@ impl SessionUserRecord {
pub fn get_strongly_typed_user_token(&self) -> UserToken {
self.user_token.clone()
}
pub fn is_mod(&self) -> bool {
self.can_ban_users
}
}
pub async fn get_user_session_by_token_pooled_connection(
@@ -54,3 +54,4 @@ tokio = { version = "1.38.0", features = ["macros"] }
serde = "1.0.125"
serde_derive = "1.0.125"
serde_json = "1.0.108"
futures = "0.3.29"
+23 -7
View File
@@ -16,6 +16,7 @@
// Strict
//#![forbid(warnings)]
use std::sync::Arc;
use elasticsearch::Elasticsearch;
use elasticsearch::http::transport::Transport;
use log::info;
@@ -28,10 +29,10 @@ use config::shared_constants::DEFAULT_RUST_LOG;
use errors::AnyhowResult;
use crate::job_state::{JobState, SleepConfigs};
use crate::main_loop::main_loop;
use crate::tasks::update_all_model_weights::update_all_model_weights::update_all_model_weights;
use crate::tasks::update_engine_media_files::update_engine_media_files::update_engine_media_files;
pub mod job_state;
pub mod main_loop;
pub mod tasks;
#[tokio::main]
async fn main() -> AnyhowResult<()> {
@@ -47,8 +48,7 @@ async fn main() -> AnyhowResult<()> {
let mysql_pool = get_mysql_pool().await?;
let elasticsearch = get_elasticsearch_client()?;
let job_state = JobState {
let job_state = Arc::new(JobState {
mysql_pool,
elasticsearch,
sleep_config: SleepConfigs {
@@ -58,9 +58,25 @@ async fn main() -> AnyhowResult<()> {
between_error_wait_millis: easyenv::get_env_num("BETWEEN_ERROR_WAIT_MILLIS", 10_000)?,
between_no_updates_wait_millis: easyenv::get_env_num("BETWEEN_NO_UPDATES_WAIT_MILLIS", 20_000)?,
},
};
});
let _r = main_loop(job_state).await;
info!("Starting thread to update all model weights...");
let job_state_1 = job_state.clone();
let handle_1 = tokio::task::spawn(async move {
let _r = update_all_model_weights(job_state_1).await;
});
info!("Starting thread to update engine media files...");
let job_state_2 = job_state.clone();
let handle_2 = tokio::task::spawn(async move {
let _r = update_engine_media_files(job_state_2).await;
});
futures::future::join_all([handle_1, handle_2]).await;
Ok(())
}
@@ -0,0 +1,2 @@
pub mod update_all_model_weights;
pub mod update_engine_media_files;
@@ -0,0 +1 @@
pub mod update_all_model_weights;
@@ -1,3 +1,4 @@
use std::sync::Arc;
use std::time::Duration;
use anyhow::anyhow;
@@ -15,7 +16,7 @@ use mysql_queries::queries::model_weights::list::list_model_weight_tokens_update
use crate::job_state::JobState;
pub async fn main_loop(job_state: JobState) {
pub async fn update_all_model_weights(job_state: Arc<JobState>) {
// TODO(bt,2024-02-05): Write this cursor to Redis so job can resume without reindexing everything.
let mut cursor = DateTime::UNIX_EPOCH;
@@ -0,0 +1 @@
pub mod update_engine_media_files;
@@ -0,0 +1,190 @@
use std::collections::HashSet;
use std::iter::FromIterator;
use std::sync::Arc;
use std::time::Duration;
use anyhow::anyhow;
use elasticsearch::{BulkOperation, BulkParts, Elasticsearch};
use log::{error, info};
use serde_json::Value;
use elasticsearch_schema::documents::media_file_document::{MEDIA_FILE_INDEX, MediaFileDocument};
use elasticsearch_schema::traits::document::Document;
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 errors::AnyhowResult;
use mysql_queries::queries::media_files::list::list_media_files_for_elastic_search_backfill_using_cursor::{list_media_files_for_elastic_search_backfill_using_cursor, ListArgs, MediaFileForElasticsearchRecord};
use crate::job_state::JobState;
const PAGE_SIZE : usize = 1000;
pub async fn update_engine_media_files(job_state: Arc<JobState>) {
// TODO(bt,2024-02-05): Write this cursor to Redis so job can resume without reindexing everything.
let mut cursor = 0;
loop {
info!("Main loop; cursor @ {:?}", &cursor);
// TODO(bt,2024-07-07): We don't have any recoverability if the update feed gets stuck. We'll stay in
// this loop and cease making progress. Furthermore, we have no observability for when this happens. :(
let result = with_database_main_loop(&mut cursor, &job_state).await;
if let Err(err) = result {
error!("Error in main loop: {:?}", err);
std::thread::sleep(Duration::from_millis(job_state.sleep_config.between_error_wait_millis));
}
// NB: Function should never return in success case, but we'll sleep just in case.
}
}
pub async fn with_database_main_loop(cursor: &mut usize, job_state: &JobState) -> AnyhowResult<()> {
loop {
info!("Querying tokens updated since: {:?}", &cursor);
// TODO(bt,2024-07-07): media_files doesn't have an index on updated_at, so we can't simply scan for updates.
// We have to do a full table walk, which is unfortunate. It should work at our current scale, but in the future
// we'll need to remedy this.
let mut results = list_media_files_for_elastic_search_backfill_using_cursor(ListArgs {
mysql_pool: &job_state.mysql_pool,
page_size: PAGE_SIZE,
cursor: *cursor,
maybe_filter_engine_categories: Some(&HashSet::from_iter(vec![
MediaFileEngineCategory::Animation,
MediaFileEngineCategory::Character,
MediaFileEngineCategory::Creature,
MediaFileEngineCategory::Expression,
MediaFileEngineCategory::ImagePlane,
MediaFileEngineCategory::Location,
MediaFileEngineCategory::Object,
MediaFileEngineCategory::Scene,
MediaFileEngineCategory::SetDressing,
MediaFileEngineCategory::Skybox,
MediaFileEngineCategory::VideoPlane,
])),
maybe_filter_media_types: Some(&HashSet::from_iter(vec![
// Engine types
MediaFileType::Csv,
MediaFileType::Fbx,
MediaFileType::Glb,
MediaFileType::SceneJson,
// Image types
MediaFileType::Gif,
MediaFileType::Jpg,
MediaFileType::Png,
])),
maybe_filter_media_classes: Some(&HashSet::from_iter(vec![
MediaFileClass::Image,
MediaFileClass::Dimensional,
])),
}).await?;
if results.is_empty() {
info!("No records after cursor: {:?}. Resetting to 0.", &cursor);
*cursor = 0;
std::thread::sleep(Duration::from_millis(job_state.sleep_config.between_no_updates_wait_millis));
continue;
}
info!("Found {} updated records", results.len());
while !results.is_empty() {
let last = 50.min(results.len());
let drained_results= results.drain(0..last)
.into_iter()
.collect::<Vec<_>>();
for record in drained_results {
let id = record.id;
create_document_from_record(&job_state.elasticsearch, record).await?;
*cursor = id as usize;
std::thread::sleep(Duration::from_millis(job_state.sleep_config.between_es_writes_wait_millis));
}
std::thread::sleep(Duration::from_millis(job_state.sleep_config.between_job_batch_wait_millis));
}
info!("Up to date at cursor = {:?}", &cursor);
std::thread::sleep(Duration::from_millis(job_state.sleep_config.between_query_wait_millis));
}
}
async fn create_document_from_record(elasticsearch: &Elasticsearch, record: MediaFileForElasticsearchRecord) -> AnyhowResult<()> {
info!("Create record for {:?} - {:?}", record.token, record.maybe_title);
let is_deleted = record.user_deleted_at.is_some() || record.mod_deleted_at.is_some();
let document = MediaFileDocument {
token: record.token,
media_class: record.media_class,
media_type: record.media_type,
maybe_media_subtype: record.maybe_media_subtype,
maybe_engine_category: record.maybe_engine_category,
maybe_animation_type: record.maybe_animation_type,
maybe_mime_type: record.maybe_mime_type,
public_bucket_directory_hash: record.public_bucket_directory_hash,
maybe_public_bucket_prefix: record.maybe_public_bucket_prefix,
maybe_public_bucket_extension: record.maybe_public_bucket_extension,
creator_set_visibility: record.creator_set_visibility,
maybe_title: record.maybe_title.clone(),
maybe_title_as_keyword: record.maybe_title,
maybe_cover_image_media_file_token: record.maybe_cover_image_media_file_token,
maybe_cover_image_public_bucket_hash: record.maybe_cover_image_public_bucket_hash,
maybe_cover_image_public_bucket_prefix: record.maybe_cover_image_public_bucket_prefix,
maybe_cover_image_public_bucket_extension: record.maybe_cover_image_public_bucket_extension,
maybe_creator_user_token: record.maybe_creator_user_token,
maybe_creator_username: record.maybe_creator_username,
maybe_creator_display_name: record.maybe_creator_display_name,
maybe_creator_gravatar_hash: record.maybe_creator_gravatar_hash,
is_featured: record.is_featured,
created_at: record.created_at,
updated_at: record.updated_at,
user_deleted_at: record.user_deleted_at,
mod_deleted_at: record.mod_deleted_at,
is_deleted,
};
let op : BulkOperation<_> = BulkOperation::index(&document)
.id(document.get_document_id())
.into();
let response = elasticsearch
.bulk(BulkParts::Index(MEDIA_FILE_INDEX))
.body(vec![op])
.send()
.await?;
let json: Value = response.json().await?;
let had_errors = json["errors"].as_bool().unwrap_or(false);
if had_errors {
let failed: Vec<&Value> = json["items"]
.as_array()
.unwrap()
.iter()
.filter(|v| !v["error"].is_null())
.collect();
error!("Errors during indexing. Failures: {}", failed.len());
return Err(anyhow!("Errors during indexing. Failures: {}", failed.len()));
}
Ok(())
}
@@ -21,7 +21,7 @@ use jobs_common::job_progress_reporter::job_progress_reporter::JobProgressReport
use mimetypes::mimetype_for_file::get_mimetype_for_file;
use mysql_queries::queries::generic_inference::job::list_available_generic_inference_jobs::AvailableInferenceJob;
use mysql_queries::queries::media_files::create::insert_media_file_generic::{insert_media_file_generic, InsertArgs};
use mysql_queries::queries::media_files::get_media_file_for_inference::MediaFileForInference;
use mysql_queries::queries::media_files::get::get_media_file_for_inference::MediaFileForInference;
use subprocess_common::command_runner::command_runner_args::{FileOrCreate, RunAsSubprocessArgs, StreamRedirection};
use crate::job::job_loop::job_success_result::{JobSuccessResult, ResultEntity};
@@ -3,7 +3,7 @@ use log::error;
use enums::by_table::generic_inference_jobs::inference_model_type::InferenceModelType;
use mysql_queries::queries::generic_inference::job::list_available_generic_inference_jobs::AvailableInferenceJob;
use mysql_queries::queries::media_files::get_media_file_for_inference::get_media_file_for_inference;
use mysql_queries::queries::media_files::get::get_media_file_for_inference::get_media_file_for_inference;
use tokens::tokens::media_files::MediaFileToken;
use crate::job::job_loop::job_success_result::JobSuccessResult;
@@ -2,7 +2,7 @@ use anyhow::anyhow;
use log::error;
use mysql_queries::queries::generic_inference::job::list_available_generic_inference_jobs::AvailableInferenceJob;
use mysql_queries::queries::media_files::get_media_file_for_inference::get_media_file_for_inference;
use mysql_queries::queries::media_files::get::get_media_file_for_inference::get_media_file_for_inference;
use tokens::tokens::media_files::MediaFileToken;
use crate::job::job_loop::job_success_result::JobSuccessResult;
@@ -20,7 +20,7 @@ use mimetypes::mimetype_for_file::get_mimetype_for_file;
use mysql_queries::payloads::generic_inference_args::generic_inference_args::PolymorphicInferenceArgs;
use mysql_queries::queries::generic_inference::job::list_available_generic_inference_jobs::AvailableInferenceJob;
use mysql_queries::queries::media_files::create::insert_media_file_generic::{insert_media_file_generic, InsertArgs};
use mysql_queries::queries::media_files::get_media_file_for_inference::MediaFileForInference;
use mysql_queries::queries::media_files::get::get_media_file_for_inference::MediaFileForInference;
use subprocess_common::command_runner::command_runner_args::{RunAsSubprocessArgs, StreamRedirection};
use crate::job::job_loop::job_success_result::{JobSuccessResult, ResultEntity};
@@ -2,7 +2,7 @@ use std::path::PathBuf;
use buckets::public::media_files::bucket_file_path::MediaFileBucketPath;
use buckets::public::media_uploads::bucket_file_path::MediaUploadOriginalFilePath;
use mysql_queries::queries::media_files::get_media_file_for_inference::MediaFileForInference;
use mysql_queries::queries::media_files::get::get_media_file_for_inference::MediaFileForInference;
use mysql_queries::queries::media_uploads::get_media_upload_for_inference::MediaUploadRecordForInference;
/// Dynamic payload for inference job media that also serves as an adapter
@@ -4,7 +4,7 @@ use log::{error, info};
use enums::by_table::generic_inference_jobs::inference_input_source_token_type::InferenceInputSourceTokenType;
use migration::voice_conversion::query_vc_model_for_migration::{query_vc_model_for_migration, VcModelType};
use mysql_queries::queries::generic_inference::job::list_available_generic_inference_jobs::AvailableInferenceJob;
use mysql_queries::queries::media_files::get_media_file_for_inference::get_media_file_for_inference;
use mysql_queries::queries::media_files::get::get_media_file_for_inference::get_media_file_for_inference;
use mysql_queries::queries::media_uploads::get_media_upload_for_inference::get_media_upload_for_inference;
use tokens::tokens::media_files::MediaFileToken;
use tokens::tokens::media_uploads::MediaUploadToken;
@@ -1,14 +1,14 @@
use std::path::PathBuf;
use std::time::{Duration, Instant};
use reqwest::StatusCode;
use reqwest::StatusCode;
use tokio::process::Command;
use tokio::time::sleep;
use easyenv::get_env_string_optional;
use easyenv::get_env_string_optional;
use errors::AnyhowResult;
use crate::job::job_types::workflow::comfy_ui::video_style_transfer::comfy_ui_inference_command::ComfyInferenceCommand;
use crate::job::job_types::workflow::comfy_ui::video_style_transfer::util::comfy_ui_inference_command::ComfyInferenceCommand;
use crate::util::common_commands::ffmpeg_command_runner::FfmpegCommandRunner;
use crate::util::common_commands::ffmpeg_logo_watermark_command::FfmpegLogoWatermarkCommand;
@@ -1,5 +1,3 @@
pub mod comfy_ui_inference_command;
pub mod process_job;
pub mod steps;
pub mod video_paths;
pub mod write_workflow_prompt;
pub mod util;
@@ -44,17 +44,19 @@ use videos::ffprobe_get_dimensions::ffprobe_get_dimensions;
use crate::job::job_loop::job_success_result::{JobSuccessResult, ResultEntity};
use crate::job::job_loop::process_single_job_error::ProcessSingleJobError;
use crate::job::job_types::workflow::comfy_ui::comfy_process_job_args::ComfyProcessJobArgs;
use crate::job::job_types::workflow::comfy_ui::video_style_transfer::comfy_ui_inference_command::{InferenceArgs, InferenceDetails};
use crate::job::job_types::workflow::comfy_ui::video_style_transfer::steps::check_and_validate_job::check_and_validate_job;
use crate::job::job_types::workflow::comfy_ui::video_style_transfer::steps::download_global_ipa_image::{download_global_ipa_image, DownloadGlobalIpaImageArgs};
use crate::job::job_types::workflow::comfy_ui::video_style_transfer::steps::download_input_video::{download_input_video, DownloadInputVideoArgs};
use crate::job::job_types::workflow::comfy_ui::video_style_transfer::steps::download_input_videos::{download_input_videos, DownloadInputVideoArgs};
use crate::job::job_types::workflow::comfy_ui::video_style_transfer::steps::post_process_add_watermark::{post_process_add_watermark, PostProcessAddWatermarkArgs};
use crate::job::job_types::workflow::comfy_ui::video_style_transfer::steps::post_process_restore_audio::{post_process_restore_audio, PostProcessRestoreVideoArgs};
use crate::job::job_types::workflow::comfy_ui::video_style_transfer::steps::preprocess_save_audio::{preprocess_save_audio, ProcessSaveAudioArgs};
use crate::job::job_types::workflow::comfy_ui::video_style_transfer::steps::preprocess_trim_and_resample_video::{preprocess_trim_and_resample_video, ProcessTrimAndResampleVideoArgs};
use crate::job::job_types::workflow::comfy_ui::video_style_transfer::steps::preprocess_trim_and_resample_videos::{preprocess_trim_and_resample_videos, ProcessTrimAndResampleVideoArgs};
use crate::job::job_types::workflow::comfy_ui::video_style_transfer::steps::validate_and_save_results::{SaveResultsArgs, validate_and_save_results};
use crate::job::job_types::workflow::comfy_ui::video_style_transfer::video_paths::VideoPaths;
use crate::job::job_types::workflow::comfy_ui::video_style_transfer::write_workflow_prompt::{WorkflowPromptArgs, write_workflow_prompt};
use crate::job::job_types::workflow::comfy_ui::video_style_transfer::util::comfy_dirs::ComfyDirs;
use crate::job::job_types::workflow::comfy_ui::video_style_transfer::util::comfy_ui_inference_command::{InferenceArgs, InferenceDetails};
use crate::job::job_types::workflow::comfy_ui::video_style_transfer::util::video_pathing::{PrimaryInputVideoAndPaths, SecondaryInputVideoAndPaths, VideoDownloads};
use crate::job::job_types::workflow::comfy_ui::video_style_transfer::util::video_pathing_deprecated::VideoPaths;
use crate::job::job_types::workflow::comfy_ui::video_style_transfer::util::write_workflow_prompt::{WorkflowPromptArgs, write_workflow_prompt};
use crate::job_dependencies::JobDependencies;
use crate::util::common_commands::ffmpeg_audio_replace_args::FfmpegAudioReplaceArgs;
use crate::util::common_commands::ffmpeg_logo_watermark_command::WatermarkArgs;
@@ -69,7 +71,7 @@ pub async fn process_job(args: ComfyProcessJobArgs<'_>) -> Result<JobSuccessResu
.new_generic_inference(job.inference_job_token.as_str())
.map_err(|e| ProcessSingleJobError::Other(anyhow!(e)))?;
let model_dependencies = args
let comfy_deps = args
.job_dependencies
.job
.job_specific_dependencies
@@ -95,7 +97,10 @@ pub async fn process_job(args: ComfyProcessJobArgs<'_>) -> Result<JobSuccessResu
// ===================== DOWNLOAD REQUIRED MODELS IF NOT EXIST ===================== //
let root_comfy_path = model_dependencies.inference_command.mounts_directory.clone();
// TODO: Replace all other paths with this
let comfy_dirs = ComfyDirs::new(&comfy_deps);
let root_comfy_path = comfy_deps.inference_command.mounts_directory.clone();
let remote_cloud_file_client = RemoteCloudFileClient::get_remote_cloud_file_client().await;
let remote_cloud_file_client = match remote_cloud_file_client {
@@ -158,7 +163,7 @@ pub async fn process_job(args: ComfyProcessJobArgs<'_>) -> Result<JobSuccessResu
workflow_path = write_workflow_prompt(WorkflowPromptArgs {
workflow_path: &workflow_path,
comfy_args: &comfy_args,
model_dependencies: &model_dependencies,
model_dependencies: &comfy_deps,
maybe_positive_prompt: comfy_args.positive_prompt.as_deref(),
maybe_negative_prompt: comfy_args.negative_prompt.as_deref(),
})?;
@@ -215,7 +220,7 @@ pub async fn process_job(args: ComfyProcessJobArgs<'_>) -> Result<JobSuccessResu
info!("Root comfy path: {:?}", &root_comfy_path);
info!("Job output path will be (fix this code! the job shouldn't set this path!): {:?}", &job_args.output_path);
let mut videos = VideoPaths::new(&root_comfy_path, job_args.output_path);
//let mut videos = VideoPaths::new(&root_comfy_path, job_args.output_path);
// TODO(bt,2024-04-20): Clean up this mess.
@@ -236,36 +241,37 @@ pub async fn process_job(args: ComfyProcessJobArgs<'_>) -> Result<JobSuccessResu
global_ipa_image = Some(results);
}
// ==================== DOWNLOAD VIDEO ==================== //
// ==================== DOWNLOAD VIDEOS ==================== //
let download_video = download_input_video(DownloadInputVideoArgs {
let mut videos = download_input_videos(DownloadInputVideoArgs {
job_args: &job_args,
videos: &videos,
comfy_dirs: &comfy_dirs,
mysql_pool: &deps.db.mysql_pool,
remote_cloud_file_client: &remote_cloud_file_client,
}).await?;
info!("Downloaded video!");
videos.debug_print_paths_after_download();
//videos.debug_print_paths_after_download();
if let Ok(Some(dimensions)) = ffprobe_get_dimensions(&videos.original_video_path) {
if let Ok(Some(dimensions)) = ffprobe_get_dimensions(&videos.input_video.original_download_path) {
info!("Download video dimensions: {}x{}", dimensions.width, dimensions.height);
}
// ========================= TRIM AND PREPROCESS VIDEO ======================== //
preprocess_trim_and_resample_video(ProcessTrimAndResampleVideoArgs {
preprocess_trim_and_resample_videos(ProcessTrimAndResampleVideoArgs {
comfy_args,
comfy_deps: model_dependencies,
videos: &videos,
comfy_deps,
comfy_dirs: &comfy_dirs,
videos: &mut videos,
})?;
// ========================= PREPROCESS AUDIO ======================== //
let mut lipsync_enabled = comfy_args.lipsync_enabled.unwrap_or(false);
if let Err(err) = preprocess_save_audio(ProcessSaveAudioArgs {
comfy_deps: model_dependencies,
comfy_deps,
videos: &mut videos,
}) {
error!("Audio extraction failed: {:?}", err);
@@ -284,18 +290,17 @@ pub async fn process_job(args: ComfyProcessJobArgs<'_>) -> Result<JobSuccessResu
// ==================== RUN COMFY INFERENCE ==================== //
info!("Ready for ComfyUI inference...");
info!("Preparing for ComfyUI inference...");
job_progress_reporter.log_status("running inference")
.map_err(|e| ProcessSingleJobError::Other(e))?;
info!("Running ComfyUI inference...");
let stderr_output_file = work_temp_dir.path().join("stderr.txt");
let stdout_output_file = work_temp_dir.path().join("stdout.txt");
let positive_prompt_file = work_temp_dir.path().join("positive_prompt.txt");
let negative_prompt_file = work_temp_dir.path().join("negative_prompt.txt");
let travel_prompt_file = work_temp_dir.path().join("travel_prompt.txt");
let maybe_positive_prompt_filename = comfy_args.positive_prompt
.as_deref()
@@ -315,15 +320,27 @@ pub async fn process_job(args: ComfyProcessJobArgs<'_>) -> Result<JobSuccessResu
})
.transpose()?;
let maybe_travel_prompt_filename = comfy_args.travel_prompt
.as_deref()
.map(|prompt| {
std::fs::write(&travel_prompt_file, prompt)
.map(|_| travel_prompt_file.as_path())
.map_err(|e| ProcessSingleJobError::IoError(e))
})
.transpose()?;
let inference_details = InferenceDetails::NewPythonArgs {
maybe_style: comfy_args.style_name,
maybe_positive_prompt_filename,
maybe_negative_prompt_filename,
maybe_travel_prompt_filename,
};
let inference_start_time = Instant::now();
let command_exit_status = model_dependencies
info!("Running ComfyUI inference...");
let command_exit_status = comfy_deps
.inference_command
.execute_inference(InferenceArgs {
stderr_output_file: &stderr_output_file,
@@ -335,10 +352,20 @@ pub async fn process_job(args: ComfyProcessJobArgs<'_>) -> Result<JobSuccessResu
disable_lcm: comfy_args.disable_lcm.unwrap_or(false),
use_cinematic: comfy_args.use_cinematic.unwrap_or(false),
maybe_strength: comfy_args.strength,
frame_skip: comfy_args.frame_skip,
global_ipa_image_filename: global_ipa_image
.as_ref()
.map(|image| path_to_string(&image.ipa_image_path)),
global_ipa_strength: None, // TODO: Expose a UI slider
depth_video_path: videos.maybe_depth.as_ref()
.map(|v| v.maybe_processed_path.as_deref())
.flatten(),
normal_video_path: videos.maybe_normal.as_ref()
.map(|v| v.maybe_processed_path.as_deref())
.flatten(),
outline_video_path: videos.maybe_outline.as_ref()
.map(|v| v.maybe_processed_path.as_deref())
.flatten(),
});
let inference_duration = Instant::now().duration_since(inference_start_time);
@@ -352,15 +379,15 @@ pub async fn process_job(args: ComfyProcessJobArgs<'_>) -> Result<JobSuccessResu
info!("Captured stduout output: {}", contents);
}
videos.debug_print_paths_after_comfy();
//videos.debug_print_paths_after_comfy();
if let Ok(Some(dimensions)) = ffprobe_get_dimensions(&videos.comfy_output_video_path) {
if let Ok(Some(dimensions)) = ffprobe_get_dimensions(&videos.input_video.comfy_output_video_path) {
info!("Comfy output video dimensions: {}x{}", dimensions.width, dimensions.height);
}
// ==================== CHECK OUTPUT FILE ======================== //
if let Err(err) = check_file_exists(&videos.comfy_output_video_path) {
if let Err(err) = check_file_exists(&videos.input_video.comfy_output_video_path) {
error!("Output file does not exist: {:?}", err);
error!("Inference failed: {:?}", command_exit_status);
@@ -373,6 +400,7 @@ pub async fn process_job(args: ComfyProcessJobArgs<'_>) -> Result<JobSuccessResu
safe_delete_temp_file(&stdout_output_file);
safe_delete_temp_directory(&work_temp_dir);
safe_delete_temp_file(&workflow_path);
safe_delete_all_input_videos(&videos);
// TODO(bt,2024-04-21): Not sure we want to delete the LoRA?
if let Some(lora_path) = maybe_lora_path {
@@ -383,37 +411,31 @@ pub async fn process_job(args: ComfyProcessJobArgs<'_>) -> Result<JobSuccessResu
safe_delete_temp_file(ipa_path.ipa_image_path);
}
safe_delete_temp_file(&videos.comfy_input_video_path);
safe_delete_temp_file(&videos.trimmed_resampled_video_path);
safe_delete_temp_file(&videos.trimmed_audio_path);
safe_delete_temp_file(&videos.original_video_path);
let output_dir = root_comfy_path.join("output");
safe_recursively_delete_files(&output_dir);
safe_recursively_delete_files(&comfy_dirs.comfy_output_dir);
return Err(ProcessSingleJobError::Other(anyhow!("Output file did not exist: {:?}",
&videos.comfy_output_video_path)));
&videos.input_video.comfy_output_video_path)));
}
// ==================== COPY BACK AUDIO ==================== //
post_process_restore_audio(PostProcessRestoreVideoArgs {
comfy_deps: model_dependencies,
comfy_deps,
videos: &mut videos,
});
// ==================== OPTIONAL WATERMARK ==================== //
post_process_add_watermark(PostProcessAddWatermarkArgs {
comfy_deps: model_dependencies,
comfy_deps,
videos: &mut videos,
});
// ==================== DEBUG ======================== //
videos.debug_print_paths_after_post_processing();
//videos.debug_print_paths_after_post_processing();
if let Ok(Some(dimensions)) = ffprobe_get_dimensions(&videos.get_final_video_to_upload()) {
if let Ok(Some(dimensions)) = ffprobe_get_dimensions(&videos.input_video.get_final_video_to_upload()) {
info!("Final video upload dimensions: {}x{}", dimensions.width, dimensions.height);
}
@@ -423,11 +445,10 @@ pub async fn process_job(args: ComfyProcessJobArgs<'_>) -> Result<JobSuccessResu
job,
deps: &deps,
job_args: &job_args,
comfy_deps: model_dependencies,
comfy_deps,
comfy_args,
videos: &videos,
job_progress_reporter: &mut job_progress_reporter,
download_video,
inference_duration,
}).await?;
@@ -444,12 +465,7 @@ pub async fn process_job(args: ComfyProcessJobArgs<'_>) -> Result<JobSuccessResu
safe_delete_temp_file(&stderr_output_file);
safe_delete_temp_file(&stdout_output_file);
safe_delete_temp_file(&videos.original_video_path);
safe_delete_temp_file(&videos.trimmed_resampled_video_path);
safe_delete_temp_file(&videos.comfy_output_video_path);
safe_delete_temp_file(videos.video_to_watermark());
safe_delete_temp_file(videos.get_final_video_to_upload());
safe_delete_temp_file(videos.get_non_watermarked_video_to_upload());
safe_delete_all_input_videos(&videos);
// TODO(bt,2024-03-01): Do we really want to delete the workflow, models, etc.?
@@ -489,3 +505,37 @@ pub async fn process_job(args: ComfyProcessJobArgs<'_>) -> Result<JobSuccessResu
inference_duration,
})
}
fn safe_delete_all_input_videos(videos: &VideoDownloads) {
safe_delete_primary_videos(&videos.input_video);
if let Some(depth) = &videos.maybe_depth {
safe_delete_secondary_videos(depth);
}
if let Some(normal) = &videos.maybe_normal {
safe_delete_secondary_videos(normal);
}
if let Some(outline) = &videos.maybe_outline {
safe_delete_secondary_videos(outline);
}
}
fn safe_delete_primary_videos(video: &PrimaryInputVideoAndPaths) {
safe_delete_temp_file(&video.original_download_path);
safe_delete_temp_file(&video.comfy_output_video_path);
safe_delete_temp_file(video.video_to_watermark());
safe_delete_temp_file(video.get_final_video_to_upload());
safe_delete_temp_file(video.get_non_watermarked_video_to_upload());
if let Some(processed_path) = &video.maybe_processed_path {
safe_delete_temp_file(processed_path);
}
}
fn safe_delete_secondary_videos(video: &SecondaryInputVideoAndPaths) {
safe_delete_temp_file(&video.original_download_path);
if let Some(processed_path) = &video.maybe_processed_path {
safe_delete_temp_file(processed_path);
}
}
@@ -17,6 +17,11 @@ pub struct JobArgs<'a> {
pub maybe_json_modifications: &'a Option<HashMap<String, NewValue>>,
pub maybe_lora_model: &'a Option<ModelWeightToken>,
pub maybe_input_file: &'a Option<MediaFileToken>,
// Secondary videos (optional) that aid processing
pub maybe_depth_input_file: Option<&'a MediaFileToken>,
pub maybe_normal_input_file: Option<&'a MediaFileToken>,
pub maybe_outline_input_file: Option<&'a MediaFileToken>,
}
pub fn check_and_validate_job(job: &AvailableInferenceJob) -> Result<JobArgs, ProcessSingleJobError> {
@@ -54,7 +59,6 @@ pub fn check_and_validate_job(job: &AvailableInferenceJob) -> Result<JobArgs, Pr
}
};
// check if job is legacy
let is_legacy = inference_args.maybe_json_modifications.is_some();
@@ -68,12 +72,14 @@ pub fn check_and_validate_job(job: &AvailableInferenceJob) -> Result<JobArgs, Pr
}
};
Ok(JobArgs {
workflow_source: &inference_args.maybe_workflow_config,
output_path,
maybe_lora_model: &inference_args.maybe_lora_model,
maybe_json_modifications: &inference_args.maybe_json_modifications,
maybe_input_file: &inference_args.maybe_input_file,
maybe_depth_input_file: inference_args.maybe_input_depth_file.as_ref(),
maybe_normal_input_file: inference_args.maybe_input_normal_file.as_ref(),
maybe_outline_input_file: inference_args.maybe_input_outline_file.as_ref(),
})
}
@@ -13,7 +13,7 @@ use videos::ffprobe_get_dimensions::ffprobe_get_dimensions;
use crate::job::job_loop::process_single_job_error::ProcessSingleJobError;
use crate::job::job_types::workflow::comfy_ui::video_style_transfer::steps::check_and_validate_job::JobArgs;
use crate::job::job_types::workflow::comfy_ui::video_style_transfer::video_paths::VideoPaths;
use crate::job::job_types::workflow::comfy_ui::video_style_transfer::util::video_pathing_deprecated::VideoPaths;
const DEFAULT_SUFFIX : &str = ".jpg";
@@ -1,84 +0,0 @@
use anyhow::anyhow;
use log::{error, info};
use sqlx::MySqlPool;
use buckets::public::media_files::bucket_file_path::MediaFileBucketPath;
use cloud_storage::remote_file_manager::remote_cloud_file_manager::RemoteCloudFileClient;
use filesys::path_to_string::path_to_string;
use mysql_queries::queries::media_files::get::get_media_file::{get_media_file, MediaFile};
use videos::ffprobe_get_dimensions::ffprobe_get_dimensions;
use crate::job::job_loop::process_single_job_error::ProcessSingleJobError;
use crate::job::job_types::workflow::comfy_ui::video_style_transfer::steps::check_and_validate_job::JobArgs;
use crate::job::job_types::workflow::comfy_ui::video_style_transfer::video_paths::VideoPaths;
pub struct VideoDownloadDetails {
pub input_video_media_file: MediaFile,
}
pub struct DownloadInputVideoArgs<'a> {
pub job_args: &'a JobArgs<'a>,
pub videos: &'a VideoPaths,
pub mysql_pool: &'a MySqlPool,
pub remote_cloud_file_client: &'a RemoteCloudFileClient,
}
pub async fn download_input_video(
args: DownloadInputVideoArgs<'_>
) -> Result<VideoDownloadDetails, ProcessSingleJobError> {
let input_media_file_token = match args.job_args.maybe_input_file {
None => return Err(ProcessSingleJobError::InvalidJob(anyhow!("No input video file provided"))),
Some(token) => token.clone(),
};
info!("Querying input media file by token: {:?} ...", &input_media_file_token);
let mut input_media_file = get_media_file(
&input_media_file_token,
false,
args.mysql_pool
).await?.ok_or_else(|| {
error!("input media_file not found: {:?}", &input_media_file_token);
ProcessSingleJobError::Other(anyhow!("input media_file not found: {:?}", &input_media_file_token))
})?;
if let Some(source_media_file_token) = &input_media_file.maybe_style_transfer_source_media_file_token {
// NB: We do this to avoid deep-frying the video.
// This also lets us hide the engine renders from users.
// This shouldn't ever become a deeply nested tree of children, but rather a single root
// with potentially many direct children.
// TODO(bt,2024-05-14): Perhaps fail open and use the first media file if the original
// isn't found?
info!("Looking up original style transfer source media file...");
input_media_file = get_media_file(
&source_media_file_token,
true, // NB: In case the original was deleted, allow this to continue.
args.mysql_pool
).await?.ok_or_else(|| {
error!("source input media_file not found: {:?}", &input_media_file_token);
ProcessSingleJobError::Other(anyhow!("source input media_file not found: {:?}",
&input_media_file_token))
})?;
}
let media_file_bucket_path = MediaFileBucketPath::from_object_hash(
&input_media_file.public_bucket_directory_hash,
input_media_file.maybe_public_bucket_prefix.as_deref(),
input_media_file.maybe_public_bucket_extension.as_deref());
info!("Input media file cloud bucket path: {:?}", media_file_bucket_path.get_full_object_path_str());
info!("Downloading input file to {:?}", args.videos.original_video_path);
args.remote_cloud_file_client.download_media_file(
&media_file_bucket_path,
path_to_string(&args.videos.original_video_path)
).await?;
info!("Downloaded video!");
Ok(VideoDownloadDetails {
input_video_media_file: input_media_file,
})
}
@@ -0,0 +1,216 @@
use std::path::{Path, PathBuf};
use anyhow::anyhow;
use log::{error, info};
use sqlx::MySqlPool;
use buckets::public::media_files::bucket_file_path::MediaFileBucketPath;
use cloud_storage::remote_file_manager::remote_cloud_file_manager::RemoteCloudFileClient;
use errors::AnyhowResult;
use filesys::path_to_string::path_to_string;
use mysql_queries::queries::media_files::get::batch_get_media_files_by_tokens::{batch_get_media_files_by_tokens, MediaFilesByTokensRecord};
use mysql_queries::queries::media_files::get::get_media_file::{get_media_file, MediaFile};
use tokens::tokens::media_files::MediaFileToken;
use videos::ffprobe_get_dimensions::ffprobe_get_dimensions;
use crate::job::job_loop::process_single_job_error::ProcessSingleJobError;
use crate::job::job_types::workflow::comfy_ui::comfy_ui_dependencies::ComfyDependencies;
use crate::job::job_types::workflow::comfy_ui::video_style_transfer::steps::check_and_validate_job::JobArgs;
use crate::job::job_types::workflow::comfy_ui::video_style_transfer::util::comfy_dirs::ComfyDirs;
use crate::job::job_types::workflow::comfy_ui::video_style_transfer::util::video_pathing::{PrimaryInputVideoAndPaths, SecondaryInputVideoAndPaths, VideoDownloads};
use crate::job::job_types::workflow::comfy_ui::video_style_transfer::util::video_pathing_deprecated::VideoPaths;
pub struct DownloadInputVideoArgs<'a> {
pub job_args: &'a JobArgs<'a>,
pub comfy_dirs: &'a ComfyDirs,
pub mysql_pool: &'a MySqlPool,
pub remote_cloud_file_client: &'a RemoteCloudFileClient,
}
pub async fn download_input_videos(
args: DownloadInputVideoArgs<'_>
) -> Result<VideoDownloads, ProcessSingleJobError> {
let video_downloads = download_primary_video(&args).await?;
let video_downloads = maybe_download_secondary_videos(video_downloads, &args).await?;
Ok(video_downloads)
}
// TODO: Consolidate primary video download with secondary download logic.
async fn download_primary_video(
args: &DownloadInputVideoArgs<'_>,
) -> Result<VideoDownloads, ProcessSingleJobError> {
let input_media_file_token = match args.job_args.maybe_input_file {
None => return Err(ProcessSingleJobError::InvalidJob(anyhow!("No primary input video file provided"))),
Some(token) => token.clone(),
};
info!("Querying primary input media file by token: {:?} ...", &input_media_file_token);
let mut input_media_file = get_media_file(
&input_media_file_token,
false,
args.mysql_pool
).await?.ok_or_else(|| {
error!("primary input media_file not found: {:?}", &input_media_file_token);
ProcessSingleJobError::Other(anyhow!("primary input media_file not found: {:?}", &input_media_file_token))
})?;
if let Some(source_media_file_token) = &input_media_file.maybe_style_transfer_source_media_file_token {
// NB: We do this to avoid deep-frying the video.
// This also lets us hide the engine renders from users.
// This shouldn't ever become a deeply nested tree of children, but rather a single root
// with potentially many direct children.
// TODO(bt,2024-05-14): Perhaps fail open and use the first media file if the original
// isn't found?
info!("Looking up original style transfer source media file...");
input_media_file = get_media_file(
&source_media_file_token,
true, // NB: In case the original was deleted, allow this to continue.
args.mysql_pool
).await?.ok_or_else(|| {
error!("source input media_file not found: {:?}", &input_media_file_token);
ProcessSingleJobError::Other(anyhow!("source input media_file not found: {:?}",
&input_media_file_token))
})?;
}
let media_file_bucket_path = MediaFileBucketPath::from_object_hash(
&input_media_file.public_bucket_directory_hash,
input_media_file.maybe_public_bucket_prefix.as_deref(),
input_media_file.maybe_public_bucket_extension.as_deref());
info!("Primary input media file cloud bucket path: {:?}", media_file_bucket_path.get_full_object_path_str());
// NB(bt,2024-07-09): This convention is muddled with the python side.
// We may not have flexibility to change this pathing for a while.
let download_path = args.comfy_dirs.comfy_input_dir.join("video.mp4");
info!("Downloading primary input file to {:?}", download_path);
args.remote_cloud_file_client.download_media_file(
&media_file_bucket_path,
path_to_string(&download_path)
).await?;
info!("Downloaded primary input video!");
// TODO: This monumentally sucks.
// The upstream shouldn't be telling us what to do about this at all.
let job_output_path = args.job_args.output_path;
Ok(VideoDownloads {
input_video: PrimaryInputVideoAndPaths::new(
input_media_file, &args.comfy_dirs, job_output_path),
maybe_depth: None,
maybe_normal: None,
maybe_outline: None,
})
}
#[derive(Clone, Copy)]
enum SecondaryVideoType {
Depth,
Normal,
Outline,
}
async fn maybe_download_secondary_videos(
mut video_downloads: VideoDownloads,
args: &DownloadInputVideoArgs<'_>
) -> Result<VideoDownloads, ProcessSingleJobError> {
const CAN_SEE_DELETED : bool = true; // We don't need to care about the deleted flag.
let mut tokens = Vec::with_capacity(3);
if let Some(token) = args.job_args.maybe_depth_input_file {
tokens.push(token.clone());
}
if let Some(token) = args.job_args.maybe_normal_input_file {
tokens.push(token.clone());
}
if let Some(token) = args.job_args.maybe_outline_input_file {
tokens.push(token.clone());
}
if tokens.is_empty() {
return Ok(video_downloads);
}
let results = batch_get_media_files_by_tokens(args.mysql_pool, &tokens, CAN_SEE_DELETED)
.await
.map_err(|err| {
ProcessSingleJobError::Other(anyhow!("error querying secondary videos: {:?}", &err))
})?;
if let Some(token) = args.job_args.maybe_depth_input_file {
video_downloads.maybe_depth =
download_secondary_video(SecondaryVideoType::Depth, token, &results, args).await?;
}
if let Some(token) = args.job_args.maybe_normal_input_file {
video_downloads.maybe_normal =
download_secondary_video(SecondaryVideoType::Normal, token, &results, args).await?;
}
if let Some(token) = args.job_args.maybe_outline_input_file {
video_downloads.maybe_outline =
download_secondary_video(SecondaryVideoType::Outline, token, &results, args).await?;
}
Ok(video_downloads)
}
async fn download_secondary_video(
secondary_video_type: SecondaryVideoType,
desired_token: &MediaFileToken,
all_video_media_files: &[MediaFilesByTokensRecord],
args: &DownloadInputVideoArgs<'_>
) -> Result<Option<SecondaryInputVideoAndPaths>, ProcessSingleJobError> {
let file_description = match secondary_video_type {
SecondaryVideoType::Depth => "depth video",
SecondaryVideoType::Normal => "normal video",
SecondaryVideoType::Outline => "outline video",
};
let download_path = match secondary_video_type {
SecondaryVideoType::Depth => args.comfy_dirs.comfy_input_dir.join("depth_download.mp4"),
SecondaryVideoType::Normal => args.comfy_dirs.comfy_input_dir.join("normal_download.mp4"),
SecondaryVideoType::Outline => args.comfy_dirs.comfy_input_dir.join("outline_download.mp4"),
};
let maybe_media_file = all_video_media_files
.iter()
.find(|media_file| media_file.token.eq(desired_token));
let media_file = match maybe_media_file {
Some(media_file) => media_file,
None => {
error!("secondary input media_file '{}' not found: {:?}", file_description, desired_token);
return Ok(None);
}
};
let media_file_bucket_path = MediaFileBucketPath::from_object_hash(
&media_file.public_bucket_directory_hash,
media_file.maybe_public_bucket_prefix.as_deref(),
media_file.maybe_public_bucket_extension.as_deref());
info!("Media file '{}' cloud bucket path: {:?}",file_description,
media_file_bucket_path.get_full_object_path_str());
info!("Downloading '{}' media file to {:?}", file_description, download_path);
args.remote_cloud_file_client.download_media_file(
&media_file_bucket_path,
path_to_string(&download_path)
).await?;
Ok(Some(SecondaryInputVideoAndPaths {
record: media_file.clone(),
original_download_path: download_path,
maybe_processed_path: None,
}))
}
@@ -1,8 +1,8 @@
pub mod check_and_validate_job;
pub mod download_global_ipa_image;
pub mod download_input_video;
pub mod download_input_videos;
pub mod post_process_add_watermark;
pub mod post_process_restore_audio;
pub mod preprocess_trim_and_resample_video;
pub mod validate_and_save_results;
pub mod preprocess_save_audio;
pub mod preprocess_trim_and_resample_videos;
pub mod validate_and_save_results;
@@ -3,12 +3,13 @@ use log::{error, info};
use filesys::check_file_exists::check_file_exists;
use crate::job::job_types::workflow::comfy_ui::comfy_ui_dependencies::ComfyDependencies;
use crate::job::job_types::workflow::comfy_ui::video_style_transfer::video_paths::VideoPaths;
use crate::job::job_types::workflow::comfy_ui::video_style_transfer::util::video_pathing::VideoDownloads;
use crate::job::job_types::workflow::comfy_ui::video_style_transfer::util::video_pathing_deprecated::VideoPaths;
use crate::util::common_commands::ffmpeg_logo_watermark_command::WatermarkArgs;
pub struct PostProcessAddWatermarkArgs<'a> {
pub comfy_deps: &'a ComfyDependencies,
pub videos: &'a mut VideoPaths,
pub videos: &'a mut VideoDownloads,
}
/// NB: Purposefully infallible.
@@ -21,13 +22,14 @@ pub fn post_process_add_watermark(
info!("Adding watermark...");
let output_video_fs_path_watermark = args.videos
.input_video
.comfy_output_video_path
.with_extension("_watermark.mp4");
let command_exit_status = args.comfy_deps
.ffmpeg_watermark_command
.execute(WatermarkArgs {
video_path: args.videos.video_to_watermark(),
video_path: args.videos.input_video.video_to_watermark(),
maybe_override_logo_path: None,
alpha: 0.6,
scale: 0.1, // NB: 0.1 is good for the Storyteller logo @ 2653x512 placed on 1024x576 output.
@@ -49,6 +51,6 @@ pub fn post_process_add_watermark(
if use_watermarked_file {
info!("Success generating watermarked file: {:?}", output_video_fs_path_watermark);
args.videos.watermarked_video_path = Some(output_video_fs_path_watermark);
args.videos.input_video.watermarked_video_path = Some(output_video_fs_path_watermark);
}
}
@@ -6,12 +6,13 @@ use subprocess_common::command_runner::command_runner_args::{RunAsSubprocessArgs
use crate::job::job_loop::process_single_job_error::ProcessSingleJobError;
use crate::job::job_types::workflow::comfy_ui::comfy_ui_dependencies::ComfyDependencies;
use crate::job::job_types::workflow::comfy_ui::video_style_transfer::video_paths::VideoPaths;
use crate::job::job_types::workflow::comfy_ui::video_style_transfer::util::video_pathing::VideoDownloads;
use crate::job::job_types::workflow::comfy_ui::video_style_transfer::util::video_pathing_deprecated::VideoPaths;
use crate::util::common_commands::ffmpeg_audio_replace_args::FfmpegAudioReplaceArgs;
pub struct PostProcessRestoreVideoArgs<'a> {
pub comfy_deps: &'a ComfyDependencies,
pub videos: &'a mut VideoPaths,
pub videos: &'a mut VideoDownloads,
}
/// NB: Purposefully infallible.
@@ -20,7 +21,18 @@ pub fn post_process_restore_audio(
) -> () {
info!("Restoring audio...");
// Use the original downloaded video if we didn't trim and resample it.
let input_video_file = args
.videos
.input_video
.maybe_processed_path
.as_deref()
.unwrap_or(
&args.videos.input_video.original_download_path
);
let output_video_fs_path_restored = args.videos
.input_video
.comfy_output_video_path
.with_extension("_restored.mp4");
@@ -29,8 +41,8 @@ pub fn post_process_restore_audio(
.ffmpeg_command_runner
.run_with_subprocess(RunAsSubprocessArgs {
args: Box::new(&FfmpegAudioReplaceArgs {
input_video_file: &args.videos.comfy_output_video_path,
input_audio_file: &args.videos.trimmed_resampled_video_path,
input_video_file: &args.videos.input_video.comfy_output_video_path,
input_audio_file: &input_video_file,
output_video_file: &output_video_fs_path_restored,
}),
stderr: StreamRedirection::None,
@@ -52,6 +64,6 @@ pub fn post_process_restore_audio(
if use_restored_audio {
info!("Success generating restored audio file: {:?}", output_video_fs_path_restored);
args.videos.audio_restored_video_path = Some(output_video_fs_path_restored);
args.videos.input_video.audio_restored_video_path = Some(output_video_fs_path_restored);
}
}
@@ -7,13 +7,14 @@ use subprocess_common::command_runner::command_runner_args::{RunAsSubprocessArgs
use crate::job::job_loop::process_single_job_error::ProcessSingleJobError;
use crate::job::job_types::workflow::comfy_ui::comfy_ui_dependencies::ComfyDependencies;
use crate::job::job_types::workflow::comfy_ui::video_style_transfer::video_paths::VideoPaths;
use crate::job::job_types::workflow::comfy_ui::video_style_transfer::util::video_pathing::VideoDownloads;
use crate::job::job_types::workflow::comfy_ui::video_style_transfer::util::video_pathing_deprecated::VideoPaths;
use crate::util::common_commands::ffmpeg_audio_replace_args::FfmpegAudioReplaceArgs;
use crate::util::common_commands::ffmpeg_extract_audio_args::FfmpegExtractAudioArgs;
pub struct ProcessSaveAudioArgs<'a> {
pub comfy_deps: &'a ComfyDependencies,
pub videos: &'a mut VideoPaths,
pub videos: &'a mut VideoDownloads,
}
pub fn preprocess_save_audio(
@@ -21,14 +22,23 @@ pub fn preprocess_save_audio(
) -> Result<(), ProcessSingleJobError> {
info!("Extracting audio...");
// Use the original downloaded video if we didn't trim and resample it.
let input_video_file = args
.videos
.input_video
.maybe_processed_path
.as_deref()
.unwrap_or(
&args.videos.input_video.original_download_path
);
let command_exit_status = args
.comfy_deps
.ffmpeg_command_runner
.run_with_subprocess(RunAsSubprocessArgs {
args: Box::new(&FfmpegExtractAudioArgs {
input_video_file: &args.videos.trimmed_resampled_video_path,
output_file: &args.videos.trimmed_audio_path
input_video_file,
output_file: &args.videos.input_video.trimmed_audio_path
}),
stderr: StreamRedirection::None,
stdout: StreamRedirection::None,
@@ -1,106 +0,0 @@
use std::path::Path;
use std::process::{Command, Stdio};
use anyhow::anyhow;
use log::{error, info};
use sqlx::MySqlPool;
use cloud_storage::remote_file_manager::remote_cloud_file_manager::RemoteCloudFileClient;
use filesys::path_to_string::path_to_string;
use mysql_queries::payloads::generic_inference_args::workflow_payload::WorkflowArgs;
use tokens::tokens::media_files::MediaFileToken;
use videos::ffprobe_get_dimensions::ffprobe_get_dimensions;
use crate::job::job_loop::process_single_job_error::ProcessSingleJobError;
use crate::job::job_types::workflow::comfy_ui::comfy_ui_dependencies::ComfyDependencies;
use crate::job::job_types::workflow::comfy_ui::video_style_transfer::video_paths::VideoPaths;
pub struct ProcessTrimAndResampleVideoArgs<'a> {
pub comfy_args: &'a WorkflowArgs,
pub comfy_deps: &'a ComfyDependencies,
pub videos: &'a VideoPaths,
}
pub fn preprocess_trim_and_resample_video(
args: ProcessTrimAndResampleVideoArgs<'_>
) -> Result<(), ProcessSingleJobError> {
let target_fps = args.comfy_args.target_fps.unwrap_or(24);
let trim_start_millis = args.comfy_args.trim_start_milliseconds
.or_else(|| args.comfy_args.trim_start_seconds.map(|s| s as u64 * 1_000))
.unwrap_or(0);
let trim_end_millis = args.comfy_args.trim_end_milliseconds
.or_else(|| args.comfy_args.trim_end_seconds.map(|s| s as u64 * 1_000))
.unwrap_or(3_000);
info!("trim start millis: {trim_start_millis}");
info!("trim end millis: {trim_end_millis}");
info!("target FPS: {target_fps}");
let skip_process_video = args.comfy_args.skip_process_video.unwrap_or(false);
if skip_process_video {
info!("Skipping video trim / resample...");
info!("(This might break if we need to copy the video path. Salt's code implicitly expects videos to be in certain places, but doesn't allow passing of config, and that's horrible.)");
std::fs::copy(&args.videos.original_video_path, &args.videos.trimmed_resampled_video_path)
.map_err(|err| {
error!("Error copying video (1): {:?}", err);
ProcessSingleJobError::IoError(err)
})?;
std::fs::copy(&args.videos.original_video_path, &args.videos.comfy_output_video_path)
.map_err(|err| {
error!("Error copying video (2): {:?}", err);
ProcessSingleJobError::IoError(err)
})?;
} else {
info!("Calling video trim / resample...");
info!("Script: {:?}", &args.comfy_deps.inference_command.processing_script);
// shell out to python script
let output = Command::new("python3")
.stdout(Stdio::inherit()) // NB: This should emit to the rust job's stdout
.stderr(Stdio::inherit()) // NB: This should emit to the rust job's stderr
.arg(path_to_string(&args.comfy_deps.inference_command.processing_script))
.arg(path_to_string(&args.videos.original_video_path))
.arg(format!("{:?}", trim_start_millis))
.arg(format!("{:?}", trim_end_millis))
.arg(format!("{:?}", target_fps))
.output()
.map_err(|e| {
error!("Error running inference: {:?}", e);
ProcessSingleJobError::Other(e.into())
})?;
// check if the command was successful
if !output.status.success() {
// print stdout and stderr
error!("Video processing failed: {:?}", output.status);
error!("stdout: {}", String::from_utf8_lossy(&output.stdout));
error!("stderr: {}", String::from_utf8_lossy(&output.stderr));
return Err(ProcessSingleJobError::Other(anyhow!("Command failed: {:?}", output.status)));
}
info!("Finished video trim / resample.");
// NB: The process video script implicitly saves the above video as "input.mp4"
// Comfy sometimes overwrites this, so we need to make a copy.
std::fs::copy(&args.videos.comfy_input_video_path, &args.videos.trimmed_resampled_video_path)
.map_err(|err| {
error!("Error copying trimmed video: {:?}", err);
ProcessSingleJobError::IoError(err)
})?;
}
args.videos.debug_print_paths_after_trim();
if let Ok(Some(dimensions)) = ffprobe_get_dimensions(&args.videos.trimmed_resampled_video_path) {
info!("Trimmed / resampled video dimensions: {}x{}", dimensions.width, dimensions.height);
}
Ok(())
}
@@ -0,0 +1,239 @@
use std::path::{Path, PathBuf};
use std::process::{Command, Stdio};
use anyhow::anyhow;
use log::{error, info};
use sqlx::MySqlPool;
use cloud_storage::remote_file_manager::remote_cloud_file_manager::RemoteCloudFileClient;
use filesys::path_to_string::path_to_string;
use mysql_queries::payloads::generic_inference_args::workflow_payload::WorkflowArgs;
use tokens::tokens::media_files::MediaFileToken;
use videos::ffprobe_get_dimensions::ffprobe_get_dimensions;
use crate::job::job_loop::process_single_job_error::ProcessSingleJobError;
use crate::job::job_types::workflow::comfy_ui::comfy_ui_dependencies::ComfyDependencies;
use crate::job::job_types::workflow::comfy_ui::video_style_transfer::util::comfy_dirs::ComfyDirs;
use crate::job::job_types::workflow::comfy_ui::video_style_transfer::util::video_pathing::{SecondaryInputVideoAndPaths, VideoDownloads};
use crate::job::job_types::workflow::comfy_ui::video_style_transfer::util::video_pathing_deprecated::VideoPaths;
pub struct ProcessTrimAndResampleVideoArgs<'a> {
pub comfy_args: &'a WorkflowArgs,
pub comfy_deps: &'a ComfyDependencies,
pub comfy_dirs: &'a ComfyDirs,
pub videos: &'a mut VideoDownloads,
}
pub fn preprocess_trim_and_resample_videos(
args: ProcessTrimAndResampleVideoArgs<'_>
) -> Result<(), ProcessSingleJobError> {
let target_fps = args.comfy_args.target_fps.unwrap_or(24);
let trim_start_millis = args.comfy_args.trim_start_milliseconds
.or_else(|| args.comfy_args.trim_start_seconds.map(|s| s as u64 * 1_000))
.unwrap_or(0);
let trim_end_millis = args.comfy_args.trim_end_milliseconds
.or_else(|| args.comfy_args.trim_end_seconds.map(|s| s as u64 * 1_000))
.unwrap_or(3_000);
info!("trim start millis: {trim_start_millis}");
info!("trim end millis: {trim_end_millis}");
info!("target FPS: {target_fps}");
let resample_details = ResampleDetails {
target_fps,
trim_start_millis,
trim_end_millis,
};
let skip_resampling_video = args.comfy_args.skip_process_video.unwrap_or(false);
preprocess_trim_and_resample_primary_video(
&args.comfy_deps,
&args.comfy_dirs,
&resample_details,
skip_resampling_video,
args.videos)?;
if !skip_resampling_video {
preprocess_trim_and_resample_secondary_videos(
&args.comfy_deps,
&resample_details,
args.videos)?;
}
Ok(())
}
struct ResampleDetails {
target_fps: u32,
trim_start_millis: u64,
trim_end_millis: u64,
}
fn preprocess_trim_and_resample_primary_video(
comfy_deps: &ComfyDependencies,
comfy_dirs: &ComfyDirs,
resample_details: &ResampleDetails,
skip_process_video: bool,
videos: &mut VideoDownloads,
) -> Result<(), ProcessSingleJobError> {
let resampled_path = comfy_dirs.comfy_input_dir.join("trimmed.mp4");
if skip_process_video {
info!("Skipping video trim / resample...");
info!("(This might break if we need to copy the video path. Salt's code implicitly expects videos to be in certain places, but doesn't allow passing of config, and that's horrible.)");
std::fs::copy(&videos.input_video.original_download_path, &resampled_path)
.map_err(|err| {
error!("Error copying video (1): {:?}", err);
ProcessSingleJobError::IoError(err)
})?;
std::fs::copy(&videos.input_video.original_download_path, &videos.input_video.comfy_output_video_path)
.map_err(|err| {
error!("Error copying video (2): {:?}", err);
ProcessSingleJobError::IoError(err)
})?;
} else {
info!("Calling video trim / resample...");
info!("Script: {:?}", &comfy_deps.inference_command.processing_script);
// NB(bt,2024-07-09): Despite what the comments on this field say, the script `format_video.py` writes
// to a file named 'input.mp4', not 'trimmed.mp4'. This pathing really needs to be cleaned up.
let comfy_input_video_path = comfy_dirs.comfy_input_dir.join("input.mp4");
// shell out to python script
let output = Command::new("python3")
.stdout(Stdio::inherit()) // NB: This should emit to the rust job's stdout
.stderr(Stdio::inherit()) // NB: This should emit to the rust job's stderr
.arg(path_to_string(&comfy_deps.inference_command.processing_script))
.arg("DO_NOT_USE_THIS_ARG_ANYMORE")
.arg(format!("{:?}", resample_details.trim_start_millis))
.arg(format!("{:?}", resample_details.trim_end_millis))
.arg(format!("{:?}", resample_details.target_fps))
.arg("--input")
.arg(path_to_string(&videos.input_video.original_download_path))
.arg("--output")
// NB(bt,2024-07-09): Despite what the comments on this field say, the script `format_video.py` writes
// to a file named 'input.mp4', not 'trimmed.mp4'. This pathing really needs to be cleaned up.
.arg(path_to_string(&comfy_input_video_path))
.output()
.map_err(|e| {
error!("Error running inference: {:?}", e);
ProcessSingleJobError::Other(e.into())
})?;
// check if the command was successful
if !output.status.success() {
// print stdout and stderr
error!("Video processing failed: {:?}", output.status);
error!("stdout: {}", String::from_utf8_lossy(&output.stdout));
error!("stderr: {}", String::from_utf8_lossy(&output.stderr));
return Err(ProcessSingleJobError::Other(anyhow!("Command failed: {:?}", output.status)));
}
info!("Finished video trim / resample.");
// NB: The process video script implicitly saves the above video as "input.mp4"
// Comfy sometimes overwrites this, so we need to make a copy.
std::fs::copy(&comfy_input_video_path, &resampled_path)
.map_err(|err| {
error!("Error copying trimmed video: {:?}", err);
ProcessSingleJobError::IoError(err)
})?;
}
//primary_video_paths.debug_print_paths_after_trim();
if let Ok(Some(dimensions)) = ffprobe_get_dimensions(&resampled_path) {
info!("Trimmed / resampled video dimensions: {}x{}", dimensions.width, dimensions.height);
}
// NB(bt,2024-07-10): Even if we don't resample, the python side still expects certain pathing for now
videos.input_video.maybe_processed_path = Some(resampled_path);
Ok(())
}
fn preprocess_trim_and_resample_secondary_videos(
comfy_deps: &ComfyDependencies,
resample_details: &ResampleDetails,
secondary_videos: &mut VideoDownloads,
) -> Result<(), ProcessSingleJobError> {
if let Some(depth) = secondary_videos.maybe_depth.as_mut() {
let resampled_path = preprocess_trim_and_resample_secondary_video(comfy_deps, resample_details, depth)?;
depth.maybe_processed_path = Some(resampled_path);
}
if let Some(normal) = secondary_videos.maybe_normal.as_mut() {
let resampled_path = preprocess_trim_and_resample_secondary_video(comfy_deps, resample_details, normal)?;
normal.maybe_processed_path = Some(resampled_path);
}
if let Some(outline) = secondary_videos.maybe_outline.as_mut() {
let resampled_path = preprocess_trim_and_resample_secondary_video(comfy_deps, resample_details, outline)?;
outline.maybe_processed_path = Some(resampled_path);
}
Ok(())
}
fn preprocess_trim_and_resample_secondary_video(
comfy_deps: &ComfyDependencies,
resample_details: &ResampleDetails,
video: &SecondaryInputVideoAndPaths,
) -> Result<PathBuf, ProcessSingleJobError> {
info!("Calling video trim / resample...");
info!("Script: {:?}", &comfy_deps.inference_command.processing_script);
let output_path = {
let filename = video.original_download_path
.file_stem()
.map(|s| s.to_str())
.flatten()
.ok_or_else(|| {
ProcessSingleJobError::Other(anyhow!("Error getting filename: {:?}", video.original_download_path))
})?;
let filename = format!("{}_resampled.mp4", filename);
video.original_download_path.with_file_name(filename)
};
// shell out to python script
let output = Command::new("python3")
.stdout(Stdio::inherit()) // NB: This should emit to the rust job's stdout
.stderr(Stdio::inherit()) // NB: This should emit to the rust job's stderr
.arg(path_to_string(&comfy_deps.inference_command.processing_script))
.arg(path_to_string("DO_NOT_USE_THIS_ARG_ANYMORE"))
.arg(format!("{:?}", resample_details.trim_start_millis))
.arg(format!("{:?}", resample_details.trim_end_millis))
.arg(format!("{:?}", resample_details.target_fps))
.arg("--input")
.arg(format!("{:?}", &video.original_download_path))
.arg("--output")
.arg(format!("{:?}", &output_path))
.output()
.map_err(|e| {
error!("Error running inference: {:?}", e);
ProcessSingleJobError::Other(e.into())
})?;
// check if the command was successful
if !output.status.success() {
// print stdout and stderr
error!("Video processing failed: {:?}", output.status);
error!("stdout: {}", String::from_utf8_lossy(&output.stdout));
error!("stderr: {}", String::from_utf8_lossy(&output.stderr));
return Err(ProcessSingleJobError::Other(anyhow!("Command failed: {:?}", output.status)));
}
info!("Finished video trim / resample.");
Ok(output_path)
}
@@ -33,8 +33,8 @@ use crate::job::job_loop::process_single_job_error::ProcessSingleJobError;
use crate::job::job_types::workflow::comfy_ui::comfy_process_job_args::ComfyProcessJobArgs;
use crate::job::job_types::workflow::comfy_ui::comfy_ui_dependencies::ComfyDependencies;
use crate::job::job_types::workflow::comfy_ui::video_style_transfer::steps::check_and_validate_job::JobArgs;
use crate::job::job_types::workflow::comfy_ui::video_style_transfer::steps::download_input_video::VideoDownloadDetails;
use crate::job::job_types::workflow::comfy_ui::video_style_transfer::video_paths::VideoPaths;
use crate::job::job_types::workflow::comfy_ui::video_style_transfer::util::video_pathing::VideoDownloads;
use crate::job::job_types::workflow::comfy_ui::video_style_transfer::util::video_pathing_deprecated::VideoPaths;
use crate::job_dependencies::JobDependencies;
fn get_file_extension(mimetype: &str) -> anyhow::Result<&'static str> {
@@ -55,8 +55,7 @@ pub struct SaveResultsArgs<'a> {
pub comfy_deps: &'a ComfyDependencies,
pub job_progress_reporter: &'a mut Box<dyn JobProgressReporter>,
pub download_video: VideoDownloadDetails,
pub videos: &'a VideoPaths,
pub videos: &'a VideoDownloads,
pub inference_duration: Duration,
pub comfy_args: &'a WorkflowArgs,
@@ -71,7 +70,7 @@ pub async fn validate_and_save_results(args: SaveResultsArgs<'_>) -> Result<Medi
info!("Interrogating result file size ...");
let final_video = args.videos.get_final_video_to_upload();
let final_video = args.videos.input_video.get_final_video_to_upload();
let file_size_bytes = file_size(final_video)
.map_err(|err| ProcessSingleJobError::Other(err))?;
@@ -175,7 +174,7 @@ pub async fn validate_and_save_results(args: SaveResultsArgs<'_>) -> Result<Medi
let result = args.deps.buckets.public_bucket_client.upload_filename_with_content_type(
&result_bucket_object_pathbuf.with_extension("no_watermark.mp4"),
&args.videos.get_non_watermarked_video_to_upload(),
&args.videos.input_video.get_non_watermarked_video_to_upload(),
&mimetype) // TODO: We should check the mimetype to make sure bad payloads can't get uploaded
.await;
@@ -200,11 +199,9 @@ pub async fn validate_and_save_results(args: SaveResultsArgs<'_>) -> Result<Medi
// This also lets us hide the engine renders from users.
// This shouldn't ever become a deeply nested tree of children, but rather a single root
// with potentially many direct children.
let style_transfer_source_media_file_token = args.download_video
.input_video_media_file
.maybe_style_transfer_source_media_file_token
.as_ref()
.unwrap_or_else(|| &args.download_video.input_video_media_file.token);
let style_transfer_source_media_file_token = args.videos
.input_video.record.maybe_style_transfer_source_media_file_token.as_ref()
.unwrap_or_else(|| &args.videos.input_video.record.token);
let prompt_token = PromptToken::generate();
@@ -212,9 +209,10 @@ pub async fn validate_and_save_results(args: SaveResultsArgs<'_>) -> Result<Medi
pool: &args.deps.db.mysql_pool,
job: &args.job,
maybe_mime_type: Some(&mimetype),
maybe_title: args.download_video.input_video_media_file.maybe_title.as_deref(),
maybe_title: args.videos.input_video.record.maybe_title.as_deref(),
maybe_style_transfer_source_media_file_token: Some(&style_transfer_source_media_file_token),
maybe_scene_source_media_file_token: args.download_video.input_video_media_file.maybe_scene_source_media_file_token.as_ref(),
maybe_scene_source_media_file_token: args.videos
.input_video.record.maybe_scene_source_media_file_token.as_ref(),
file_size_bytes,
sha256_checksum: &file_checksum,
maybe_prompt_token: Some(&prompt_token),
@@ -238,6 +236,7 @@ pub async fn validate_and_save_results(args: SaveResultsArgs<'_>) -> Result<Medi
|| args.comfy_args.lipsync_enabled.is_some()
|| args.comfy_args.negative_prompt.is_some()
|| args.comfy_args.positive_prompt.is_some()
|| args.comfy_args.travel_prompt.is_some()
|| args.comfy_args.strength.is_some()
|| args.comfy_args.style_name.is_some()
|| args.comfy_args.use_cinematic.is_some()
@@ -289,6 +288,14 @@ pub async fn validate_and_save_results(args: SaveResultsArgs<'_>) -> Result<Medi
other_args_builder.set_global_ipa_token(args.comfy_args.global_ip_adapter_token.clone());
}
if args.comfy_args.travel_prompt.is_some() {
other_args_builder.set_travel_prompt(args.comfy_args.travel_prompt.clone());
}
if args.comfy_args.frame_skip.is_some() {
other_args_builder.set_frame_skip(args.comfy_args.frame_skip.clone());
}
let maybe_other_args = other_args_builder.build();
info!("maybe other prompt args: {:?}", maybe_other_args);
@@ -0,0 +1,24 @@
use std::path::PathBuf;
use crate::job::job_types::workflow::comfy_ui::comfy_ui_dependencies::ComfyDependencies;
/// This is a temporary measure until we start using tempdirs for intermediate inputs and outputs.
pub struct ComfyDirs {
pub comfy_input_dir: PathBuf,
pub comfy_output_dir: PathBuf,
}
impl ComfyDirs {
pub fn new(comfy_deps: &ComfyDependencies) -> Self {
// TODO(bt,2024-04-21): The pathing for this job is complicated. ComfyUI
// was set up with some of these expectations, which is bad. Worse, the
// jobs enqueue the expected output path -- no idea why that was done.
// This should all be fixed.
let root_comfy_path = comfy_deps.inference_command.mounts_directory.as_path();
Self {
comfy_input_dir: root_comfy_path.join("input"),
comfy_output_dir: root_comfy_path.join("output"),
}
}
}
@@ -99,8 +99,14 @@ pub struct InferenceArgs<'s> {
pub maybe_strength: Option<f32>,
pub frame_skip: Option<u8>,
pub global_ipa_image_filename: Option<String>,
pub global_ipa_strength: Option<f32>,
pub depth_video_path: Option<&'s Path>,
pub normal_video_path: Option<&'s Path>,
pub outline_video_path: Option<&'s Path>,
}
#[derive(Debug)]
@@ -121,6 +127,11 @@ pub enum InferenceDetails<'s> {
/// with the correct workflow args.
maybe_negative_prompt_filename: Option<&'s Path>,
/// Travel prompt file.
/// Optional: If set, Python will be in charge of overwriting the prompt JSON file
/// with the correct workflow args.
maybe_travel_prompt_filename: Option<&'s Path>,
/// Style name
/// Optional: If set, Python will be in charge of overwriting the prompt JSON file
/// with the correct workflow args.
@@ -254,6 +265,7 @@ impl ComfyInferenceCommand {
InferenceDetails::NewPythonArgs {
maybe_positive_prompt_filename,
maybe_negative_prompt_filename,
maybe_travel_prompt_filename,
maybe_style
} => {
if let Some(positive_prompt_filename) = maybe_positive_prompt_filename {
@@ -268,6 +280,12 @@ impl ComfyInferenceCommand {
command.push_str(" ");
}
if let Some(travel_prompt_filename) = maybe_travel_prompt_filename {
command.push_str(" --travel_prompt_filename ");
command.push_str(&path_to_string(travel_prompt_filename));
command.push_str(" ");
}
if let Some(style) = maybe_style {
command.push_str(" --style ");
command.push_str(style.to_str());
@@ -302,6 +320,12 @@ impl ComfyInferenceCommand {
command.push_str(" ");
}
if let Some(frame_skip) = args.frame_skip {
command.push_str(" --frame_skip ");
command.push_str(&frame_skip.to_string());
command.push_str(" ");
}
if let Some(global_ipa_image_filename) = &args.global_ipa_image_filename {
// NB: This is very dangerous. We're not handling escaping well.
command.push_str(" --global_ipa_image_filename ");
@@ -315,6 +339,23 @@ impl ComfyInferenceCommand {
command.push_str(" ");
}
if let Some(depth_video_path) = args.depth_video_path {
command.push_str(" --depth_video_filename ");
command.push_str(&path_to_string(depth_video_path));
command.push_str(" ");
}
if let Some(normal_video_path) = args.normal_video_path {
command.push_str(" --normals_video_filename ");
command.push_str(&path_to_string(normal_video_path));
command.push_str(" ");
}
if let Some(outline_video_path) = args.outline_video_path {
command.push_str(" --outline_video_filename ");
command.push_str(&path_to_string(outline_video_path));
command.push_str(" ");
}
if let Some(docker_options) = self.maybe_docker_options.as_ref() {
command = docker_options.to_command_string(&command);
@@ -0,0 +1,5 @@
pub mod comfy_dirs;
pub mod comfy_ui_inference_command;
pub mod video_pathing;
pub mod video_pathing_deprecated;
pub mod write_workflow_prompt;
@@ -0,0 +1,177 @@
use std::path::{Path, PathBuf};
use anyhow::anyhow;
use mysql_queries::queries::media_files::get::batch_get_media_files_by_tokens::MediaFilesByTokensRecord;
use mysql_queries::queries::media_files::get::get_media_file::MediaFile;
use tokens::tokens::media_files::MediaFileToken;
use crate::job::job_loop::process_single_job_error::ProcessSingleJobError;
use crate::job::job_types::workflow::comfy_ui::video_style_transfer::util::comfy_dirs::ComfyDirs;
/// Keep track of where we download videos to and where they end up after processing.
pub struct VideoDownloads {
/// The main video
pub input_video: PrimaryInputVideoAndPaths,
// Secondary videos that provide enrichment signals.
pub maybe_depth: Option<SecondaryInputVideoAndPaths>,
pub maybe_normal: Option<SecondaryInputVideoAndPaths>,
pub maybe_outline: Option<SecondaryInputVideoAndPaths>,
}
/// This is the primary input video for style transfer.
pub struct PrimaryInputVideoAndPaths {
/// Media file record.
pub record: MediaFile,
/// Filesystem path of the downloaded original video
/// This is the input.
pub original_download_path: PathBuf,
// TODO: Rename back to `trimmed_resampled...`
/// Filesystem path of the trimmed and resampled video
/// This is the first output we generate.
/// We'll use this downstream once it's available.
pub maybe_processed_path: Option<PathBuf>, //pub trimmed_resampled_video_path: PathBuf,
/// Filesystem path of the trimmed audio
pub trimmed_audio_path: PathBuf,
/// This is the input into Comfy.
/// This is typically the `trimmed_resampled_video_path`, but since Comfy
/// can overwrite the source, we'll keep a separate copy of that file for
/// later downstream sound restoration (since comfy wipes sound).
pub comfy_input_video_path: PathBuf,
/// Filesystem path of the style transfer output
/// This is the main purpose of the job, and the second output we generate.
pub comfy_output_video_path: PathBuf,
/// Filesystem path of the audio-restored output
/// This is the third output we generate.
/// We'll want to upload this as a result, if available.
pub audio_restored_video_path: Option<PathBuf>,
/// Watermarked final result
/// This is the fourth output we generate.
/// We'll want to upload this as a result, if available.
pub watermarked_video_path: Option<PathBuf>,
// Nasty hack to reuse this elsewhere.
pub comfy_input_dir: PathBuf,
}
/// This is for the secondary depth, normal, and outline videos.
/// We attach metadata as these videos progress through the system (if they're present).
pub struct SecondaryInputVideoAndPaths {
/// Media file record.
pub record: MediaFilesByTokensRecord,
/// Filesystem path of the downloaded original video
pub original_download_path: PathBuf,
/// Filesystem path of the trimmed and resampled video
pub maybe_processed_path: Option<PathBuf>,
}
// Adapter enum
// pub enum VideoMediaFileRecord {
// // Records returned by single lookup
// Single(MediaFile),
// // Records returned by batch query
// Bulk(MediaFilesByTokensRecord),
// }
pub trait CommonVideoPathing {
fn original_video_path(&self) -> &Path;
}
impl CommonVideoPathing for PrimaryInputVideoAndPaths {
fn original_video_path(&self) -> &Path {
self.original_download_path.as_path()
}
}
impl CommonVideoPathing for SecondaryInputVideoAndPaths {
fn original_video_path(&self) -> &Path {
self.original_download_path.as_path()
}
}
impl PrimaryInputVideoAndPaths {
pub fn new(media_file_record: MediaFile, comfy_dirs: &ComfyDirs, job_output_path: &str) -> Self {
let original_video_path = comfy_dirs.comfy_input_dir.join("video.mp4");
let trimmed_resampled_video_path = comfy_dirs.comfy_input_dir.join("trimmed.mp4");
let trimmed_audio_path = comfy_dirs.comfy_input_dir.join("trimmed.wav");
let comfy_input_video_path = comfy_dirs.comfy_input_dir.join("input.mp4");
let comfy_output_video_path = comfy_dirs.comfy_output_dir.join(job_output_path); // TODO: This sucks.
Self {
record: media_file_record,
original_download_path: original_video_path,
maybe_processed_path: Some(trimmed_resampled_video_path),
trimmed_audio_path,
comfy_input_video_path,
comfy_output_video_path,
audio_restored_video_path: None,
watermarked_video_path: None,
comfy_input_dir: comfy_dirs.comfy_input_dir.clone(),
}
}
pub fn video_to_watermark(&self) -> &PathBuf {
// Try to use the audio-restored video if it's available
self.audio_restored_video_path.as_ref()
.unwrap_or(&self.comfy_output_video_path)
}
pub fn get_final_video_to_upload(&self) -> &PathBuf {
// This is the video to upload as the result and save in the media_files table.
self.watermarked_video_path.as_ref()
.or(self.audio_restored_video_path.as_ref())
.unwrap_or(&self.comfy_output_video_path)
}
pub fn get_non_watermarked_video_to_upload(&self) -> &PathBuf {
// We'll upload this for internal use and for premium users.
// Same as "video_to_watermark()"
self.audio_restored_video_path.as_ref()
.unwrap_or(&self.comfy_output_video_path)
}
}
// impl VideoMediaFileRecord {
// pub fn token(&self) -> &MediaFileToken {
// match self {
// VideoMediaFileRecord::Single(m) => &m.token,
// VideoMediaFileRecord::Bulk(m) => &m.token,
// }
// }
//
// pub fn maybe_title(&self) -> Option<&str> {
// match self {
// VideoMediaFileRecord::Single(m) => m.maybe_title.as_deref(),
// VideoMediaFileRecord::Bulk(m) => m.maybe_title.as_deref(),
// }
// }
//
// pub fn maybe_style_transfer_source_media_file_token(&self) -> Result<Option<&MediaFileToken>, ProcessSingleJobError> {
// // TODO(bt,2024-07-09): Future proofing this to deliberately explode in case I query the
// // primary input media files with the bulk query.
// match self {
// VideoMediaFileRecord::Single(m) => Ok(m.maybe_style_transfer_source_media_file_token.as_ref()),
// VideoMediaFileRecord::Bulk(_m) => Err(ProcessSingleJobError::Other(anyhow!("bad refactor?: failed to query foreign key"))),
// }
// }
//
// pub fn maybe_scene_source_media_file_token(&self) -> Result<Option<&MediaFileToken>, ProcessSingleJobError> {
// // TODO(bt,2024-07-09): Future proofing this to deliberately explode in case I query the
// // primary input media files with the bulk query.
// match self {
// VideoMediaFileRecord::Single(m) => Ok(m.maybe_scene_source_media_file_token.as_ref()),
// VideoMediaFileRecord::Bulk(_m) => Err(ProcessSingleJobError::Other(anyhow!("bad refactor?: failed to query foreign key"))),
// }
// }
// }
@@ -4,10 +4,11 @@ use log::info;
use filesys::file_exists::file_exists;
#[deprecated]
pub struct VideoPaths {
/// Filesystem path of the downloaded original video
/// This is the input.
pub original_video_path: PathBuf,
original_video_path: PathBuf,
/// Filesystem path of the trimmed and resampled video
/// This is the first output we generate.
@@ -36,6 +37,9 @@ pub struct VideoPaths {
/// This is the fourth output we generate.
/// We'll want to upload this as a result, if available.
pub watermarked_video_path: Option<PathBuf>,
// Nasty hack to reuse this elsewhere.
pub comfy_input_dir: PathBuf,
}
@@ -62,6 +66,7 @@ impl VideoPaths {
comfy_output_video_path,
audio_restored_video_path: None,
watermarked_video_path: None,
comfy_input_dir: input_dir,
}
}
-77
View File
@@ -1,77 +0,0 @@
[package]
publish = false
name = "users_component"
edition = "2018"
version = "0.0.1"
authors = [
"Brandon Thomas <bt@brand.io>",
"Brandon Thomas <echelon@gmail.com>",
]
[lib]
name = "users_component"
path = "src/lib.rs"
[package.metadata.cargo-udeps.ignore]
normal = ["my-workspace-hack"]
[dependencies]
# Internal
actix_helpers = { path = "../../../lib/actix_helpers" }
cookies = { path = "../../../lib/cookies" }
crockford = { path = "../../../lib/data/crockford" }
email_sender = { path = "../../../lib/email_sender" }
enums = { path = "../../../_types/enums" }
errors = { path = "../../../lib/errors" }
http_server_common = { path = "../../../lib/deprecated/http_server_common" }
idempotency = { path = "../../../lib/idempotency" }
mysql_queries = { path = "../../../lib/mysql_queries" }
password = { path = "../../../lib/password" }
redis_caching = { path = "../../../lib/caching/redis_caching" }
redis_common = { path = "../../../lib/redis_common" }
reusable_types = { path = "../../../_types/reusable_types" }
server_environment = { path = "../../../lib/server_environment" }
tokens = { path = "../../../_types/tokens" }
user_input_common = { path = "../../../lib/user_input_common" }
user_traits_component = { path = "../user_traits_component" }
# Workspace hack (faster compile times)
my-workspace-hack = { version = "0.1", path = "../../../my-workspace-hack" }
# External
anyhow = "1.0.86"
base64 = "0.22.1"
chrono = { version = "0.4.22", features = ["serde"] }
dotenv = "0.15.0"
lazy_static = "1.4.0" # Regex caching (TODO: Use OnceCell instead)
log = "0.4.14"
once_cell = "1.18.0"
rand = "0.8.5"
regex = "1.5.4"
time = "0.3.30"
# Web
actix-web = "4.7.0"
jwt = "0.13.0"
# Crypto
hmac = "0.10.1" # cookies
md-5 = "0.10.6" # gravatar
sha2 = "0.9.9" # file hashes, inference text hashes, etc.
# Serialization
serde = "1.0.125"
serde_derive = "1.0.125"
serde_json = "1.0.108"
toml = "0.8.14"
sqlx = { version = "0.7.2", features = [ "mysql", "runtime-tokio", "tls-rustls", "chrono" ] }
# TODO: Shouldn't really be a dependency, but need to add caching for the DDoS attack.
# Redis Database
r2d2_redis = "0.14.0"
byteorder = "1.5.0"
strum_macros = "0.26.4"
utoipa = { version = "4", features = ["chrono","actix_extras"] }
@@ -1,2 +0,0 @@
pub mod user_avatars;
pub mod user_details_lite;
@@ -1,83 +0,0 @@
//! These routes are recommended, but do not have to be used by consumers of the user system.
use actix_web::{App, HttpResponse, web};
use actix_web::body::MessageBody;
use actix_web::dev::{ServiceFactory, ServiceRequest, ServiceResponse};
use actix_web::error::Error;
use crate::endpoints::create_account_handler::create_account_handler;
use crate::endpoints::login_handler::login_handler;
use crate::endpoints::logout_handler::logout_handler;
use crate::endpoints::password_reset_redeem_handler::password_reset_redeem_handler;
use crate::endpoints::password_reset_request_handler::password_reset_request_handler;
use crate::endpoints::session_info_handler::session_info_handler;
// NB: This does not include the user edit endpoints since FakeYou's API mounts them alongside other things.
// A 'v2' API would mount under a v2 prefix and the entity type.
pub fn add_suggested_api_v1_account_creation_and_session_routes<T, B> (app: App<T>) -> App<T>
where
B: MessageBody,
T: ServiceFactory<
ServiceRequest,
Config = (),
Response = ServiceResponse<B>,
Error = Error,
InitError = (),
>,
{
app
.service(
// TODO(bt,2022-11-16): non-/v1/ endpoints are deprecated and subject for future removal
web::resource("/create_account")
.route(web::post().to(create_account_handler))
.route(web::head().to(|| HttpResponse::Ok()))
)
.service(
web::resource("/v1/create_account")
.route(web::post().to(create_account_handler))
.route(web::head().to(|| HttpResponse::Ok()))
)
.service(
// TODO(bt,2022-11-16): non-/v1/ endpoints are deprecated and subject for future removal
web::resource("/login")
.route(web::post().to(login_handler))
.route(web::head().to(|| HttpResponse::Ok()))
)
.service(
web::resource("/v1/login")
.route(web::post().to(login_handler))
.route(web::head().to(|| HttpResponse::Ok()))
)
.service(
// TODO(bt,2022-11-16): non-/v1/ endpoints are deprecated and subject for future removal
web::resource("/logout")
.route(web::post().to(logout_handler))
.route(web::head().to(|| HttpResponse::Ok()))
)
.service(
web::resource("/v1/logout")
.route(web::post().to(logout_handler))
.route(web::head().to(|| HttpResponse::Ok()))
)
.service(
// TODO(bt,2022-11-16): non-/v1/ endpoints are deprecated and subject for future removal
web::resource("/session")
.route(web::get().to(session_info_handler))
.route(web::head().to(|| HttpResponse::Ok()))
)
.service(
web::resource("/v1/session")
.route(web::get().to(session_info_handler))
.route(web::head().to(|| HttpResponse::Ok()))
)
.service(
web::resource("/v1/password_reset/request")
.route(web::post().to(password_reset_request_handler))
.route(web::head().to(|| HttpResponse::Ok()))
)
.service(
web::resource("/v1/password_reset/redeem")
.route(web::post().to(password_reset_redeem_handler))
.route(web::head().to(|| HttpResponse::Ok()))
)
}
-13
View File
@@ -1,13 +0,0 @@
#[macro_use] extern crate lazy_static;
#[macro_use] extern crate serde_derive;
pub const RESERVED_USERNAMES : &str = include_str!("../../../../../includes/binary_includes/reserved_usernames.txt");
pub const RESERVED_SUBSTRINGS : &str = include_str!("../../../../../includes/binary_includes/reserved_usernames_including.txt");
pub mod common_responses;
pub mod cookies;
pub mod default_routes;
pub mod endpoints;
pub mod session;
pub mod utils;
pub mod validations;
@@ -1 +0,0 @@
pub mod email_to_gravatar;
@@ -54,6 +54,7 @@ memory_caching = { path = "../../../lib/caching/memory_caching" }
migration = { path = "../../../lib/migration" }
mimetypes = { path = "../../../lib/files/mimetypes" }
mysql_queries = { path = "../../../lib/mysql_queries" }
password = { path = "../../../lib/password" }
primitives = { path = "../../../lib/data/primitives" }
prompt_engineering = { path = "../../../lib/prompt_engineering" }
redis_caching = { path = "../../../lib/caching/redis_caching" }
@@ -69,7 +70,6 @@ videos = { path = "../../../lib/files/videos" }
# Internal Components
billing_component = { path = "../../plugins/billing" }
user_traits_component = { path = "../../plugins/user_traits_component" }
users_component = { path = "../../plugins/users" }
# Workspace hack (faster compile times)
my-workspace-hack = { version = "0.1", path = "../../../my-workspace-hack" }
@@ -86,6 +86,7 @@ log = "0.4.14"
once_cell = "1.18.0"
rand = "0.8.5"
regex = "1.5.4"
strum_macros = "0.26.4"
#symphonia = { version = "0.5.2", features = ["aac", "alac", "flac", "isomp4", "mkv", "mp3", "ogg", "opus", "vorbis", "wav"] }
# Opus support still hasn't landed in Symphonia, so we have to use an older version (~0.5) off-master.
@@ -17,50 +17,25 @@
#[macro_use] extern crate magic_crypt;
#[macro_use] extern crate serde_derive;
use utoipa_swagger_ui::SwaggerUi;
use actix_web::{App, HttpServer, middleware, web};
use actix_web::middleware::{DefaultHeaders, Logger};
use r2d2_redis::r2d2;
use r2d2_redis::redis::Commands;
use r2d2_redis::RedisConnectionManager;
use sqlx::mysql::MySqlPoolOptions;
use sqlx::MySqlPool;
use tokio::runtime::Runtime;
use errors::AnyhowResult;
use crate::configs::static_api_tokens::{StaticApiTokenConfig, StaticApiTokens, StaticApiTokenSet};
use crate::server_state::{DurableInMemoryCaches, EnvConfig, EphemeralInMemoryCaches, InMemoryCaches, RedisRateLimiters, ServerInfo, ServerState, StaticFeatureFlags, StripeSettings, TrollBans};
pub mod billing;
pub mod configs;
pub mod cookies;
pub mod http_server;
pub mod memory_cache;
pub mod model;
pub mod routes;
pub mod server_state;
pub mod subscriptions;
pub mod threads;
pub mod util;
pub mod validations;
use std::{
error::Error,
future::{self, Ready},
net::Ipv4Addr,
};
use std::error::Error;
use std::fs::File;
use std::io::Write;
pub mod api_doc;
use api_doc::ApiDoc;
use futures::future::LocalBoxFuture;
use r2d2_redis::redis::Commands;
use utoipa::OpenApi;
use docs::api_doc::ApiDoc;
use errors::AnyhowResult;
pub mod billing;
pub mod configs;
pub mod http_server;
pub mod state;
pub mod threads;
pub mod util;
pub mod docs;
#[actix_web::main]
async fn main() -> AnyhowResult<()> {
let api_json = ApiDoc::openapi().to_pretty_json()?;
@@ -25,26 +25,20 @@ use sqlx::MySqlPool;
use utoipa::OpenApi;
use utoipa_swagger_ui::SwaggerUi;
use api_doc::ApiDoc;
use docs::api_doc::ApiDoc;
use errors::AnyhowResult;
use crate::configs::static_api_tokens::StaticApiTokenSet;
use crate::server_state::ServerState;
use crate::state::server_state::ServerState;
pub mod billing;
pub mod configs;
pub mod cookies;
pub mod http_server;
pub mod memory_cache;
pub mod model;
pub mod routes;
pub mod server_state;
pub mod subscriptions;
pub mod state;
pub mod threads;
pub mod util;
pub mod validations;
pub mod api_doc;
pub mod docs;
#[actix_web::main]
async fn main() -> Result<(), impl Error> {
@@ -3,7 +3,7 @@ use actix_web::HttpRequest;
use redis_caching::redis_ttl_cache::RedisTtlCache;
use redis_common::redis_cache_keys::RedisCacheKeys;
use user_traits_component::traits::internal_session_cache_purge::InternalSessionCachePurge;
use users_component::session::session_checker::SessionChecker;
use crate::http_server::session::session_checker::SessionChecker;
pub struct InternalSessionCachePurgeImpl {
session_checker: SessionChecker,
@@ -2,11 +2,10 @@ use actix_web::HttpRequest;
use async_trait::async_trait;
use chrono::Utc;
use log::warn;
use sqlx::MySqlPool;
use billing_component::stripe::traits::internal_user_lookup::{InternalUserLookup, InternalUserLookupError, SubscriptionKey, UserMetadata};
use users_component::session::session_checker::SessionChecker;
use crate::MySqlPool;
use crate::http_server::session::session_checker::SessionChecker;
/// A simple Actix injectable action
#[derive(Clone)]
@@ -8,7 +8,7 @@ use limitation::Limiter;
use crate::configs::app_startup::username_set::UsernameSet;
use crate::http_server::web_utils::redis_rate_limiter::RedisRateLimiter;
use crate::server_state::RedisRateLimiters;
use crate::state::server_state::RedisRateLimiters;
/// Build the various rate limiters
pub fn configure_redis_rate_limiters(common_env: &CommonEnv) -> AnyhowResult<RedisRateLimiters> {
@@ -1,3 +1,5 @@
pub mod app_startup;
pub mod plans;
pub mod static_api_tokens;
pub mod reserved_names;
pub mod static_api_tokens;
pub mod static_model;
@@ -3,10 +3,10 @@ use std::ops::Deref;
use chrono::Utc;
use reusable_types::server_environment::ServerEnvironment;
use users_component::session::lookup::user_session_extended::UserSessionExtended;
use crate::configs::plans::plan::Plan;
use crate::configs::plans::plan_list::{DEVELOPMENT_PREMIUM_PLANS_BY_SLUG, FREE_LOGGED_IN_PLAN, FREE_LOGGED_OUT_PLAN, LOYALTY_PLANS_BY_SLUG, PRODUCTION_PREMIUM_PLANS_BY_SLUG};
use crate::http_server::session::lookup::user_session_extended::UserSessionExtended;
/// Look up the most appropriate plan for the session.
/// This will probably grow to include a lot of factors.
@@ -68,10 +68,10 @@ pub mod tests {
use chrono::{Duration, Utc};
use reusable_types::server_environment::ServerEnvironment;
use users_component::session::lookup::user_session_extended::{UserSessionExtended, UserSessionSubscriptionPlan};
use crate::configs::plans::get_correct_plan_for_session::get_correct_plan_for_session;
use crate::configs::plans::plan_list::{ALL_PLANS_BY_SLUG, FREE_LOGGED_IN_PLAN, FREE_LOGGED_OUT_PLAN};
use crate::http_server::session::lookup::user_session_extended::{UserSessionExtended, UserSessionSubscriptionPlan};
#[test]
fn test_free_logged_out_plan() {
@@ -0,0 +1,6 @@
/// Full string matches that can not be used as usernames.
pub const RESERVED_USERNAMES : &str = include_str!("../../../../../../includes/binary_includes/reserved_usernames.txt");
/// Substrings that are reserved and cannot be used as part of a username.
pub const RESERVED_SUBSTRINGS : &str = include_str!("../../../../../../includes/binary_includes/reserved_usernames_including.txt");
@@ -2,7 +2,7 @@ use once_cell::sync::Lazy;
use enums::by_table::model_categories::model_type::ModelType;
use crate::model::categories::synthetic_category::SyntheticCategory;
use crate::configs::static_model::categories::synthetic_category::SyntheticCategory;
pub const SYNTHETIC_CATEGORY_LATEST_TTS_MODELS : SyntheticCategory = SyntheticCategory {
category_token: "SYNTHETIC_CATEGORY:LATEST_MODELS",
@@ -0,0 +1 @@
pub mod categories;
@@ -36,12 +36,8 @@ use tokens::tokens::prompts::*;
use tokens::tokens::user_bookmarks::*;
use tokens::tokens::users::*;
use tokens::tokens::zs_voice_datasets::*;
use users_component::common_responses::user_details_lite::{UserDefaultAvatarInfo, UserDetailsLight};
use users_component::endpoints::get_profile_handler::*;
use users_component::endpoints::login_handler::*;
use users_component::endpoints::logout_handler::*;
use users_component::endpoints::session_info_handler::*;
use crate::http_server::common_requests::media_file_token_path_info::MediaFileTokenPathInfo;
use crate::http_server::common_responses::media_file_cover_image_details::MediaFileCoverImageDetails;
use crate::http_server::common_responses::media_file_cover_image_details::MediaFileDefaultCover;
use crate::http_server::common_responses::media_file_origin_details::*;
@@ -49,6 +45,8 @@ use crate::http_server::common_responses::media_file_social_meta_lite::MediaFile
use crate::http_server::common_responses::pagination_cursors::PaginationCursors;
use crate::http_server::common_responses::pagination_page::PaginationPage;
use crate::http_server::common_responses::simple_entity_stats::SimpleEntityStats;
use crate::http_server::common_responses::simple_response::SimpleResponse;
use crate::http_server::common_responses::user_details_lite::{UserDefaultAvatarInfo, UserDetailsLight};
use crate::http_server::common_responses::weights_cover_image_details::*;
use crate::http_server::endpoints::analytics::log_browser_session_handler::*;
use crate::http_server::endpoints::beta_keys::create_beta_keys_handler::*;
@@ -70,6 +68,9 @@ use crate::http_server::endpoints::inference_job::get_inference_job_status_handl
use crate::http_server::endpoints::inference_job::list_session_jobs_handler::*;
use crate::http_server::endpoints::inference_job::terminate_inference_job_handler::*;
use crate::http_server::endpoints::media_files::delete::delete_media_file_handler::*;
use crate::http_server::endpoints::media_files::edit::change_media_file_animation_type_handler::*;
use crate::http_server::endpoints::media_files::edit::change_media_file_engine_category_handler::*;
use crate::http_server::endpoints::stats::get_unified_queue_stats_handler::*;
use crate::http_server::endpoints::media_files::edit::change_media_file_visibility_handler::*;
use crate::http_server::endpoints::media_files::edit::rename_media_file_handler::*;
use crate::http_server::endpoints::media_files::edit::set_media_file_cover_image_handler::*;
@@ -80,8 +81,8 @@ use crate::http_server::endpoints::media_files::list::list_media_files_by_batch_
use crate::http_server::endpoints::media_files::list::list_media_files_for_user_handler::*;
use crate::http_server::endpoints::media_files::list::list_media_files_handler::*;
use crate::http_server::endpoints::media_files::list::list_pinned_media_files_handler::*;
use crate::http_server::endpoints::media_files::search::search_session_media_files_handler::*;
use crate::http_server::endpoints::media_files::search::search_featured_media_files_handler::*;
use crate::http_server::endpoints::media_files::search::search_session_media_files_handler::*;
use crate::http_server::endpoints::media_files::upload::upload_audio_media_file_handler::*;
use crate::http_server::endpoints::media_files::upload::upload_engine_asset::upload_engine_asset_media_file_handler::*;
use crate::http_server::endpoints::media_files::upload::upload_error::MediaFileUploadError;
@@ -109,6 +110,10 @@ use crate::http_server::endpoints::user_bookmarks::list_user_bookmarks_for_user_
use crate::http_server::endpoints::user_ratings::batch_get_user_rating_handler::*;
use crate::http_server::endpoints::user_ratings::get_user_rating_handler::*;
use crate::http_server::endpoints::user_ratings::set_user_rating_handler::*;
use crate::http_server::endpoints::users::get_profile_handler::*;
use crate::http_server::endpoints::users::login_handler::*;
use crate::http_server::endpoints::users::logout_handler::*;
use crate::http_server::endpoints::users::session_info_handler::*;
use crate::http_server::endpoints::voice_conversion::inference::enqueue_voice_conversion_inference::*;
use crate::http_server::endpoints::voice_designer::inference::enqueue_tts_request::*;
use crate::http_server::endpoints::voice_designer::voice_datasets::list_datasets_by_user::*;
@@ -149,18 +154,21 @@ use crate::http_server::web_utils::response_success_helpers::*;
crate::http_server::endpoints::inference_job::list_session_jobs_handler::list_session_jobs_handler,
crate::http_server::endpoints::inference_job::terminate_inference_job_handler::terminate_inference_job_handler,
crate::http_server::endpoints::media_files::delete::delete_media_file_handler::delete_media_file_handler,
crate::http_server::endpoints::media_files::edit::change_media_file_animation_type_handler::change_media_file_animation_type_handler,
crate::http_server::endpoints::media_files::edit::change_media_file_engine_category_handler::change_media_file_engine_category_handler,
crate::http_server::endpoints::media_files::edit::change_media_file_visibility_handler::change_media_file_visibility_handler,
crate::http_server::endpoints::media_files::edit::rename_media_file_handler::rename_media_file_handler,
crate::http_server::endpoints::media_files::edit::set_media_file_cover_image_handler::set_media_file_cover_image_handler,
crate::http_server::endpoints::media_files::get::batch_get_media_files_handler::batch_get_media_files_handler,
crate::http_server::endpoints::stats::get_unified_queue_stats_handler::get_unified_queue_stats_handler,
crate::http_server::endpoints::media_files::get::get_media_file_handler::get_media_file_handler,
crate::http_server::endpoints::media_files::list::list_featured_media_files_handler::list_featured_media_files_handler,
crate::http_server::endpoints::media_files::list::list_media_files_by_batch_token_handler::list_media_files_by_batch_token_handler,
crate::http_server::endpoints::media_files::list::list_media_files_for_user_handler::list_media_files_for_user_handler,
crate::http_server::endpoints::media_files::list::list_media_files_handler::list_media_files_handler,
crate::http_server::endpoints::media_files::list::list_pinned_media_files_handler::list_pinned_media_files_handler,
crate::http_server::endpoints::media_files::search::search_session_media_files_handler::search_session_media_files_handler,
crate::http_server::endpoints::media_files::search::search_featured_media_files_handler::search_featured_media_files_handler,
crate::http_server::endpoints::media_files::search::search_session_media_files_handler::search_session_media_files_handler,
crate::http_server::endpoints::media_files::upload::upload_audio_media_file_handler::upload_audio_media_file_handler,
crate::http_server::endpoints::media_files::upload::upload_engine_asset::upload_engine_asset_media_file_handler::upload_engine_asset_media_file_handler,
crate::http_server::endpoints::media_files::upload::upload_generic::upload_media_file_handler::upload_media_file_handler,
@@ -186,6 +194,10 @@ use crate::http_server::web_utils::response_success_helpers::*;
crate::http_server::endpoints::user_ratings::batch_get_user_rating_handler::batch_get_user_rating_handler,
crate::http_server::endpoints::user_ratings::get_user_rating_handler::get_user_rating_handler,
crate::http_server::endpoints::user_ratings::set_user_rating_handler::set_user_rating_handler,
crate::http_server::endpoints::users::get_profile_handler::get_profile_handler,
crate::http_server::endpoints::users::login_handler::login_handler,
crate::http_server::endpoints::users::logout_handler::logout_handler,
crate::http_server::endpoints::users::session_info_handler::session_info_handler,
crate::http_server::endpoints::voice_conversion::inference::enqueue_voice_conversion_inference::enqueue_voice_conversion_inference_handler,
crate::http_server::endpoints::voice_designer::inference::enqueue_tts_request::enqueue_tts_request,
crate::http_server::endpoints::voice_designer::voice_datasets::list_datasets_by_user::list_datasets_by_user_handler,
@@ -199,10 +211,6 @@ use crate::http_server::web_utils::response_success_helpers::*;
crate::http_server::endpoints::weights::set_model_weight_cover_image_handler::set_model_weight_cover_image_handler,
crate::http_server::endpoints::weights::update_weight_handler::update_weight_handler,
crate::http_server::endpoints::workflows::enqueue_video_style_transfer_handler::enqueue_video_style_transfer_handler,
users_component::endpoints::get_profile_handler::get_profile_handler,
users_component::endpoints::login_handler::login_handler,
users_component::endpoints::logout_handler::logout_handler,
users_component::endpoints::session_info_handler::session_info_handler,
),
components(schemas(
// Tokens
@@ -239,6 +247,9 @@ use crate::http_server::web_utils::response_success_helpers::*;
WeightsCategory,
WeightsType,
// Common path info
MediaFileTokenPathInfo,
// Common response structs
MediaFileModelDetails,
MediaFileOriginDetails,
@@ -249,6 +260,7 @@ use crate::http_server::web_utils::response_success_helpers::*;
PaginationPage,
SimpleEntityStats,
SimpleGenericJsonSuccess,
SimpleResponse,
UserDetailsLight,
Visibility,
@@ -280,8 +292,12 @@ use crate::http_server::web_utils::response_success_helpers::*;
BatchStatusDetailsResponse,
BetaKeyItem,
BookmarkRow,
ByQueueStats,
ChangeMediaFileAnimationTypeError,
ChangeMediaFileAnimationTypeRequest,
ChangeMediaFileEngineCategoryError,
ChangeMediaFileEngineCategoryRequest,
ChangeMediaFileVisibilityError,
ChangeMediaFileVisibilityPathInfo,
ChangeMediaFileVisibilityRequest,
CreateBetaKeysError,
CreateBetaKeysRequest,
@@ -297,14 +313,6 @@ use crate::http_server::web_utils::response_success_helpers::*;
CreateFeaturedItemSuccessResponse,
CreateSceneError,
CreateSceneSuccessResponse,
SearchMediaFilesError,
SearchMediaFilesSuccessResponse,
SearchMediaFilesQueryParams,
SearchMediaFileListItem,
SearchFeaturedMediaFilesError,
SearchFeaturedMediaFilesSuccessResponse,
SearchFeaturedMediaFilesQueryParams,
SearchFeaturedMediaFileListItem,
CreateUserBookmarkError,
CreateUserBookmarkRequest,
CreateUserBookmarkSuccessResponse,
@@ -367,16 +375,19 @@ use crate::http_server::web_utils::response_success_helpers::*;
GetPromptError,
GetPromptPathInfo,
GetPromptSuccessResponse,
GetUnifiedQueueStatsError,
GetUnifiedQueueStatsSuccessResponse,
GetUserRatingError,
GetUserRatingResponse,
GetWeightError,
GetWeightPathInfo,
GetWeightResponse,
InferenceJobStatusResponsePayload,
InferenceJobTokenType,
InferTtsError,
InferTtsRequest,
InferTtsSuccessResponse,
InferenceJobStatusResponsePayload,
InferenceJobTokenType,
LegacyQueueDetails,
ListActiveUserSubscriptionsError,
ListActiveUserSubscriptionsResponse,
ListAvailableWeightsQuery,
@@ -397,7 +408,6 @@ use crate::http_server::web_utils::response_success_helpers::*;
ListFeaturedWeightsError,
ListFeaturedWeightsQueryParams,
ListFeaturedWeightsSuccessResponse,
SubscriptionProductKey,
ListMediaFilesByBatchError,
ListMediaFilesByBatchPathInfo,
ListMediaFilesByBatchSuccessResponse,
@@ -442,10 +452,11 @@ use crate::http_server::web_utils::response_success_helpers::*;
MediaFileForUserListItem,
MediaFileInfo,
MediaFileListItem,
MediaFileUploadError,
MediaFilesByBatchListItem,
MediaFileUploadError,
ModelWeightForList,
ModelWeightSearchResult,
ModernInferenceQueueStats,
PinnedMediaFile,
PinnedModelWeightForList,
ProfileError,
@@ -455,10 +466,17 @@ use crate::http_server::web_utils::response_success_helpers::*;
RedeemBetaKeyRequest,
RedeemBetaKeySuccessResponse,
RenameMediaFileError,
RenameMediaFilePathInfo,
RenameMediaFileRequest,
RequestDetailsResponse,
ResultDetailsResponse,
SearchFeaturedMediaFileListItem,
SearchFeaturedMediaFilesError,
SearchFeaturedMediaFilesQueryParams,
SearchFeaturedMediaFilesSuccessResponse,
SearchMediaFileListItem,
SearchMediaFilesError,
SearchMediaFilesQueryParams,
SearchMediaFilesSuccessResponse,
SearchModelWeightsError,
SearchModelWeightsRequest,
SearchModelWeightsSuccessResponse,
@@ -466,7 +484,6 @@ use crate::http_server::web_utils::response_success_helpers::*;
SessionInfoSuccessResponse,
SessionUserInfo,
SetMediaFileCoverImageError,
SetMediaFileCoverImagePathInfo,
SetMediaFileCoverImageRequest,
SetModelWeightCoverImageError,
SetModelWeightCoverImagePathInfo,
@@ -481,6 +498,7 @@ use crate::http_server::web_utils::response_success_helpers::*;
StatusAlertResponse,
StatusDetailsResponse,
StorytellerStreamPlan,
SubscriptionProductKey,
TerminateInferenceJobError,
TerminateInferenceJobPathInfo,
TerminateInferenceJobSuccessResponse,
@@ -0,0 +1 @@
pub mod api_doc;
@@ -0,0 +1,8 @@
use utoipa::ToSchema;
use tokens::tokens::media_files::MediaFileToken;
/// Media File Token in URL PathInfo
#[derive(Deserialize, ToSchema)]
pub struct MediaFileTokenPathInfo {
pub token: MediaFileToken,
}
@@ -0,0 +1 @@
pub mod media_file_token_path_info;
@@ -1,7 +1,10 @@
pub mod weights_cover_image_details;
pub mod media_file_cover_image_details;
pub mod media_file_origin_details;
pub mod media_file_social_meta_lite;
pub mod pagination_cursors;
pub mod pagination_page;
pub mod simple_entity_stats;
pub mod simple_response;
pub mod user_avatars;
pub mod user_details_lite;
pub mod weights_cover_image_details;
@@ -0,0 +1,6 @@
use utoipa::ToSchema;
#[derive(Serialize, ToSchema)]
pub struct SimpleResponse {
pub success: bool,
}
@@ -33,7 +33,7 @@ mod tests {
use rand::distributions::{Alphanumeric, DistString};
use crate::common_responses::user_avatars::default_avatar_color_from_username::default_avatar_color_from_username;
use crate::http_server::common_responses::user_avatars::default_avatar_color_from_username::default_avatar_color_from_username;
#[test]
fn test_stability() {
@@ -32,7 +32,7 @@ mod tests {
use rand::distributions::{Alphanumeric, DistString};
use crate::common_responses::user_avatars::default_avatar_from_username::default_avatar_from_username;
use crate::http_server::common_responses::user_avatars::default_avatar_from_username::default_avatar_from_username;
#[test]
fn test_stability() {
@@ -2,8 +2,8 @@ use utoipa::ToSchema;
use tokens::tokens::users::UserToken;
use crate::common_responses::user_avatars::default_avatar_color_from_username::default_avatar_color_from_username;
use crate::common_responses::user_avatars::default_avatar_from_username::default_avatar_from_username;
use crate::http_server::common_responses::user_avatars::default_avatar_color_from_username::default_avatar_color_from_username;
use crate::http_server::common_responses::user_avatars::default_avatar_from_username::default_avatar_from_username;
/// Everything we need to refer to a user on the public web interface.
#[derive(Clone, Serialize, ToSchema)]
@@ -107,7 +107,7 @@ impl UserDefaultAvatarInfo {
mod tests {
use tokens::tokens::users::UserToken;
use crate::common_responses::user_details_lite::UserDetailsLight;
use crate::http_server::common_responses::user_details_lite::UserDetailsLight;
#[test]
fn test_from_optional_db_fields() {
@@ -8,7 +8,7 @@ use cookies::jwt_signer::JwtSigner;
use errors::AnyhowResult;
use tokens::tokens::anonymous_visitor_tracking::AnonymousVisitorTrackingToken;
use crate::cookies::anonymous_visitor_tracking::avt_cookie_payload::AvtCookiePayload;
use crate::http_server::cookies::anonymous_visitor_tracking::avt_cookie_payload::AvtCookiePayload;
const VISITOR_COOKIE_NAME : &str = "visitor";
@@ -68,7 +68,8 @@ impl AvtCookiePayload {
#[cfg(test)]
mod tests {
use tokens::tokens::anonymous_visitor_tracking::AnonymousVisitorTrackingToken;
use crate::cookies::anonymous_visitor_tracking::avt_cookie_payload::AvtCookiePayload;
use crate::http_server::cookies::anonymous_visitor_tracking::avt_cookie_payload::AvtCookiePayload;
#[test]
fn test_new() {
@@ -32,7 +32,7 @@ use crate::http_server::endpoints::moderation::user_feature_flags::edit_user_fea
use crate::http_server::web_utils::require_moderator::{require_moderator, RequireModeratorError};
use crate::http_server::web_utils::require_user_session::RequireUserSessionError;
use crate::http_server::web_utils::response_error_helpers::to_simple_json_error;
use crate::server_state::ServerState;
use crate::state::server_state::ServerState;
#[derive(Deserialize, ToSchema)]
pub struct LogBrowserSessionRequest {
@@ -24,11 +24,11 @@ use tokens::tokens::media_files::MediaFileToken;
use tokens::tokens::media_uploads::MediaUploadToken;
use tokens::tokens::users::UserToken;
use tokens::tokens::voice_conversion_results::VoiceConversionResultToken;
use users_component::session::lookup::user_session_extended::UserSessionExtended;
use crate::http_server::session::lookup::user_session_extended::UserSessionExtended;
use crate::configs::plans::get_correct_plan_for_session::get_correct_plan_for_session;
use crate::http_server::web_utils::response_error_helpers::to_simple_json_error;
use crate::server_state::ServerState;
use crate::state::server_state::ServerState;
/// Debug requests can get routed to special "debug-only" workers, which can
/// be used to trial new code, run debugging, etc.
@@ -23,11 +23,11 @@ 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 users_component::session::lookup::user_session_extended::UserSessionExtended;
use crate::http_server::session::lookup::user_session_extended::UserSessionExtended;
use crate::configs::plans::get_correct_plan_for_session::get_correct_plan_for_session;
use crate::http_server::web_utils::response_error_helpers::to_simple_json_error;
use crate::server_state::ServerState;
use crate::state::server_state::ServerState;
/// Debug requests can get routed to special "debug-only" workers, which can
/// be used to trial new code, run debugging, etc.
@@ -10,7 +10,7 @@ use http_server_common::request::get_request_ip::get_request_ip;
use http_server_common::response::serialize_as_json_error::serialize_as_json_error;
use mysql_queries::queries::api_tokens::create_api_token::create_api_token_for_user;
use crate::server_state::ServerState;
use crate::state::server_state::ServerState;
// =============== Request ===============
@@ -12,7 +12,7 @@ use http_server_common::response::serialize_as_json_error::serialize_as_json_err
use mysql_queries::queries::api_tokens::delete_api_token::delete_api_token;
use mysql_queries::queries::api_tokens::list_available_api_tokens_for_user::list_available_api_tokens_for_user;
use crate::server_state::ServerState;
use crate::state::server_state::ServerState;
// =============== Request ===============

Some files were not shown because too many files have changed in this diff Show More