studio shot handler WIP - handle zip archives

This commit is contained in:
Brandon Thomas committed 2024-08-25 11:32:35 -04:00
1 parent 3121eb75ac
commit ba369dab95
11 files changed
+320 -99

No files matched your search

Generated
+1
View File
@@ -6862,6 +6862,7 @@ dependencies = [
"strum",
"strum_macros",
"symphonia",
"tempdir",
"thumbnail_generator",
"time",
"tokens",
@@ -133,6 +133,7 @@ lru_time_cache = "0.11.11"
# Files
infer = "0.16.0"
tempdir = "0.3.7"
# The "never-encode-slash" feature seems to have been removed:
# https://github.com/durch/rust-s3/pull/254
@@ -0,0 +1,154 @@
use std::collections::{HashMap, HashSet};
use std::io::{BufReader, Cursor, Read};
use std::path::{Path, PathBuf};
use log::{error, info, warn};
use once_cell::sync::Lazy;
use zip::ZipArchive;
use filesys::path_to_string::path_to_string;
use crate::http_server::web_utils::open_zip_archive::{open_zip_archive, OpenZipError};
static ALLOWED_FRAME_EXTENSIONS: Lazy<HashSet<&'static str>> = Lazy::new(|| {
HashSet::from([
".png",
".jpg",
])
});
const TOO_MANY_FILES_COUNT: usize = 8_000;
#[derive(Debug)]
pub enum ExtractFramesError {
InvalidArchive,
NoImageFiles,
TooFewImageFiles,
TooManyFiles,
ExtractionError,
UploadError,
FileError,
}
pub fn extract_frames_from_zip<P: AsRef<Path>>(
zip_container_file_bytes: &[u8],
frame_temp_dir: P,
) -> Result<(), ExtractFramesError> {
info!("Opening archive...");
let mut archive = match open_zip_archive(zip_container_file_bytes, Some(TOO_MANY_FILES_COUNT)) {
Ok(archive) => archive,
Err(OpenZipError::InvalidArchive) => return Err(ExtractFramesError::InvalidArchive),
Err(OpenZipError::TooManyFiles) => return Err(ExtractFramesError::TooManyFiles),
};
info!("Reading archive contents...");
let entries = get_relevant_zip_entries(&mut archive)?;
for entry in entries.iter() {
info!("Entry: {:?}", entry);
if !entry.is_frame {
info!("Skipping entry (not a frame): {:?}", entry);
continue;
}
let safe_enclosed_name = path_to_string(&entry.safe_enclosed_name);
let mut file = archive.by_name(&safe_enclosed_name)
.map_err(|err| {
error!("Problem reading file from archive: {:?}", err);
ExtractFramesError::InvalidArchive
})?;
let maybe_filesystem_name = entry.safe_enclosed_name.file_name()
.map(|s| s.to_str())
.flatten();
let filesystem_name = match maybe_filesystem_name {
None => continue,
Some(name) => name,
};
let output_path = frame_temp_dir.as_ref().join(filesystem_name);
let mut output_file = std::fs::File::create(&output_path)
.map_err(|err| {
error!("Problem creating file {:?} : {:?}", output_path, err);
ExtractFramesError::ExtractionError
})?;
std::io::copy(&mut file, &mut output_file)
.map_err(|err| {
error!("Problem copying file bytes {:?} : {:?}", output_path, err);
ExtractFramesError::ExtractionError
})?;
}
Ok(())
}
#[derive(Debug, Clone)]
struct FrameEntryDetail {
safe_enclosed_name: PathBuf,
is_frame: bool,
//file_size: u64,
//maybe_better_alternative_output_name: Option<PathBuf>,
//sequence_number: u64, // Based on the filename, e.g. "0001.png" -> 1
}
fn get_relevant_zip_entries(archive: &mut ZipArchive<BufReader<Cursor<&[u8]>>>) -> Result<Vec<FrameEntryDetail>, ExtractFramesError> {
let mut entries = Vec::new();
for i in 0..(archive.len()) {
info!("Reading file {}...", i);
let mut file = archive.by_index(i)
.map_err(|err| {
error!("Problem reading file from archive: {:?}", err);
ExtractFramesError::InvalidArchive
})?;
let filename = file.name();
let filename_lowercase = filename.to_lowercase();
info!("File {} is {:?} - is file = {}", i, filename, file.is_file());
info!("Enclosed name: {:?}", file.enclosed_name());
if file.is_dir() {
continue;
}
if filename_lowercase.starts_with("__macosx/") {
// Mac users sometimes have a bogus __MACOSX directory, which may double the file count.
continue;
}
let enclosed_name = match file.enclosed_name() {
None => return Err(ExtractFramesError::FileError),
Some(name) => name,
};
let file_is_frame = ALLOWED_FRAME_EXTENSIONS.iter()
.any(|ext| filename_lowercase.ends_with(ext));
if file_is_frame {
entries.push(FrameEntryDetail {
safe_enclosed_name: enclosed_name.to_path_buf(),
is_frame: true,
//maybe_better_alternative_output_name: None,
//file_size: file.size(),
});
}
// TODO: We may have metadata files or sound files in the future.
}
for entry in entries.iter() {
info!("Entry: {:?}", entry);
}
Ok(entries)
}
@@ -1 +1,2 @@
pub mod extract_frames_from_zip;
pub mod upload_studio_shot_media_file_handler;
@@ -33,9 +33,11 @@ use crate::http_server::endpoints::media_files::upload::upload_engine_asset::dra
use crate::http_server::endpoints::media_files::upload::upload_error::MediaFileUploadError;
use crate::http_server::endpoints::media_files::upload::upload_new_scene_media_file_handler::UploadNewSceneMediaFileForm;
use crate::http_server::endpoints::media_files::upload::upload_pmx::extract_and_upload_pmx_files::{extract_and_upload_pmx_files, PmxError};
use crate::http_server::endpoints::media_files::upload::upload_studio_shot::extract_frames_from_zip::{extract_frames_from_zip, ExtractFramesError};
use crate::http_server::web_utils::user_session::require_moderator::{require_moderator, RequireModeratorError, UseDatabase};
use crate::state::server_state::ServerState;
use crate::http_server::validations::validate_idempotency_token_format::validate_idempotency_token_format;
use crate::http_server::web_utils::user_session::require_user_session_using_connection::require_user_session_using_connection;
/// Form-multipart request fields.
///
@@ -54,12 +56,6 @@ pub struct UploadStudioShotFileForm {
#[schema(value_type = Vec<u8>, format = Binary)]
file: TempFile,
/// The category of engine asset: character, animation, etc.
/// See the documentation on `MediaFileEngineCategory`.
#[multipart(limit = "2 KiB")]
#[schema(value_type = String, format = Binary)]
engine_category: Text<MediaFileEngineCategory>,
/// Optional: Title (name) of the scene
#[multipart(limit = "2 KiB")]
#[schema(value_type = Option<String>, format = Binary)]
@@ -70,12 +66,6 @@ pub struct UploadStudioShotFileForm {
#[schema(value_type = Option<String>, format = Binary)]
maybe_visibility: Option<Text<Visibility>>,
/// Optional: the type of animation (if this is a character or animation)
/// See the documentation on `MediaFileAnimationType`.
#[multipart(limit = "2 KiB")]
#[schema(value_type = Option<String>, format = Binary)]
maybe_animation_type: Option<Text<MediaFileAnimationType>>,
/// Optional: The duration, for files that are animations.
#[multipart(limit = "2 KiB")]
#[schema(value_type = Option<u64>, format = Binary)]
@@ -92,7 +82,7 @@ pub struct UploadStudioShotSuccessResponse {
#[utoipa::path(
post,
tag = "Media Files (Upload)",
path = "/v1/media_files/upload/pmx",
path = "/v1/media_files/upload/studio_shot",
responses(
(status = 200, description = "Success Update", body = UploadStudioShotSuccessResponse),
(status = 400, description = "Bad input", body = MediaFileUploadError),
@@ -124,28 +114,17 @@ pub async fn upload_studio_shot_media_file_handler(
// ==================== READ SESSION ==================== //
// NB: We require a moderator to upload PMX files.
let user_session = require_moderator(&http_request, &server_state, UseDatabase::FromPool(&mut mysql_connection))
let user_session = require_user_session_using_connection(&http_request, &server_state.session_checker, &mut mysql_connection)
.await
.map_err(|err| match err {
RequireModeratorError::ServerError => MediaFileUploadError::ServerError,
RequireModeratorError::NotAuthorized => MediaFileUploadError::NotAuthorized,
.map_err(|e| {
error!("User session error: {:?}", e);
MediaFileUploadError::NotAuthorized
})?;
let maybe_avt_token = server_state
.avt_cookie_manager
.get_avt_token_from_request(&http_request);
// // ==================== READ SESSION ==================== //
//
// let maybe_user_session = server_state
// .session_checker
// .maybe_get_user_session_from_connection(&http_request, &mut mysql_connection)
// .await
// .map_err(|e| {
// error!("Session checker error: {:?}", e);
// MediaFileUploadError::ServerError
// })?;
//
// // ==================== BANNED USERS ==================== //
//
// if let Some(ref user) = maybe_user_session {
@@ -183,18 +162,6 @@ pub async fn upload_studio_shot_media_file_handler(
// ==================== UPLOAD METADATA ==================== //
let engine_category = form.engine_category.0;
let maybe_duration_millis = form.maybe_duration_millis
.map(|duration| duration.0);
let mut maybe_animation_type = form.maybe_animation_type
.map(|t| t.0);
if engine_category == MediaFileEngineCategory::Expression {
// NB: Expressions are exclusively ArKit for now (and probably well into the future).
maybe_animation_type = Some(MediaFileAnimationType::ArKit);
}
let maybe_title = form.maybe_title
.map(|title| title.trim().to_string())
@@ -202,12 +169,9 @@ pub async fn upload_studio_shot_media_file_handler(
let creator_set_visibility = form.maybe_visibility
.map(|visibility| visibility.0)
.or_else(|| {
//maybe_user_session
// .as_ref()
// .map(|user_session| user_session.preferred_tts_result_visibility)
Some(user_session.preferred_tts_result_visibility)
})
//.or_else(|| {
// Some(user_session.preferred_tts_result_visibility)
//})
.unwrap_or(Visibility::default());
// ==================== USER DATA ==================== //
@@ -228,13 +192,6 @@ pub async fn upload_studio_shot_media_file_handler(
.and_then(|filename| filename.extension())
.and_then(|ext| ext.to_str());
let mut file_bytes = Vec::new();
form.file.file.read_to_end(&mut file_bytes)
.map_err(|e| {
error!("Problem reading file: {:?}", e);
MediaFileUploadError::ServerError
})?;
match maybe_file_extension {
Some("zip") => {},
_ => {
@@ -243,55 +200,78 @@ pub async fn upload_studio_shot_media_file_handler(
}
}
// ==================== UPLOAD AND SAVE ==================== //
const PREFIX : Option<&str> = Some("upload_");
const SUFFIX : Option<&str> = Some(".pmx");
let pmx_details = extract_and_upload_pmx_files(&file_bytes, &server_state.public_bucket_client, PREFIX, SUFFIX)
.await
.map_err(|err| {
warn!("Extract and upload pmx error: {:?}", err);
match err {
PmxError::InvalidArchive => MediaFileUploadError::ServerErrorVerbose("invalid archive file".to_string()),
PmxError::TooManyFiles => MediaFileUploadError::ServerErrorVerbose("too many files".to_string()),
PmxError::NoPmxFile => MediaFileUploadError::ServerErrorVerbose("no pmx files".to_string()),
PmxError::UploadError => MediaFileUploadError::ServerErrorVerbose("upload error".to_string()),
PmxError::FileError => MediaFileUploadError::ServerErrorVerbose("file error".to_string()),
PmxError::ExtractionError => MediaFileUploadError::ServerErrorVerbose("zip extraction error".to_string()),
}
})?;
// TODO(bt, 2024-02-22): This should be a transaction.
let (token, record_id) = insert_media_file_from_file_upload(InsertMediaFileFromUploadArgs {
maybe_media_class: Some(MediaFileClass::Dimensional),
media_file_type: MediaFileType::Pmx,
maybe_creator_user_token: Some(&user_session.get_strongly_typed_user_token()),
maybe_creator_anonymous_visitor_token: maybe_avt_token.as_ref(),
creator_ip_address: &ip_address,
creator_set_visibility,
upload_type: UploadType::Filesystem,
maybe_engine_category: Some(engine_category),
maybe_animation_type,
maybe_mime_type: Some("application/octet-stream"),
file_size_bytes: pmx_details.file_size_bytes,
maybe_duration_millis,
sha256_checksum: &pmx_details.sha256_checksum,
maybe_scene_source_media_file_token: None,
is_intermediate_system_file: false,
maybe_title: maybe_title.as_deref(),
public_bucket_directory_hash: pmx_details.pmx_public_upload_path.get_object_hash(),
maybe_public_bucket_prefix: PREFIX,
maybe_public_bucket_extension: SUFFIX,
pool: &server_state.mysql_pool,
})
.await
.map_err(|err| {
warn!("New file creation DB error: {:?}", err);
let mut file_bytes = Vec::new();
form.file.file.read_to_end(&mut file_bytes)
.map_err(|e| {
error!("Problem reading file: {:?}", e);
MediaFileUploadError::ServerError
})?;
info!("new media file id: {} token: {:?}", record_id, &token);
// ==================== EXTRACT ==================== //
// TODO(bt,2024-08-25): Should include entropy so concurrent requests don't overwrite
let frame_temp_dir = server_state.temp_dir_creator.new_tempdir("frames")
.map_err(|err| {
error!("Problem creating temp dir: {:?}", err);
MediaFileUploadError::ServerError
})?;
extract_frames_from_zip(&file_bytes, frame_temp_dir)
.map_err(|err| {
warn!("Extract frames error: {:?}", err);
match err {
ExtractFramesError::InvalidArchive => MediaFileUploadError::ServerErrorVerbose("invalid archive file".to_string()),
ExtractFramesError::NoImageFiles => MediaFileUploadError::ServerErrorVerbose("no image files".to_string()),
ExtractFramesError::TooFewImageFiles => MediaFileUploadError::ServerErrorVerbose("too few image files".to_string()),
ExtractFramesError::TooManyFiles => MediaFileUploadError::ServerErrorVerbose("too many files".to_string()),
ExtractFramesError::ExtractionError => MediaFileUploadError::ServerErrorVerbose("zip extraction error".to_string()),
ExtractFramesError::UploadError => MediaFileUploadError::ServerErrorVerbose("upload error".to_string()),
ExtractFramesError::FileError => MediaFileUploadError::ServerErrorVerbose("file error".to_string()),
}
})?;
// ==================== FFMPEG ==================== //
// ==================== UPLOAD AND SAVE ==================== //
// TODO DELETE ME
let token = MediaFileToken::new("test_token".to_string());
// const PREFIX : Option<&str> = Some("upload_");
// const SUFFIX : Option<&str> = Some(".pmx");
//
// // TODO(bt, 2024-02-22): This should be a transaction.
// let (token, record_id) = insert_media_file_from_file_upload(InsertMediaFileFromUploadArgs {
// maybe_media_class: Some(MediaFileClass::Dimensional),
// media_file_type: MediaFileType::Pmx,
// maybe_creator_user_token: Some(&user_session.get_strongly_typed_user_token()),
// maybe_creator_anonymous_visitor_token: maybe_avt_token.as_ref(),
// creator_ip_address: &ip_address,
// creator_set_visibility,
// upload_type: UploadType::Filesystem,
// maybe_engine_category: Some(engine_category),
// maybe_animation_type,
// maybe_mime_type: Some("application/octet-stream"),
// file_size_bytes: pmx_details.file_size_bytes,
// maybe_duration_millis,
// sha256_checksum: &pmx_details.sha256_checksum,
// maybe_scene_source_media_file_token: None,
// is_intermediate_system_file: false,
// maybe_title: maybe_title.as_deref(),
// public_bucket_directory_hash: pmx_details.pmx_public_upload_path.get_object_hash(),
// maybe_public_bucket_prefix: PREFIX,
// maybe_public_bucket_extension: SUFFIX,
// pool: &server_state.mysql_pool,
// })
// .await
// .map_err(|err| {
// warn!("New file creation DB error: {:?}", err);
// MediaFileUploadError::ServerError
// })?;
//
// info!("new media file id: {} token: {:?}", record_id, &token);
Ok(Json(UploadStudioShotSuccessResponse {
success: true,
@@ -29,6 +29,7 @@ use crate::http_server::endpoints::media_files::upload::upload_new_video_media_f
use crate::http_server::endpoints::media_files::upload::upload_pmx::upload_pmx_media_file_handler::upload_pmx_media_file_handler;
use crate::http_server::endpoints::media_files::upload::upload_saved_scene_media_file_handler::upload_saved_scene_media_file_handler;
use crate::http_server::endpoints::media_files::upload::upload_scene_snapshot_media_file_handler::upload_scene_snapshot_media_file_handler;
use crate::http_server::endpoints::media_files::upload::upload_studio_shot::upload_studio_shot_media_file_handler::upload_studio_shot_media_file_handler;
use crate::http_server::endpoints::media_files::upload::upload_video::upload_video_media_file_handler::upload_video_media_file_handler;
use crate::http_server::endpoints::media_files::upsert_upload::write_engine_asset::write_engine_asset_media_file_handler::write_engine_asset_media_file_handler;
use crate::http_server::endpoints::media_files::upsert_upload::write_scene_file::write_scene_file_media_file_handler::write_scene_file_media_file_handler;
@@ -151,6 +152,10 @@ pub fn add_media_file_routes<T, B> (app: App<T>) -> App<T>
.route(web::post().to(upload_saved_scene_media_file_handler))
.route(web::head().to(|| HttpResponse::Ok()))
)
.service(web::resource("/upload/studio_shot")
.route(web::post().to(upload_studio_shot_media_file_handler))
.route(web::head().to(|| HttpResponse::Ok()))
)
.service(web::resource("/upload/audio")
.route(web::post().to(upload_audio_media_file_handler))
.route(web::head().to(|| HttpResponse::Ok()))
@@ -2,10 +2,12 @@ pub mod bucket_urls;
pub mod filter_model_name;
pub mod get_host_header;
pub mod handle_multipart_error;
pub mod open_zip_archive;
pub mod read_multipart_field_bytes;
pub mod redis_rate_limiter;
pub mod response_error_helpers;
pub mod response_success_helpers;
pub mod scoped_temp_dir_creator;
pub mod serialize_as_json_error;
pub mod try_delete_session_cache;
pub mod user_session;
@@ -0,0 +1,44 @@
use std::io::{BufReader, Cursor};
use log::{error, warn};
use zip::ZipArchive;
use mimetypes::mimetype_for_bytes::get_mimetype_for_bytes;
const ZIP_MIMETYPE : &str = "application/zip";
#[derive(Debug)]
pub enum OpenZipError {
InvalidArchive,
TooManyFiles,
}
pub fn open_zip_archive(
zip_container_file_bytes: &[u8],
maybe_max_file_count: Option<usize>,
) -> Result<ZipArchive<BufReader<Cursor<&[u8]>>>, OpenZipError>{
let maybe_mimetype = get_mimetype_for_bytes(zip_container_file_bytes);
if maybe_mimetype != Some(ZIP_MIMETYPE) {
error!("File must be an application/zip");
return Err(OpenZipError::InvalidArchive);
}
let mut cursor = Cursor::new(zip_container_file_bytes);
let reader = BufReader::new(cursor);
let mut archive = ZipArchive::new(reader)
.map_err(|err| {
error!("Problem reading zip archive: {:?}", err);
OpenZipError::InvalidArchive
})?;
if let Some(max_file_count) = maybe_max_file_count {
if archive.len() > max_file_count {
return Err(OpenZipError::TooManyFiles);
}
}
Ok(archive)
}
@@ -0,0 +1,28 @@
use std::path::{Path, PathBuf};
use tempdir::TempDir;
/// This utility will always create TempDirs at a specific mount path.
/// This is useful for k8s deployments where we may need our TempDirs
/// to live on a specific volume.
#[derive(Clone)]
pub struct ScopedTempDirCreator {
base_dir: PathBuf,
}
impl ScopedTempDirCreator {
pub fn for_directory<P: AsRef<Path>>(base_dir: P) -> Self {
Self {
base_dir: PathBuf::from(&base_dir.as_ref()),
}
}
pub fn get_base_dir(&self) -> &Path {
&self.base_dir
}
pub fn new_tempdir(&self, prefix: &str) -> std::io::Result<TempDir> {
TempDir::new_in(&self.base_dir, prefix)
}
}
@@ -82,6 +82,7 @@ use crate::http_server::routes::add_routes::add_routes;
use crate::http_server::session::http::http_user_session_manager::HttpUserSessionManager;
use crate::http_server::session::session_checker::SessionChecker;
use crate::http_server::web_utils::handle_multipart_error::handle_multipart_error;
use crate::http_server::web_utils::scoped_temp_dir_creator::ScopedTempDirCreator;
use crate::state::memory_cache::model_token_to_info_cache::ModelTokenToInfoCache;
use crate::state::server_state::{DurableInMemoryCaches, EnvConfig, EphemeralInMemoryCaches, InMemoryCaches, ServerInfo, ServerState, StaticFeatureFlags, StripeSettings, TrollBans};
use crate::threads::db_health_checker_thread::db_health_check_status::HealthCheckStatus;
@@ -461,6 +462,7 @@ async fn main() -> AnyhowResult<()> {
user_tokens: user_token_troll_bans,
ip_addresses: ip_address_troll_bans,
},
temp_dir_creator: ScopedTempDirCreator::for_directory(easyenv::get_env_pathbuf_or_default("TEMP_DIR", "/tmp")),
};
serve(server_state)
@@ -32,6 +32,7 @@ use crate::http_server::endpoints::voice_conversion::models::list_voice_conversi
use crate::http_server::session::http::http_user_session_manager::HttpUserSessionManager;
use crate::http_server::session::session_checker::SessionChecker;
use crate::http_server::web_utils::redis_rate_limiter::RedisRateLimiter;
use crate::http_server::web_utils::scoped_temp_dir_creator::ScopedTempDirCreator;
use crate::state::memory_cache::model_token_to_info_cache::ModelTokenToInfoCache;
use crate::threads::db_health_checker_thread::db_health_check_status::HealthCheckStatus;
use crate::util::encrypted_sort_id::SortKeyCrypto;
@@ -95,6 +96,8 @@ pub struct ServerState {
pub static_api_token_set: StaticApiTokenSet,
pub caches: InMemoryCaches,
pub temp_dir_creator: ScopedTempDirCreator,
}
#[derive(Clone)]