diff --git a/Cargo.lock b/Cargo.lock index 39b391b404..d4eb4ce996 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -739,6 +739,7 @@ dependencies = [ "actix-web-actors", "actix_helpers", "async-openai", + "chrono", "clap", "concurrency", "dotenv", diff --git a/crates/lib/sqlite_queries/src/queries/by_table/mod.rs b/crates/lib/sqlite_queries/src/queries/by_table/mod.rs index 688bbc1aec..4d06a0a166 100644 --- a/crates/lib/sqlite_queries/src/queries/by_table/mod.rs +++ b/crates/lib/sqlite_queries/src/queries/by_table/mod.rs @@ -1,3 +1,4 @@ +pub mod news_stories; pub mod news_story_productions; pub mod tts_render_tasks; pub mod web_scraping_targets; \ No newline at end of file diff --git a/crates/lib/sqlite_queries/src/queries/by_table/news_stories/insert_news_story.rs b/crates/lib/sqlite_queries/src/queries/by_table/news_stories/insert_news_story.rs new file mode 100644 index 0000000000..2a97d91c47 --- /dev/null +++ b/crates/lib/sqlite_queries/src/queries/by_table/news_stories/insert_news_story.rs @@ -0,0 +1,50 @@ +use chrono::{DateTime, Utc}; +use errors::{anyhow, AnyhowResult}; +use sqlx::SqlitePool; +use tokens::tokens::news_stories::NewsStoryToken; + +pub struct Args <'a> { + pub news_story_token: &'a NewsStoryToken, + + pub original_news_canonical_url: &'a str, + + pub replayable_until: DateTime, + + pub sqlite_pool: &'a SqlitePool, +} + +pub async fn insert_news_story(args: Args<'_>) -> AnyhowResult<()> { + let news_story_token = args.news_story_token.as_str(); + + let query = sqlx::query!( + r#" +INSERT INTO news_stories ( + news_story_token, + original_news_canonical_url, + replayable_until, + is_playable +) +VALUES ( + ?, + ?, + ?, + true +) + "#, + news_story_token, + args.original_news_canonical_url, + args.replayable_until + ); + + let query_result = query.execute(args.sqlite_pool) + .await; + + let _record_id = match query_result { + Ok(res) => res.last_insert_rowid(), + Err(err) => { + return Err(anyhow!("error inserting: {:?}", err)); + } + }; + + Ok(()) +} diff --git a/crates/lib/sqlite_queries/src/queries/by_table/news_stories/mod.rs b/crates/lib/sqlite_queries/src/queries/by_table/news_stories/mod.rs new file mode 100644 index 0000000000..92d60bf507 --- /dev/null +++ b/crates/lib/sqlite_queries/src/queries/by_table/news_stories/mod.rs @@ -0,0 +1 @@ +pub mod insert_news_story; \ No newline at end of file diff --git a/crates/lib/sqlite_queries/src/queries/by_table/news_story_productions/list/list_news_story_productions_ready_for_debut.rs b/crates/lib/sqlite_queries/src/queries/by_table/news_story_productions/list/list_news_story_productions_ready_for_debut.rs new file mode 100644 index 0000000000..dd3ad8fe4d --- /dev/null +++ b/crates/lib/sqlite_queries/src/queries/by_table/news_story_productions/list/list_news_story_productions_ready_for_debut.rs @@ -0,0 +1,41 @@ +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_ready_for_debut( + 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 + overall_production_status = "processing" + AND llm_rendition_status = "done" + AND audio_generation_status = "done" + 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 a9bd87a24c..29461686b9 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 @@ -2,4 +2,5 @@ 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; +pub mod list_news_story_productions_ready_for_debut; pub mod news_story_production_item; 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 58f4a2ddd9..980bc2c0fb 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,4 +1,5 @@ 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_post_production_complete; 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 index 07a36d3deb..c3cf737455 100644 --- 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 @@ -10,20 +10,18 @@ pub struct Args <'a> { } 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 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, ); diff --git a/crates/lib/sqlite_queries/src/queries/by_table/news_story_productions/update/update_news_story_post_production_complete.rs b/crates/lib/sqlite_queries/src/queries/by_table/news_story_productions/update/update_news_story_post_production_complete.rs new file mode 100644 index 0000000000..1e8f5c19c5 --- /dev/null +++ b/crates/lib/sqlite_queries/src/queries/by_table/news_story_productions/update/update_news_story_post_production_complete.rs @@ -0,0 +1,34 @@ +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_post_production_complete(args: Args<'_>) -> AnyhowResult<()> { + let mut overall_production_status = AwaitableJobStatus::Done.to_str(); + + let query = sqlx::query!( + r#" +UPDATE news_story_productions +SET + overall_production_status = ?, + version = version + 1 +WHERE + news_story_token = ? + "#, + overall_production_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/programs/aichatbot-sidecar/Cargo.toml b/crates/programs/aichatbot-sidecar/Cargo.toml index 67ed3a958d..29e2d25d70 100644 --- a/crates/programs/aichatbot-sidecar/Cargo.toml +++ b/crates/programs/aichatbot-sidecar/Cargo.toml @@ -29,6 +29,7 @@ tokens = { path = "../../_types/tokens" } web_scrapers = { path = "../../lib/web_scrapers" } # External +chrono = { version = "0.4.19", features = ["serde"] } clap = "2.34.0" # TODO: This is really out of date, but has conflicts when updated dotenv = "0.15.0" easyenv = "0.0.6" diff --git a/crates/programs/aichatbot-sidecar/src/main.rs b/crates/programs/aichatbot-sidecar/src/main.rs index 450a004bb6..220e0adfb4 100644 --- a/crates/programs/aichatbot-sidecar/src/main.rs +++ b/crates/programs/aichatbot-sidecar/src/main.rs @@ -42,6 +42,7 @@ use web_scrapers::sites::theguardian::theguardian_scraper::theguardian_scraper_t 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; +use crate::workers::news_stories::news_story_post_production_finalization::main_loop::news_story_post_production_finalization_main_loop; //#[tokio::main] //pub async fn main() -> AnyhowResult<()> { @@ -124,7 +125,7 @@ pub async fn main() -> AnyhowResult<()> { // Thus, we launch everything else into its own thread. thread::spawn(move || { let tokio_runtime = Builder::new_multi_thread() - .worker_threads(8) + .worker_threads(16) .thread_name("tokio-worker") .thread_stack_size(3 * 1024 * 1024) .enable_time() @@ -140,6 +141,7 @@ pub async fn main() -> AnyhowResult<()> { let job_state7 = job_state.clone(); let job_state8 = job_state.clone(); let job_state9 = job_state.clone(); + let job_state10 = job_state.clone(); tokio_runtime.spawn(async { let _r = web_index_ingestion_main_loop(job_state2).await; @@ -173,6 +175,9 @@ pub async fn main() -> AnyhowResult<()> { let _r = news_story_audio_final_verification_main_loop(job_state9).await; }); + tokio_runtime.spawn(async { + let _r = news_story_post_production_finalization_main_loop(job_state10).await; + }); // TODO: Final scheduling thread //tokio_runtime.spawn(async { 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 index 0adbf99a1a..b39c9ea964 100644 --- 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 @@ -97,7 +97,7 @@ async fn scrape_jobs_of_status(status: TtsRenderStatus, job_state: &Arc) { last_id = target.id; - thread::sleep(Duration::from_secs(3)); + thread::sleep(Duration::from_secs(1)); } } } diff --git a/crates/programs/aichatbot-sidecar/src/workers/news_stories/mod.rs b/crates/programs/aichatbot-sidecar/src/workers/news_stories/mod.rs index 3e35082b1f..d49f2f4b7c 100644 --- a/crates/programs/aichatbot-sidecar/src/workers/news_stories/mod.rs +++ b/crates/programs/aichatbot-sidecar/src/workers/news_stories/mod.rs @@ -2,3 +2,4 @@ pub mod news_story_audio_final_verification; pub mod news_story_audio_preprocessing; pub mod news_story_greenlighting; pub mod news_story_llm_rendition; +pub mod news_story_post_production_finalization; 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 index 31d777c35c..144088e3fb 100644 --- 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 @@ -71,7 +71,7 @@ async fn query_all_batches(job_state: &Arc) { last_id = target.id; - thread::sleep(Duration::from_secs(5)); // NB: We probably (?) need to throttle API calls to OpenAI + thread::sleep(Duration::from_millis(100)); } } } diff --git a/crates/programs/aichatbot-sidecar/src/workers/news_stories/news_story_greenlighting/main_loop.rs b/crates/programs/aichatbot-sidecar/src/workers/news_stories/news_story_greenlighting/main_loop.rs index f16b51c74c..b0d6d38e4e 100644 --- a/crates/programs/aichatbot-sidecar/src/workers/news_stories/news_story_greenlighting/main_loop.rs +++ b/crates/programs/aichatbot-sidecar/src/workers/news_stories/news_story_greenlighting/main_loop.rs @@ -67,7 +67,7 @@ async fn query_and_process_all_batches(job_state: &Arc) { } last_id = target.id; - thread::sleep(Duration::from_secs(1)); + thread::sleep(Duration::from_millis(100)); } } } diff --git a/crates/programs/aichatbot-sidecar/src/workers/news_stories/news_story_post_production_finalization/main_loop.rs b/crates/programs/aichatbot-sidecar/src/workers/news_stories/news_story_post_production_finalization/main_loop.rs new file mode 100644 index 0000000000..9cfde42ddc --- /dev/null +++ b/crates/programs/aichatbot-sidecar/src/workers/news_stories/news_story_post_production_finalization/main_loop.rs @@ -0,0 +1,78 @@ +use crate::shared_state::job_state::JobState; +use crate::workers::news_stories::news_story_post_production_finalization::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 sqlite_queries::queries::by_table::news_story_productions::list::list_news_story_productions_ready_for_debut::list_news_story_productions_ready_for_debut; +use std::sync::Arc; +use std::thread; +use std::time::Duration; + +/// Find in-production stories with all tasks done, then promote them to stories +pub async fn news_story_post_production_finalization_main_loop(job_state: Arc) { + loop { + debug!("news_story_post_production main loop"); + + query_all_batches(&job_state).await; + + debug!("news_story_post_production 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!("news_story_post_production querying targets from id > {} ...", last_id); + + let query_result = list_news_story_productions_ready_for_debut( + 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!("Final post production 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_millis(200)); + } + } +} diff --git a/crates/programs/aichatbot-sidecar/src/workers/news_stories/news_story_post_production_finalization/mod.rs b/crates/programs/aichatbot-sidecar/src/workers/news_stories/news_story_post_production_finalization/mod.rs new file mode 100644 index 0000000000..69b2dcb3e8 --- /dev/null +++ b/crates/programs/aichatbot-sidecar/src/workers/news_stories/news_story_post_production_finalization/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_post_production_finalization/process_single_item.rs b/crates/programs/aichatbot-sidecar/src/workers/news_stories/news_story_post_production_finalization/process_single_item.rs new file mode 100644 index 0000000000..ab9a6bffb2 --- /dev/null +++ b/crates/programs/aichatbot-sidecar/src/workers/news_stories/news_story_post_production_finalization/process_single_item.rs @@ -0,0 +1,39 @@ +use chrono::Duration; +use chrono::Utc; +use crate::shared_state::job_state::JobState; +use enums::by_table::tts_render_tasks::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_stories::insert_news_story::Args as InsertArgs; +use sqlite_queries::queries::by_table::news_stories::insert_news_story::insert_news_story; +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_post_production_complete::Args as UpdateArgs; +use sqlite_queries::queries::by_table::news_story_productions::update::update_news_story_post_production_complete::update_news_story_post_production_complete; +use std::ops::Add; +use std::sync::Arc; + +/// Stop playing stories after 24 hours elapses +static STORY_FRESHNESS_THRESHOLD : Lazy = Lazy::new(|| { + Duration::days(1) +}); + +pub async fn process_single_item(target: &NewsStoryProductionItem, job_state: &Arc) -> AnyhowResult<()> { + + let replayable_until = Utc::now().add(*STORY_FRESHNESS_THRESHOLD); + + insert_news_story(InsertArgs { + news_story_token: &target.news_story_token, + original_news_canonical_url: &target.original_news_canonical_url, + replayable_until, + sqlite_pool: &job_state.sqlite_pool, + }).await?; + + update_news_story_post_production_complete(UpdateArgs { + news_story_token: &target.news_story_token, + sqlite_pool: &job_state.sqlite_pool, + }).await?; + + Ok(()) +}