diff --git a/crates/lib/sqlite_queries/src/queries/by_table/news_story_productions/list/list_news_story_productions_awaiting_audio_final_verification.rs b/crates/lib/sqlite_queries/src/queries/by_table/news_story_productions/list/list_news_story_productions_awaiting_audio_final_verification.rs new file mode 100644 index 0000000000..ebf5e34a98 --- /dev/null +++ b/crates/lib/sqlite_queries/src/queries/by_table/news_story_productions/list/list_news_story_productions_awaiting_audio_final_verification.rs @@ -0,0 +1,39 @@ +use crate::queries::by_table::news_story_productions::list::news_story_production_item::NewsStoryProductionItem; +use enums::common::sqlite::awaitable_job_status::AwaitableJobStatus; +use errors::{anyhow, AnyhowResult}; +use sqlx::SqlitePool; +use tokens::tokens::news_stories::NewsStoryToken; + +pub async fn list_news_story_productions_awaiting_audio_final_verification( + last_id: i64, + limit: i64, + sqlite_pool: &SqlitePool, +) -> AnyhowResult> { + + let query = sqlx::query_as!( + NewsStoryProductionItem, + r#" +SELECT + id, + news_story_token as `news_story_token: tokens::tokens::news_stories::NewsStoryToken`, + original_news_canonical_url, + overall_production_status as `overall_production_status: enums::common::sqlite::awaitable_job_status::AwaitableJobStatus`, + llm_rendition_status as `llm_rendition_status: enums::common::sqlite::awaitable_job_status::AwaitableJobStatus`, + llm_rendition_attempts, + audio_generation_status as `audio_generation_status: enums::common::sqlite::awaitable_job_status::AwaitableJobStatus` +FROM news_story_productions +WHERE + audio_generation_status = "processing" + AND id > ? +ORDER BY id ASC +LIMIT ? + "#, + last_id, + limit, + ); + + let records = query.fetch_all(sqlite_pool) + .await?; + + Ok(records) +} diff --git a/crates/lib/sqlite_queries/src/queries/by_table/news_story_productions/list/mod.rs b/crates/lib/sqlite_queries/src/queries/by_table/news_story_productions/list/mod.rs index ae6e18e530..a9bd87a24c 100644 --- a/crates/lib/sqlite_queries/src/queries/by_table/news_story_productions/list/mod.rs +++ b/crates/lib/sqlite_queries/src/queries/by_table/news_story_productions/list/mod.rs @@ -1,3 +1,4 @@ +pub mod list_news_story_productions_awaiting_audio_final_verification; pub mod list_news_story_productions_awaiting_audio_generation; pub mod list_news_story_productions_awaiting_greenlight; pub mod list_news_story_productions_awaiting_llm_rendition; diff --git a/crates/lib/sqlite_queries/src/queries/by_table/news_story_productions/update/mod.rs b/crates/lib/sqlite_queries/src/queries/by_table/news_story_productions/update/mod.rs index 0245e7c8a2..58f4a2ddd9 100644 --- a/crates/lib/sqlite_queries/src/queries/by_table/news_story_productions/update/mod.rs +++ b/crates/lib/sqlite_queries/src/queries/by_table/news_story_productions/update/mod.rs @@ -1,3 +1,4 @@ +pub mod update_news_story_audio_finalized; pub mod update_news_story_audio_preprocessing_status; pub mod update_news_story_llm_rendition_status; pub mod update_news_story_production_greenlit_status; diff --git a/crates/lib/sqlite_queries/src/queries/by_table/news_story_productions/update/update_news_story_audio_finalized.rs b/crates/lib/sqlite_queries/src/queries/by_table/news_story_productions/update/update_news_story_audio_finalized.rs new file mode 100644 index 0000000000..07a36d3deb --- /dev/null +++ b/crates/lib/sqlite_queries/src/queries/by_table/news_story_productions/update/update_news_story_audio_finalized.rs @@ -0,0 +1,39 @@ +use enums::common::sqlite::awaitable_job_status::AwaitableJobStatus; +use errors::{anyhow, AnyhowResult}; +use sqlx::SqlitePool; +use tokens::tokens::news_stories::NewsStoryToken; + +pub struct Args <'a> { + pub news_story_token: &'a NewsStoryToken, + + pub sqlite_pool: &'a SqlitePool, +} + +pub async fn update_news_story_audio_finalized(args: Args<'_>) -> AnyhowResult<()> { + let mut overall_production_status = AwaitableJobStatus::Done.to_str(); // TODO: Changes if we include imagery. + let mut audio_generation_status = AwaitableJobStatus::Done.to_str(); + + let query = sqlx::query!( + r#" +UPDATE news_story_productions +SET + overall_production_status = ?, + audio_generation_status = ?, + version = version + 1 +WHERE + news_story_token = ? + "#, + overall_production_status, + audio_generation_status, + args.news_story_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/list/list_tts_render_targets_for_story_token.rs b/crates/lib/sqlite_queries/src/queries/by_table/tts_render_targets/list/list_tts_render_targets_for_story_token.rs new file mode 100644 index 0000000000..3da15dcdef --- /dev/null +++ b/crates/lib/sqlite_queries/src/queries/by_table/tts_render_targets/list/list_tts_render_targets_for_story_token.rs @@ -0,0 +1,48 @@ +use crate::queries::by_table::tts_render_targets::list::tts_render_target::TtsRenderTarget; +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; + +// TODO: need to move download status to a different column as it's a different job + +pub async fn list_tts_render_targets_for_story_token( + story_type: &str, + story_token: &str, + sqlite_pool: &SqlitePool, +) -> AnyhowResult> { + + let query = sqlx::query_as!( + TtsRenderTarget, + r#" +SELECT + id, + token as `token: tokens::tokens::tts_render_tasks::TtsRenderTaskToken`, + story_type, + story_token, + sequence_order, + tts_service, + tts_voice_identifier, + full_text, + maybe_inference_job_token, + maybe_result_token, + 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 + story_type = ? + AND story_token = ? +ORDER BY sequence_order ASC + "#, + story_type, + story_token, + ); + + let records = query.fetch_all(sqlite_pool) + .await?; + + Ok(records) +} diff --git a/crates/lib/sqlite_queries/src/queries/by_table/tts_render_targets/list/mod.rs b/crates/lib/sqlite_queries/src/queries/by_table/tts_render_targets/list/mod.rs index a5d3f9dd0e..9d1d5ba1af 100644 --- a/crates/lib/sqlite_queries/src/queries/by_table/tts_render_targets/list/mod.rs +++ b/crates/lib/sqlite_queries/src/queries/by_table/tts_render_targets/list/mod.rs @@ -1,3 +1,4 @@ -pub mod list_tts_render_targets_awaiting_render; pub mod list_tts_render_targets_awaiting_download; +pub mod list_tts_render_targets_awaiting_render; +pub mod list_tts_render_targets_for_story_token; pub mod tts_render_target; 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 1e6224b2e0..138e764586 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 @@ -16,8 +16,12 @@ CREATE TABLE tts_render_targets ( story_token TEXT NOT NULL, -- For each "sequence" of audio, this is the order. + -- This is 1-indexed sequence_order INTEGER NOT NULL, + -- The total number of audio files in the entire speech. + sequence_length INTEGER NOT NULL, + -- If the source came from somewhere, this is it -- maybe_canonical_url TEXT NOT NULL UNIQUE, diff --git a/crates/programs/aichatbot-sidecar/src/main.rs b/crates/programs/aichatbot-sidecar/src/main.rs index a19ccef03f..450a004bb6 100644 --- a/crates/programs/aichatbot-sidecar/src/main.rs +++ b/crates/programs/aichatbot-sidecar/src/main.rs @@ -41,6 +41,7 @@ use web_scrapers::sites::techcrunch::techcrunch_article_scraper::techcrunch_arti use web_scrapers::sites::theguardian::theguardian_scraper::theguardian_scraper_test; use crate::workers::audio::fakeyou_audio_create::main_loop::fakeyou_audio_create_main_loop; use crate::workers::audio::fakeyou_audio_download::main_loop::fakeyou_audio_download_main_loop; +use crate::workers::news_stories::news_story_audio_final_verification::main_loop::news_story_audio_final_verification_main_loop; //#[tokio::main] //pub async fn main() -> AnyhowResult<()> { @@ -138,6 +139,7 @@ pub async fn main() -> AnyhowResult<()> { let job_state6 = job_state.clone(); let job_state7 = job_state.clone(); let job_state8 = job_state.clone(); + let job_state9 = job_state.clone(); tokio_runtime.spawn(async { let _r = web_index_ingestion_main_loop(job_state2).await; @@ -167,6 +169,10 @@ pub async fn main() -> AnyhowResult<()> { let _r = fakeyou_audio_download_main_loop(job_state8).await; }); + tokio_runtime.spawn(async { + let _r = news_story_audio_final_verification_main_loop(job_state9).await; + }); + // TODO: Final scheduling thread //tokio_runtime.spawn(async { diff --git a/crates/programs/aichatbot-sidecar/src/workers/news_stories/news_story_audio_final_verification/main_loop.rs b/crates/programs/aichatbot-sidecar/src/workers/news_stories/news_story_audio_final_verification/main_loop.rs new file mode 100644 index 0000000000..31d777c35c --- /dev/null +++ b/crates/programs/aichatbot-sidecar/src/workers/news_stories/news_story_audio_final_verification/main_loop.rs @@ -0,0 +1,77 @@ +use crate::shared_state::job_state::JobState; +use crate::workers::news_stories::news_story_audio_final_verification::process_single_item::process_single_item; +use enums::by_table::web_rendition_targets::rendition_status::RenditionStatus; +use enums::common::sqlite::awaitable_job_status::AwaitableJobStatus; +use log::{debug, error, info}; +use sqlite_queries::queries::by_table::news_story_productions::list::list_news_story_productions_awaiting_audio_final_verification::list_news_story_productions_awaiting_audio_final_verification; +use sqlite_queries::queries::by_table::news_story_productions::list::list_news_story_productions_awaiting_llm_rendition::list_news_story_productions_awaiting_llm_rendition; +use std::sync::Arc; +use std::thread; +use std::time::Duration; + +/// Verify all audio files have been requested, generated, and downloaded. +pub async fn news_story_audio_final_verification_main_loop(job_state: Arc) { + loop { + debug!("news_story_audio_final_verification main loop"); + + query_all_batches(&job_state).await; + + debug!("news_story_audio_final_verification loop finished; waiting..."); + thread::sleep(Duration::from_secs(1)); + } +} + +async fn query_all_batches(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!("audio_final_verification querying targets from id > {} ...", last_id); + + let query_result = list_news_story_productions_awaiting_audio_final_verification( + 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!("Audio verification for target: {:?}", target.original_news_canonical_url); + + let result = process_single_item(&target, job_state).await; + if let Err(err) = result { + error!("Error processing target: {:?}", err); + } + + last_id = target.id; + + thread::sleep(Duration::from_secs(5)); // NB: We probably (?) need to throttle API calls to OpenAI + } + } +} diff --git a/crates/programs/aichatbot-sidecar/src/workers/news_stories/news_story_audio_final_verification/mod.rs b/crates/programs/aichatbot-sidecar/src/workers/news_stories/news_story_audio_final_verification/mod.rs index e69de29bb2..69b2dcb3e8 100644 --- a/crates/programs/aichatbot-sidecar/src/workers/news_stories/news_story_audio_final_verification/mod.rs +++ b/crates/programs/aichatbot-sidecar/src/workers/news_stories/news_story_audio_final_verification/mod.rs @@ -0,0 +1,2 @@ +pub mod main_loop; +pub mod process_single_item; \ No newline at end of file diff --git a/crates/programs/aichatbot-sidecar/src/workers/news_stories/news_story_audio_final_verification/process_single_item.rs b/crates/programs/aichatbot-sidecar/src/workers/news_stories/news_story_audio_final_verification/process_single_item.rs new file mode 100644 index 0000000000..bfc7afa849 --- /dev/null +++ b/crates/programs/aichatbot-sidecar/src/workers/news_stories/news_story_audio_final_verification/process_single_item.rs @@ -0,0 +1,41 @@ +use crate::shared_state::job_state::JobState; +use enums::by_table::tts_render_targets::tts_render_status::TtsRenderStatus; +use errors::AnyhowResult; +use log::{debug, info}; +use once_cell::sync::Lazy; +use regex::Regex; +use sqlite_queries::queries::by_table::news_story_productions::list::news_story_production_item::NewsStoryProductionItem; +use sqlite_queries::queries::by_table::news_story_productions::update::update_news_story_audio_finalized::{Args, update_news_story_audio_finalized}; +use sqlite_queries::queries::by_table::tts_render_targets::list::list_tts_render_targets_for_story_token::list_tts_render_targets_for_story_token; +use std::sync::Arc; + +pub async fn process_single_item(target: &NewsStoryProductionItem, job_state: &Arc) -> AnyhowResult<()> { + + let story_type = "news_story"; + let story_token = target.news_story_token.to_string(); + + let tts_render_targets = list_tts_render_targets_for_story_token( + story_type, + &story_token, + &job_state.sqlite_pool).await?; + + let complete_count = tts_render_targets.iter() + .filter(|target| target.tts_render_status == TtsRenderStatus::Success) + .count(); + + // TODO: Include new column "tts_render_targets.sequence_length" to verify. Math.max() it. + // This is to prevent a race condition. + let total_count = tts_render_targets.len(); + + if complete_count != total_count { + debug!("Audio not yet rendered for story: {:?}", target.news_story_token); + return Ok(()) + } + + update_news_story_audio_finalized(Args { + news_story_token: &target.news_story_token, + sqlite_pool: &job_state.sqlite_pool, + }).await?; + + Ok(()) +}