preproduction story finalization

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<Vec<NewsStoryProductionItem>> {
+
+  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)
+}

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;

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;

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)),
+  }
+}
+

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<Vec<TtsRenderTarget>> {
+
+  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)
+}

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;

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,
 

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 {

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<JobState>) {
+  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<JobState>) {
+  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
+    }
+  }
+}

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;

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<JobState>) -> 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(())
+}