diff --git a/Cargo.lock b/Cargo.lock index 28a6674963..7c2330105a 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -756,6 +756,7 @@ dependencies = [ "egui", "enums", "errors", + "fakeyou_client", "files", "glium", "hashing", diff --git a/crates/_types/enums/src/by_table/mod.rs b/crates/_types/enums/src/by_table/mod.rs index 230f8906bd..60a07c8927 100644 --- a/crates/_types/enums/src/by_table/mod.rs +++ b/crates/_types/enums/src/by_table/mod.rs @@ -1,6 +1,10 @@ +// ===== MySql ===== pub mod media_uploads; pub mod model_categories; pub mod trending_model_analytics; pub mod user_ratings; -pub mod web_rendition_targets; // NB: SqLite -pub mod web_scraping_targets; // NB: Sqlite + +// ===== Sqlite ===== +pub mod tts_render_targets; +pub mod web_rendition_targets; +pub mod web_scraping_targets; diff --git a/crates/_types/enums/src/by_table/tts_render_targets/mod.rs b/crates/_types/enums/src/by_table/tts_render_targets/mod.rs new file mode 100644 index 0000000000..1eb035720c --- /dev/null +++ b/crates/_types/enums/src/by_table/tts_render_targets/mod.rs @@ -0,0 +1 @@ +pub mod tts_render_status; \ No newline at end of file diff --git a/crates/_types/enums/src/by_table/tts_render_targets/tts_render_status.rs b/crates/_types/enums/src/by_table/tts_render_targets/tts_render_status.rs new file mode 100644 index 0000000000..678c62e717 --- /dev/null +++ b/crates/_types/enums/src/by_table/tts_render_targets/tts_render_status.rs @@ -0,0 +1,97 @@ +use serde::Deserialize; +use serde::Serialize; + +#[cfg(test)] +use strum::EnumCount; +#[cfg(test)] +use strum::EnumIter; + +/// Used in the SqLite `web_scraping_targets` table in a `TEXT` field named `scraping_status`. +#[cfg_attr(test, derive(EnumIter, EnumCount))] +#[derive(Clone, Copy, Eq, PartialEq, Hash, Deserialize, Serialize)] +pub enum TtsRenderStatus { + #[serde(rename = "new")] + New, + + #[serde(rename = "skipped")] + Skipped, + + #[serde(rename = "failed")] + Failed, + + #[serde(rename = "permanently_failed")] + PermanentlyFailed, + + #[serde(rename = "success")] + Success, +} + +// TODO(bt, 2023-01-17): This desperately needs MySQL integration tests! +impl_enum_display_and_debug_using_to_str!(TtsRenderStatus); +impl_sqlite_enum_coders!(TtsRenderStatus); + +/// NB: Legacy API for older code. +impl TtsRenderStatus { + pub fn to_str(&self) -> &'static str { + match self { + Self::New => "new", + Self::Skipped => "skipped", + Self::Failed => "failed", + Self::PermanentlyFailed => "permanently_failed", + Self::Success => "success", + } + } + + pub fn from_str(value: &str) -> Result { + match value { + "new" => Ok(Self::New), + "skipped" => Ok(Self::Skipped), + "failed" => Ok(Self::Failed), + "permanently_failed" => Ok(Self::PermanentlyFailed), + "success" => Ok(Self::Success), + _ => Err(format!("invalid value: {:?}", value)), + } + } +} + +#[cfg(test)] +mod tests { + use crate::by_table::tts_render_targets::tts_render_status::TtsRenderStatus; + use crate::test_helpers::assert_serialization; + + mod serde { + use super::*; + + #[test] + fn test_serialization() { + assert_serialization(TtsRenderStatus::New, "new"); + assert_serialization(TtsRenderStatus::Skipped, "skipped"); + assert_serialization(TtsRenderStatus::Failed, "failed"); + assert_serialization(TtsRenderStatus::PermanentlyFailed, "permanently_failed"); + assert_serialization(TtsRenderStatus::Success, "success"); + } + } + + mod impl_methods { + use super::*; + + #[test] + fn test_to_str() { + assert_eq!(TtsRenderStatus::New.to_str(), "new"); + assert_eq!(TtsRenderStatus::Skipped.to_str(), "skipped"); + assert_eq!(TtsRenderStatus::Failed.to_str(), "failed"); + assert_eq!(TtsRenderStatus::PermanentlyFailed.to_str(), "permanently_failed"); + assert_eq!(TtsRenderStatus::Success.to_str(), "success"); + } + + #[test] + fn test_from_str() { + assert_eq!(TtsRenderStatus::from_str("new").unwrap(), TtsRenderStatus::New); + assert_eq!(TtsRenderStatus::from_str("skipped").unwrap(), TtsRenderStatus::Skipped); + assert_eq!(TtsRenderStatus::from_str("failed").unwrap(), TtsRenderStatus::Failed); + assert_eq!(TtsRenderStatus::from_str("permanently_failed").unwrap(), TtsRenderStatus::PermanentlyFailed); + assert_eq!(TtsRenderStatus::from_str("success").unwrap(), TtsRenderStatus::Success); + assert!(TtsRenderStatus::from_str("foo").is_err()); + } + } +} diff --git a/crates/_types/tokens/src/prefixes.rs b/crates/_types/tokens/src/prefixes.rs index 7bfe7fc5d6..db4373fa9c 100644 --- a/crates/_types/tokens/src/prefixes.rs +++ b/crates/_types/tokens/src/prefixes.rs @@ -19,6 +19,7 @@ pub(crate) enum EntityType { ModelCategory, NewsStory, // NB: aichatbot / sqlite TtsModel, + TtsRenderTask, // NB: aichatbot / sqlite User, VoiceConversionModel, W2lTemplate, @@ -34,6 +35,7 @@ impl EntityType { Self::ModelCategory => "CAT:", // NB: Old-style prefix, do not use for future tokens. Self::NewsStory => "news_story_", Self::TtsModel => "TM:", // NB: Old-style prefix, do not use for future tokens. + Self::TtsRenderTask => "tts_task_", Self::User => "U:", // NB: Old-style prefix, do not use for future tokens. Self::VoiceConversionModel => "voco_", Self::W2lTemplate => "WT:", // NB: Old-style prefix, do not use for future tokens. diff --git a/crates/_types/tokens/src/tokens/mod.rs b/crates/_types/tokens/src/tokens/mod.rs index 871a0580b4..567f1af03f 100644 --- a/crates/_types/tokens/src/tokens/mod.rs +++ b/crates/_types/tokens/src/tokens/mod.rs @@ -1,4 +1,8 @@ +// ===== For MySQL ===== pub mod model_categories; pub mod tts_models; pub mod w2l_templates; + +// ===== For Sqlite ===== pub mod news_stories; +pub mod tts_render_tasks; diff --git a/crates/_types/tokens/src/tokens/news_stories.rs b/crates/_types/tokens/src/tokens/news_stories.rs index aae770d235..956afb4220 100644 --- a/crates/_types/tokens/src/tokens/news_stories.rs +++ b/crates/_types/tokens/src/tokens/news_stories.rs @@ -3,7 +3,7 @@ use serde::Deserialize; use serde::Serialize; use std::fmt::Debug; -/// The primary key for "generic" inference jobs. +/// The primary key for news stories (Sqlite) #[derive(Clone, PartialEq, Eq, sqlx::Type, Debug, Serialize, Deserialize)] #[sqlx(transparent)] pub struct NewsStoryToken(pub String); diff --git a/crates/_types/tokens/src/tokens/tts_render_tasks.rs b/crates/_types/tokens/src/tokens/tts_render_tasks.rs new file mode 100644 index 0000000000..e6996a5a21 --- /dev/null +++ b/crates/_types/tokens/src/tokens/tts_render_tasks.rs @@ -0,0 +1,12 @@ +use crate::prefixes::EntityType; +use serde::Deserialize; +use serde::Serialize; +use std::fmt::Debug; + +/// The primary key for TTS render tasks (Sqlite / AiChatBotSidecar) +#[derive(Clone, PartialEq, Eq, sqlx::Type, Debug, Serialize, Deserialize)] +#[sqlx(transparent)] +pub struct TtsRenderTaskToken(pub String); + +impl_string_token!(TtsRenderTaskToken); +impl_crockford_generator!(TtsRenderTaskToken, 32usize, EntityType::TtsRenderTask, CrockfordMixed); diff --git a/crates/lib/fakeyou_client/src/fakeyou_api_client.rs b/crates/lib/fakeyou_client/src/fakeyou_api_client.rs index e207676efa..621256ee83 100644 --- a/crates/lib/fakeyou_client/src/fakeyou_api_client.rs +++ b/crates/lib/fakeyou_client/src/fakeyou_api_client.rs @@ -22,18 +22,19 @@ impl FakeYouApiClient { } // TODO: need to yield better, more "library"-appropriate errors. - pub async fn post_inference(&self, request: CreateTtsInferenceRequest<'_>) -> AnyhowResult { + pub async fn create_tts_inference(&self, request: CreateTtsInferenceRequest<'_>) -> AnyhowResult { let url = format!("https://{}/tts/inference", self.api_domain); let response = self.client .post(url) .header(AUTHORIZATION_HEADER, &self.api_token) .json(&request) + .send() .await? - .bytes() + .text() .await?; - let response = serde_json::from_str(response)?; + let response = serde_json::from_str(&response)?; Ok(response) } @@ -45,11 +46,12 @@ impl FakeYouApiClient { let response = self.client .get(url) .header(AUTHORIZATION_HEADER, &self.api_token) + .send() .await? - .bytes() + .text() .await?; - let response = serde_json::from_str(response)?; + let response = serde_json::from_str(&response)?; Ok(response) } diff --git a/crates/lib/sqlite_queries/src/queries/by_table/tts_render_targets/insert_tts_render_target.rs b/crates/lib/sqlite_queries/src/queries/by_table/tts_render_targets/insert_tts_render_target.rs index 2cea30af70..63dd018e57 100644 --- a/crates/lib/sqlite_queries/src/queries/by_table/tts_render_targets/insert_tts_render_target.rs +++ b/crates/lib/sqlite_queries/src/queries/by_table/tts_render_targets/insert_tts_render_target.rs @@ -5,6 +5,7 @@ use sqlx::SqlitePool; use enums::by_table::web_scraping_targets::scraping_status::ScrapingStatus; use tokens::tokens::news_stories::NewsStoryToken; use tokens::tokens::tts_models::TtsModelToken; +use tokens::tokens::tts_render_tasks::TtsRenderTaskToken; pub struct Args <'a> { // TODO: This will be multiple types in the future @@ -21,9 +22,12 @@ pub async fn insert_tts_render_target(args: Args<'_>) -> AnyhowResult<()> { let news_story_token = args.news_story_token.to_string(); let tts_voice_identifier= args.tts_voice_identifier.to_string(); + let tts_render_task_token = TtsRenderTaskToken::generate().to_string(); + let query = sqlx::query!( r#" INSERT INTO tts_render_targets( + token, story_type, story_token, tts_service, @@ -31,6 +35,7 @@ INSERT INTO tts_render_targets( full_text ) VALUES ( + ?, "news_story", ?, "fakeyou", @@ -38,6 +43,7 @@ VALUES ( ? ) "#, + tts_render_task_token, news_story_token, tts_voice_identifier, args.full_text diff --git a/crates/lib/sqlite_queries/src/queries/by_table/tts_render_targets/list_tts_render_targets.rs b/crates/lib/sqlite_queries/src/queries/by_table/tts_render_targets/list_tts_render_targets.rs new file mode 100644 index 0000000000..0ae476692b --- /dev/null +++ b/crates/lib/sqlite_queries/src/queries/by_table/tts_render_targets/list_tts_render_targets.rs @@ -0,0 +1,105 @@ +use enums::by_table::tts_render_targets::tts_render_status::TtsRenderStatus; +use errors::{anyhow, AnyhowResult}; +use sqlx::SqlitePool; +use tokens::tokens::tts_models::TtsModelToken; +use tokens::tokens::tts_render_tasks::TtsRenderTaskToken; + +pub struct TtsRenderTarget { + pub id: i64, + pub token: TtsRenderTaskToken, + + // Composite foreign key + pub story_type: String, + pub story_token: String, + + pub tts_service: String, + pub tts_voice_identifier: String, + + pub full_text: String, + pub maybe_result_url: Option, + pub maybe_result_relative_filesystem_location: Option, + + pub tts_render_status: TtsRenderStatus, + pub tts_render_attempts: i64, +} + +pub async fn list_tts_render_targets( + tts_render_status: TtsRenderStatus, + last_id: i64, + limit: i64, + sqlite_pool: &SqlitePool, +) -> AnyhowResult> { + + // NB: Sqlx doesn't support `WHERE ... IN (...)` "yet". :( + // https://github.com/launchbadge/sqlx/blob/6d0d7402c8a9cbea2676a1795e9fb50b0cf60c03/FAQ.md?plain=1#L73 + let tts_render_status = tts_render_status.to_str().to_string(); + + let query = sqlx::query_as!( + RawInternalTtsRenderTarget, + r#" +SELECT + id, + token as `token: tokens::tokens::tts_render_tasks::TtsRenderTaskToken`, + story_type, + story_token, + tts_service, + tts_voice_identifier, + full_text, + maybe_result_url, + maybe_result_relative_filesystem_location, + tts_render_status as `tts_render_status: enums::by_table::tts_render_targets::tts_render_status::TtsRenderStatus`, + tts_render_attempts +FROM tts_render_targets +WHERE + tts_render_status = ? + AND id > ? +ORDER BY id ASC +LIMIT ? + "#, + tts_render_status, + last_id, + limit, + ); + + let records = query.fetch_all(sqlite_pool) + .await?; + + let records = records.into_iter() + .map(|record : RawInternalTtsRenderTarget| { + TtsRenderTarget { + id: record.id, + token: record.token, + story_type: record.story_type, + story_token: record.story_token, + tts_service: record.tts_service, + tts_voice_identifier: record.tts_voice_identifier, + full_text: record.full_text, + maybe_result_url: record.maybe_result_url, + maybe_result_relative_filesystem_location: record.maybe_result_relative_filesystem_location, + tts_render_status: record.tts_render_status, + tts_render_attempts: record.tts_render_attempts, + } + }) + .collect::>(); + + Ok(records) +} + +struct RawInternalTtsRenderTarget { + pub id: i64, + pub token: TtsRenderTaskToken, + + // Composite foreign key + pub story_type: String, + pub story_token: String, + + pub tts_service: String, + pub tts_voice_identifier: String, + + pub full_text: String, + pub maybe_result_url: Option, + pub maybe_result_relative_filesystem_location: Option, + + pub tts_render_status: TtsRenderStatus, + pub tts_render_attempts: i64, +} diff --git a/crates/lib/sqlite_queries/src/queries/by_table/tts_render_targets/mod.rs b/crates/lib/sqlite_queries/src/queries/by_table/tts_render_targets/mod.rs index 7a5d875341..2a81687436 100644 --- a/crates/lib/sqlite_queries/src/queries/by_table/tts_render_targets/mod.rs +++ b/crates/lib/sqlite_queries/src/queries/by_table/tts_render_targets/mod.rs @@ -1 +1,4 @@ -pub mod insert_tts_render_target; \ No newline at end of file +pub mod insert_tts_render_target; +pub mod list_tts_render_targets; +pub mod update_tts_render_target_unsuccessfully_submitted; +pub mod update_tts_render_target_successfully_submitted; \ No newline at end of file diff --git a/crates/lib/sqlite_queries/src/queries/by_table/tts_render_targets/update_tts_render_target_successfully_submitted.rs b/crates/lib/sqlite_queries/src/queries/by_table/tts_render_targets/update_tts_render_target_successfully_submitted.rs new file mode 100644 index 0000000000..29de56bff6 --- /dev/null +++ b/crates/lib/sqlite_queries/src/queries/by_table/tts_render_targets/update_tts_render_target_successfully_submitted.rs @@ -0,0 +1,33 @@ +use enums::common::sqlite::awaitable_job_status::AwaitableJobStatus; +use errors::{anyhow, AnyhowResult}; +use sqlx::SqlitePool; +use tokens::tokens::news_stories::NewsStoryToken; +use tokens::tokens::tts_render_tasks::TtsRenderTaskToken; + +pub struct Args <'a> { + pub tts_render_task_token: &'a TtsRenderTaskToken, + pub sqlite_pool: &'a SqlitePool, +} + +pub async fn update_tts_render_target_successfully_submitted(args: Args<'_>) -> AnyhowResult<()> { + let query = sqlx::query!( + r#" +UPDATE tts_render_targets +SET + tts_render_status = "processing", + tts_render_attempts = tts_render_attempts + 1, + version = version + 1 +WHERE + token = ? + "#, + args.tts_render_task_token, + ); + + let query_result = query.execute(args.sqlite_pool) + .await; + + match query_result { + Ok(_) => Ok(()), + Err(err) => Err(anyhow!("error updating: {:?}", err)), + } +} diff --git a/crates/lib/sqlite_queries/src/queries/by_table/tts_render_targets/update_tts_render_target_unsuccessfully_submitted.rs b/crates/lib/sqlite_queries/src/queries/by_table/tts_render_targets/update_tts_render_target_unsuccessfully_submitted.rs new file mode 100644 index 0000000000..b2ae5b8897 --- /dev/null +++ b/crates/lib/sqlite_queries/src/queries/by_table/tts_render_targets/update_tts_render_target_unsuccessfully_submitted.rs @@ -0,0 +1,48 @@ +use enums::common::sqlite::awaitable_job_status::AwaitableJobStatus; +use errors::{anyhow, AnyhowResult}; +use sqlx::SqlitePool; +use enums::by_table::tts_render_targets::tts_render_status::TtsRenderStatus; +use tokens::tokens::news_stories::NewsStoryToken; +use tokens::tokens::tts_render_tasks::TtsRenderTaskToken; + +pub struct Args <'a> { + pub tts_render_task_token: &'a TtsRenderTaskToken, + pub tts_render_attempts: i64, + + pub sqlite_pool: &'a SqlitePool, +} + +pub async fn update_tts_render_target_unsuccessfully_submitted(args: Args<'_>) -> AnyhowResult<()> { + let mut tts_render_status = next_status(args.tts_render_attempts); + + let query = sqlx::query!( + r#" +UPDATE tts_render_targets +SET + tts_render_status = ?, + tts_render_attempts = ?, + version = version + 1 +WHERE + token = ? + "#, + tts_render_status, + args.tts_render_attempts, + args.tts_render_task_token, + ); + + let query_result = query.execute(args.sqlite_pool) + .await; + + match query_result { + Ok(_) => Ok(()), + Err(err) => Err(anyhow!("error updating: {:?}", err)), + } +} + +fn next_status(attempts: i64) -> TtsRenderStatus { + if attempts >= 2 { + TtsRenderStatus::PermanentlyFailed + } else { + TtsRenderStatus::Failed + } +} diff --git a/crates/programs/aichatbot-sidecar/Cargo.toml b/crates/programs/aichatbot-sidecar/Cargo.toml index c604948d64..c5ced174a4 100644 --- a/crates/programs/aichatbot-sidecar/Cargo.toml +++ b/crates/programs/aichatbot-sidecar/Cargo.toml @@ -19,6 +19,7 @@ actix_helpers = { path = "../../lib/actix_helpers" } concurrency = { path = "../../lib/concurrency" } enums = { path = "../../_types/enums" } errors = { path = "../../lib/errors" } +fakeyou_client = { path = "../../lib/fakeyou_client" } files = { path = "../../lib/files" } hashing = { path = "../../lib/hashing" } http_server_common = { path = "../../lib/http_server_common" } diff --git a/crates/programs/aichatbot-sidecar/_sql/migrations/2023-02-10-073239_tts_render_targets/up.sql b/crates/programs/aichatbot-sidecar/_sql/migrations/2023-02-10-073239_tts_render_targets/up.sql index 362e384d30..cb45cbe53a 100644 --- a/crates/programs/aichatbot-sidecar/_sql/migrations/2023-02-10-073239_tts_render_targets/up.sql +++ b/crates/programs/aichatbot-sidecar/_sql/migrations/2023-02-10-073239_tts_render_targets/up.sql @@ -7,6 +7,10 @@ CREATE TABLE tts_render_targets ( id INTEGER NOT NULL PRIMARY KEY AUTOINCREMENT, + -- Effective primary key. + -- Follows a single TTS audio through the phases of rendering + token TEXT NOT NULL UNIQUE, + -- Composite primary key for what this TTS render job belongs to (eg. news story) story_type TEXT NOT NULL, story_token TEXT NOT NULL, @@ -30,7 +34,7 @@ CREATE TABLE tts_render_targets ( -- For audio results that get downloaded, they live on the filesystem here -- This is relative to the runtime data directory. - maybe_filesystem_relative_location TEXT, + maybe_result_relative_filesystem_location TEXT, -- TTS render status: "new", "processing", "failed", "permanently_failed", "success", "skipped" tts_render_status TEXT DEFAULT "new" NOT NULL, diff --git a/crates/programs/aichatbot-sidecar/src/main.rs b/crates/programs/aichatbot-sidecar/src/main.rs index 33b1e49b49..ed79728537 100644 --- a/crates/programs/aichatbot-sidecar/src/main.rs +++ b/crates/programs/aichatbot-sidecar/src/main.rs @@ -35,9 +35,11 @@ use std::sync::Arc; use std::thread; use std::time::Duration; use tokio::runtime::{Builder, Runtime}; +use fakeyou_client::fakeyou_api_client::FakeYouApiClient; use web_scrapers::sites::cnn::cnn_article_scraper::cnn_article_scraper; use web_scrapers::sites::techcrunch::techcrunch_article_scraper::techcrunch_article_scraper; use web_scrapers::sites::theguardian::theguardian_scraper::theguardian_scraper_test; +use crate::workers::audio::fakeyou_audio_create::main_loop::fakeyou_audio_create_main_loop; //#[tokio::main] //pub async fn main() -> AnyhowResult<()> { @@ -87,15 +89,20 @@ pub async fn main() -> AnyhowResult<()> { let openai_client = Arc::new(Client::new() .with_api_key(startup_args.openai_secret_key.clone())); + let fakeyou_client = Arc::new(FakeYouApiClient::make_production_client( + &startup_args.fakeyou_api_token)); + let save_directory = SaveDirectory::new(&startup_args.save_directory); let database_url = easyenv::get_env_string_required("DATABASE_URL")?; + let pool = SqlitePoolOptions::new() .max_connections(5) .connect(&database_url).await?; let job_state = Arc::new(JobState { openai_client: openai_client.clone(), + fakeyou_client: fakeyou_client.clone(), save_directory: save_directory.clone(), sqlite_pool: pool, app_control_state: app_control_state_inner.clone(), @@ -124,6 +131,7 @@ pub async fn main() -> AnyhowResult<()> { let job_state4 = job_state.clone(); let job_state5 = job_state.clone(); let job_state6 = job_state.clone(); + let job_state7 = job_state.clone(); tokio_runtime.spawn(async { let _r = web_index_ingestion_main_loop(job_state2).await; @@ -145,6 +153,10 @@ pub async fn main() -> AnyhowResult<()> { let _r = news_story_audio_preprocessing_main_loop(job_state6).await; }); + tokio_runtime.spawn(async { + let _r = fakeyou_audio_create_main_loop(job_state7).await; + }); + // TODO: Final scheduling thread //tokio_runtime.spawn(async { // // TODO... diff --git a/crates/programs/aichatbot-sidecar/src/shared_state/job_state.rs b/crates/programs/aichatbot-sidecar/src/shared_state/job_state.rs index 719a61079d..4d6760fdff 100644 --- a/crates/programs/aichatbot-sidecar/src/shared_state/job_state.rs +++ b/crates/programs/aichatbot-sidecar/src/shared_state/job_state.rs @@ -1,12 +1,15 @@ use async_openai::Client; use crate::persistence::save_directory::SaveDirectory; use crate::shared_state::app_control_state::AppControlState; +use fakeyou_client::fakeyou_api_client::FakeYouApiClient; use sqlx::{Pool, Sqlite}; use std::sync::Arc; #[derive(Clone)] pub struct JobState { pub openai_client: Arc, + pub fakeyou_client: Arc, + pub save_directory: SaveDirectory, pub sqlite_pool: Pool, diff --git a/crates/programs/aichatbot-sidecar/src/startup_args.rs b/crates/programs/aichatbot-sidecar/src/startup_args.rs index 4044ef1ca4..606a442239 100644 --- a/crates/programs/aichatbot-sidecar/src/startup_args.rs +++ b/crates/programs/aichatbot-sidecar/src/startup_args.rs @@ -12,6 +12,7 @@ pub struct StartupArgs { pub save_directory: String, pub openai_secret_key: String, + pub fakeyou_api_token: String, } /// Read in environment and CLI args and output the consolidated startup state @@ -28,11 +29,14 @@ pub fn get_startup_args() -> AnyhowResult { let openai_secret_key = easyenv::get_env_string_required("OPENAI_SECRET_KEY")?; + let fakeyou_api_token = easyenv::get_env_string_required("FAKEYOU_API_TOKEN")?; + let save_directory = get_save_directory(&matches)?; Ok(StartupArgs { save_directory, openai_secret_key, + fakeyou_api_token, }) } diff --git a/crates/programs/aichatbot-sidecar/src/workers/audio/fakeyou_audio_create/main_loop.rs b/crates/programs/aichatbot-sidecar/src/workers/audio/fakeyou_audio_create/main_loop.rs new file mode 100644 index 0000000000..60b5e4ba26 --- /dev/null +++ b/crates/programs/aichatbot-sidecar/src/workers/audio/fakeyou_audio_create/main_loop.rs @@ -0,0 +1,103 @@ +use crate::shared_state::job_state::JobState; +use enums::by_table::tts_render_targets::tts_render_status::TtsRenderStatus; +use enums::by_table::web_scraping_targets::scraping_status::ScrapingStatus; +use errors::AnyhowResult; +use log::{debug, error, info}; +use sqlite_queries::queries::by_table::tts_render_targets::list_tts_render_targets::list_tts_render_targets; +use sqlite_queries::queries::by_table::web_scraping_targets::insert_web_scraping_target::{Args, insert_web_scraping_target}; +use sqlite_queries::queries::by_table::web_scraping_targets::list_web_scraping_targets::WebScrapingTarget as WebScrapingTargetRecord; +use sqlite_queries::queries::by_table::web_scraping_targets::list_web_scraping_targets::list_web_scraping_targets; +use sqlx::sqlite::SqlitePoolOptions; +use std::future::Future; +use std::sync::Arc; +use std::thread; +use std::time::Duration; +use strum::IntoEnumIterator; +use web_scrapers::payloads::web_scraping_result::ScrapedWebArticle; +use web_scrapers::payloads::web_scraping_target::WebScrapingTarget; +use web_scrapers::sites::cnn::cnn_indexer::{cnn_indexer, CnnFeed}; +use web_scrapers::sites::techcrunch::techcrunch_indexer::{techcrunch_indexer, TechcrunchFeed}; +use crate::workers::audio::fakeyou_audio_create::process_single_record::process_single_record; + +/// Get audio jobs and kick off FakeYou requests +pub async fn fakeyou_audio_create_main_loop(job_state: Arc) { + loop { + debug!("fakeyou_audio_create main loop"); + + while job_state.app_control_state.is_fakeyou_paused() { + thread::sleep(Duration::from_secs(5)); + } + + single_job_loop_iteration(&job_state).await; + + info!("fakeyou_audio_create loop finished; waiting..."); + thread::sleep(Duration::from_secs(60)); + } +} + +// NB: No failures at this level, because we don't want to prevent progress on a stuck feed. +async fn single_job_loop_iteration(job_state: &Arc) { + let statuses = vec![TtsRenderStatus::New, TtsRenderStatus::Failed]; + for status in statuses { + scrape_jobs_of_status(status, job_state).await; + } +} + +async fn scrape_jobs_of_status(status: TtsRenderStatus, job_state: &Arc) { + const BATCH_SIZE : i64 = 10; + + let mut last_id = 0; + let mut failure_count = 0; + + loop { + // NB: Protect sqlite from contention. + thread::sleep(Duration::from_millis(500)); + + debug!("fakeyou_audio_create querying {:?} targets from id > {} ...", &status, last_id); + + let query_result = list_tts_render_targets( + status, last_id, BATCH_SIZE, &job_state.sqlite_pool).await; + + let targets = match query_result { + Ok(targets) => targets, + Err(err) => { + error!("failure querying batch: {:?}", err); + + failure_count += 1; + + // NB: Don't starve progress. + if failure_count > 2 { + failure_count = 0; + last_id += 1; + } else if failure_count > 3 { + return; + } + + continue; + } + }; + + failure_count = 0; + + if targets.is_empty() { + return; // Done with batches. + } + + for target in targets { + debug!("Render audio: {}/{}", target.story_type, target.story_token); + + while job_state.app_control_state.is_fakeyou_paused() { + thread::sleep(Duration::from_secs(5)); + } + + let result = process_single_record(&target, job_state).await; + if let Err(err) = result { + error!("Error processing target: {:?}", err); + } + + last_id = target.id; + + thread::sleep(Duration::from_secs(3)); + } + } +} diff --git a/crates/programs/aichatbot-sidecar/src/workers/audio/fakeyou_audio_create/mod.rs b/crates/programs/aichatbot-sidecar/src/workers/audio/fakeyou_audio_create/mod.rs index e69de29bb2..c8e8378aeb 100644 --- a/crates/programs/aichatbot-sidecar/src/workers/audio/fakeyou_audio_create/mod.rs +++ b/crates/programs/aichatbot-sidecar/src/workers/audio/fakeyou_audio_create/mod.rs @@ -0,0 +1,2 @@ +pub mod main_loop; +pub mod process_single_record; \ No newline at end of file diff --git a/crates/programs/aichatbot-sidecar/src/workers/audio/fakeyou_audio_create/process_single_record.rs b/crates/programs/aichatbot-sidecar/src/workers/audio/fakeyou_audio_create/process_single_record.rs new file mode 100644 index 0000000000..1a62e9f781 --- /dev/null +++ b/crates/programs/aichatbot-sidecar/src/workers/audio/fakeyou_audio_create/process_single_record.rs @@ -0,0 +1,58 @@ +use crate::shared_state::job_state::JobState; +use errors::AnyhowResult; +use fakeyou_client::api::tts_inference::CreateTtsInferenceRequest; +use log::{error, info}; +use sqlite_queries::queries::by_table::tts_render_targets::list_tts_render_targets::TtsRenderTarget; +use sqlite_queries::queries::by_table::tts_render_targets::update_tts_render_target_successfully_submitted::Args as SuccessArgs; +use sqlite_queries::queries::by_table::tts_render_targets::update_tts_render_target_successfully_submitted::update_tts_render_target_successfully_submitted; +use sqlite_queries::queries::by_table::tts_render_targets::update_tts_render_target_unsuccessfully_submitted::Args as UnsuccessfulArgs; +use sqlite_queries::queries::by_table::tts_render_targets::update_tts_render_target_unsuccessfully_submitted::update_tts_render_target_unsuccessfully_submitted; +use std::sync::Arc; +use tokens::tokens::tts_models::TtsModelToken; + +pub async fn process_single_record(target: &TtsRenderTarget, job_state: &Arc) -> AnyhowResult<()> { + + info!("Posting FakeYou create TTS request..."); + + let tts_model_token = TtsModelToken::new_from_str(&target.tts_voice_identifier); + + let tts_render_attempts = target.tts_render_attempts + 1; + + let result = job_state + .fakeyou_client + .create_tts_inference(CreateTtsInferenceRequest { + uuid_idempotency_token: "", + tts_model_token: &tts_model_token, + inference_text: &target.full_text, + }).await; + + match result { + Err(err) => { + error!("problem submitting to FakeYou: {:}", err); + + update_tts_render_target_unsuccessfully_submitted(UnsuccessfulArgs { + tts_render_task_token: &target.token, + tts_render_attempts, + sqlite_pool: &job_state.sqlite_pool, + }).await?; + }, + Ok(res) => { + if !res.success { + error!("unknown problem in submitting to FakeYou"); + + update_tts_render_target_unsuccessfully_submitted(UnsuccessfulArgs { + tts_render_task_token: &target.token, + tts_render_attempts, + sqlite_pool: &job_state.sqlite_pool, + }).await?; + } else { + update_tts_render_target_successfully_submitted(SuccessArgs { + tts_render_task_token: &target.token, + sqlite_pool: &job_state.sqlite_pool, + }).await?; + } + }, + } + + Ok(()) +}