worker threads

Cargo.lock

@@ -772,6 +772,7 @@ dependencies = [
  "serde_yaml",
  "sqlite_queries",
  "sqlx",
+ "strum",
  "tokens",
  "tokio 1.25.0",
  "web_scrapers",
@@ -8189,6 +8190,7 @@ dependencies = [
  "serde",
  "serde_derive",
  "serde_json",
+ "strum",
  "toml",
 ]
 

crates/lib/web_scrapers/Cargo.toml

@@ -22,6 +22,7 @@ bytes = "1.3"
 chrono = { version = "0.4.19", features = ["serde"] }
 log = "0.4.14"
 once_cell = "1.9.0"
+strum = { version = "0.24", features = ["derive"] } # Fancy enum iteration, etc.
 
 # Serialization
 serde = "1.0.125"

crates/lib/web_scrapers/src/lib.rs

@@ -19,6 +19,7 @@
 #![allow(non_snake_case)]
 
 #[macro_use] extern crate serde_derive;
+#[macro_use] extern crate strum;
 
 pub mod payloads;
 pub mod sites;

crates/lib/web_scrapers/src/sites/mod.rs

@@ -4,4 +4,5 @@ pub mod reddit;
 pub mod reuters;
 pub mod slashdot;
 pub mod techcrunch;
-pub mod theguardian;
+pub mod theguardian;
+pub mod twitter;

crates/lib/web_scrapers/src/sites/techcrunch/mod.rs

@@ -1,2 +1,2 @@
-pub mod techcrunch_scraper;
+pub mod techcrunch_indexer;
 pub mod techcrunch_article_scraper;

crates/lib/web_scrapers/src/sites/techcrunch/techcrunch_indexer.rs

@@ -1,20 +1,37 @@
-use std::ops::Deref;
 use bytes::Bytes;
 use crate::payloads::web_scraping_target::WebScrapingTarget;
 use enums::by_table::web_scraping_targets::web_content_type::WebContentType;
 use errors::AnyhowResult;
 use log::warn;
 use rss::Channel;
+use std::ops::Deref;
 
-// NB: Contains 20 items
+// NB: Feed contains ~20 items.
 const MAIN_RSS_FEED : &'static str = "https://techcrunch.com/feed/";
 
 const VENTURE_FEED : &'static str = "https://techcrunch.com/category/venture/feed/";
 
 const STARTUPS_FEED : &'static str = "https://techcrunch.com/category/startups/feed/";
 
-pub async fn techcrunch_scraper_test() -> AnyhowResult<Vec<WebScrapingTarget>> {
-  let content = reqwest::get(MAIN_RSS_FEED)
+#[derive(Copy, Clone, Debug, EnumIter, EnumCount)]
+pub enum TechcrunchFeed {
+  Main,
+  Venture,
+  Startups,
+}
+
+impl TechcrunchFeed {
+  fn url(&self) -> &'static str {
+    match self {
+      TechcrunchFeed::Main => MAIN_RSS_FEED,
+      TechcrunchFeed::Venture => VENTURE_FEED,
+      TechcrunchFeed::Startups => STARTUPS_FEED,
+    }
+  }
+}
+
+pub async fn techcrunch_indexer(feed: TechcrunchFeed) -> AnyhowResult<Vec<WebScrapingTarget>> {
+  let content = reqwest::get(feed.url())
       .await?
       .bytes()
       .await?;

crates/programs/aichatbot-sidecar/Cargo.toml

@@ -27,10 +27,11 @@ web_scrapers = { path = "../../lib/web_scrapers" }
 
 # External
 clap = "2.34.0" # TODO: This is really out of date, but has conflicts when updated
+dotenv = "0.15.0"
 easyenv = "0.0.6"
 log = "0.4.14"
-dotenv = "0.15.0"
 path-absolutize = "3.0.14" # Absolute paths have not stabilized in Rust yet
+strum = { version = "0.24", features = ["derive"] } # Fancy enum iteration, etc.
 
 # SQlite Database
 # "offline" supports typesafe builds without a live database

crates/programs/aichatbot-sidecar/src/ingestion/mod.rs

@@ -1 +0,0 @@
-pub mod ingest_url_scrape_and_save;

crates/programs/aichatbot-sidecar/src/jobs/mod.rs

@@ -1,4 +0,0 @@
-pub mod fakeyou;
-pub mod openai;
-pub mod reddit;
-pub mod twitter;

crates/programs/aichatbot-sidecar/src/main.rs

@@ -3,13 +3,12 @@
 #![cfg_attr(not(debug_assertions), windows_subsystem = "windows")] // hide console window on Windows in release
 
 pub mod gui;
-pub mod ingestion;
-pub mod jobs;
 pub mod main_loop;
 pub mod persistence;
 pub mod shared_state;
 pub mod startup_args;
 pub mod web_server;
+pub mod workers;
 
 #[macro_use] extern crate serde_derive;
 
@@ -34,12 +33,13 @@ use sqlite_queries::queries::by_table::web_scraping_targets::insert_web_scraping
 use web_scrapers::sites::cnn::cnn_article_scraper::cnn_article_scraper;
 use web_scrapers::sites::cnn::cnn_indexer::cnn_scraper_test;
 use web_scrapers::sites::techcrunch::techcrunch_article_scraper::techcrunch_article_scraper;
-use web_scrapers::sites::techcrunch::techcrunch_scraper::techcrunch_scraper_test;
 use web_scrapers::sites::theguardian::theguardian_scraper::theguardian_scraper_test;
-use crate::ingestion::ingest_url_scrape_and_save::ingest_url_scrape_and_save;
+use workers::web_content_scraping::ingest_url_scrape_and_save::ingest_url_scrape_and_save;
+use crate::shared_state::job_state::JobState;
+use crate::workers::web_index_ingestion::main_loop::web_index_ingestion_main_loop;
 
 #[tokio::main]
-pub async fn main() -> AnyhowResult<()> {
+pub async fn main2() -> AnyhowResult<()> {
   let database_url = easyenv::get_env_string_required("DATABASE_URL")?;
   let pool = SqlitePoolOptions::new()
       .max_connections(5)
@@ -85,7 +85,7 @@ pub async fn main() -> AnyhowResult<()> {
 }
 
 #[actix_web::main]
-pub async fn main2() -> AnyhowResult<()> {
+pub async fn main() -> AnyhowResult<()> {
   easyenv::init_all_with_default_logging(Some("info"));
 
   // NB: Do not check this secrets-containing dotenv file into VCS.
@@ -99,30 +99,57 @@ pub async fn main2() -> AnyhowResult<()> {
   let openai_client = Arc::new(Client::new()
       .with_api_key(startup_args.openai_secret_key.clone()));
 
-  let tokio_runtime = Runtime::new()?;
-
   let save_directory = SaveDirectory::new(&startup_args.save_directory);
 
-  info!("Starting async processes...");
+  let database_url = easyenv::get_env_string_required("DATABASE_URL")?;
+  let pool = SqlitePoolOptions::new()
+      .max_connections(5)
+      .connect(&database_url).await?;
 
-  tokio_runtime.spawn(async {
-    let _r = main_loop().await;
+  let job_state = Arc::new(JobState {
+    sqlite_pool: pool,
+    save_directory: save_directory.clone(),
   });
 
   info!("Starting web server...");
 
   let app_control_state2 = app_control_state.clone();
   let openai_client2 = openai_client.clone();
+  let job_state2 = job_state.clone();
 
+  // NB: both egui and imgui (which we aren't using) complain about launching on a non-main thread.
+  // They even complain that this is impossible on Windows (and our program aims to be multiplatform)
+  // Thus, we launch everything else into its own thread. (TODO: Jobs + Server in one thread)
   thread::spawn(move || {
     let server_future = launch_web_server(LaunchWebServerArgs {
       app_control_state: app_control_state2,
       openai_client: openai_client2,
       save_directory,
     });
-    actix_web::rt::System::new().block_on(server_future)
+
+    let tokio_runtime = Runtime::new()?;
+
+    tokio_runtime.spawn(async {
+      let _r = web_index_ingestion_main_loop(job_state2).await;
+    });
+
+    let runtime = actix_web::rt::System::new();
+
+    runtime.block_on(server_future)
   });
 
+//  info!("Starting async processes...");
+//
+//  thread::spawn(move|| {
+//
+//    tokio_runtime.block_on(async || {
+//      loop {
+//        thread::sleep(Duration::from_secs(60))
+//        std::future::
+//      }
+//    }.await)
+//  });
+
   info!("Starting GUI ...");
 
   let _r = launch_gui(startup_args.clone(), app_control_state.clone());

crates/programs/aichatbot-sidecar/src/shared_state/job_state.rs

@@ -0,0 +1,8 @@
+use crate::persistence::save_directory::SaveDirectory;
+use sqlx::{Pool, Sqlite};
+
+pub struct JobState {
+  pub sqlite_pool: Pool<Sqlite>,
+  pub save_directory: SaveDirectory,
+}
+

crates/programs/aichatbot-sidecar/src/workers/mod.rs

@@ -0,0 +1,4 @@
+pub mod fakeyou_enrichment;
+pub mod gpt_transform;
+pub mod web_content_scraping;
+pub mod web_index_ingestion;

crates/programs/aichatbot-sidecar/src/workers/web_content_scraping/mod.rs

@@ -0,0 +1 @@
+pub mod ingest_url_scrape_and_save;

crates/programs/aichatbot-sidecar/src/workers/web_index_ingestion/main_loop.rs

@@ -0,0 +1,57 @@
+use crate::shared_state::job_state::JobState;
+use errors::AnyhowResult;
+use log::{error, info};
+use sqlite_queries::queries::by_table::web_scraping_targets::insert_web_scraping_target::{Args, insert_web_scraping_target};
+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::WebScrapingResult;
+use web_scrapers::payloads::web_scraping_target::WebScrapingTarget;
+use web_scrapers::sites::techcrunch::techcrunch_indexer::{techcrunch_indexer, TechcrunchFeed};
+
+pub async fn web_index_ingestion_main_loop(job_state: Arc<JobState>) {
+
+  loop {
+    info!("web_index_ingestion main loop");
+
+    match single_iteration(&job_state).await {
+      Ok(_) => thread::sleep(Duration::from_secs(300)),
+      Err(err) => {
+        error!("Error with indexing: {:?}", err);
+        thread::sleep(Duration::from_secs(60));
+      }
+    }
+  }
+}
+
+async fn single_iteration(job_state: &Arc<JobState>) -> AnyhowResult<()> {
+
+  for variant in TechcrunchFeed::iter() {
+    let index_targets = techcrunch_indexer(variant).await?;
+    insert_targets(job_state, &index_targets).await?;
+  }
+
+
+  //let index_targets = techcrunch_indexer(TechcrunchFeed::Main).await?;
+  //insert_targets(&index_targets).await?;
+
+  Ok(())
+}
+
+async fn insert_targets(job_state: &Arc<JobState>, targets: &Vec<WebScrapingTarget>) -> AnyhowResult<()> {
+  for target in targets.iter() {
+    let _r = insert_web_scraping_target(Args {
+      canonical_url: &target.canonical_url,
+      web_content_type: target.web_content_type,
+      maybe_title: target.maybe_title.as_deref(),
+      maybe_article_full_image_url: target.maybe_full_image_url.as_deref(),
+      maybe_article_thumbnail_image_url: target.maybe_thumbnail_image_url.as_deref(),
+      sqlite_pool: &job_state.sqlite_pool,
+    }).await;
+  }
+
+  Ok(())
+}

crates/programs/aichatbot-sidecar/src/workers/web_index_ingestion/mod.rs

@@ -0,0 +1 @@
+pub mod main_loop;