scraping appears to work

crates/lib/sqlite_queries/src/queries/by_table/web_scraping_targets/list_web_scraping_targets.rs

@@ -4,6 +4,7 @@ use sqlx::SqlitePool;
 use enums::by_table::web_scraping_targets::web_content_type::WebContentType;
 
 pub struct WebScrapingTarget {
+  pub id: i64,
   pub canonical_url: String,
   pub web_content_type: WebContentType,
   pub maybe_title: Option<String>,
@@ -13,18 +14,22 @@ pub struct WebScrapingTarget {
   pub scrape_attempts: i64,
 }
 
-pub async fn insert_web_scraping_target(
+pub async fn list_web_scraping_targets(
   scraping_status: ScrapingStatus,
+  last_id: i64,
   limit: i64,
   sqlite_pool: &SqlitePool,
 ) -> AnyhowResult<Vec<WebScrapingTarget>> {
 
+  // NB: Sqlx doesn't support `WHERE ... IN (...)` "yet". :(
+  // https://github.com/launchbadge/sqlx/blob/6d0d7402c8a9cbea2676a1795e9fb50b0cf60c03/FAQ.md?plain=1#L73
   let scraping_status = scraping_status.to_str().to_string();
 
   let query = sqlx::query_as!(
     RawInternalWebScrapingTarget,
         r#"
 SELECT
+  id,
   canonical_url,
   web_content_type as `web_content_type: enums::by_table::web_scraping_targets::web_content_type::WebContentType`,
   maybe_title,
@@ -35,20 +40,22 @@ SELECT
 FROM web_scraping_targets
 WHERE
   scraping_status = ?
+  AND id > ?
 ORDER BY id DESC
 LIMIT ?
         "#,
         scraping_status,
+        last_id,
         limit,
     );
 
   let records = query.fetch_all(sqlite_pool)
       .await?;
 
-
   let records = records.into_iter()
       .map(|record : RawInternalWebScrapingTarget| {
         WebScrapingTarget {
+          id: record.id,
           canonical_url: record.canonical_url,
           web_content_type: record.web_content_type,
           maybe_title: record.maybe_title,
@@ -64,6 +71,7 @@ LIMIT ?
 }
 
 struct RawInternalWebScrapingTarget {
+  id: i64,
   canonical_url: String,
   web_content_type: WebContentType,
   maybe_title: Option<String>,

crates/lib/sqlite_queries/src/queries/by_table/web_scraping_targets/mod.rs

@@ -1,2 +1,3 @@
 pub mod insert_web_scraping_target;
-pub mod list_web_scraping_targets;
+pub mod list_web_scraping_targets;
+pub mod update_web_scraping_target;

crates/lib/sqlite_queries/src/queries/by_table/web_scraping_targets/update_web_scraping_target.rs

@@ -0,0 +1,39 @@
+use enums::by_table::web_scraping_targets::web_content_type::WebContentType;
+use errors::{anyhow, AnyhowResult};
+use sqlx::SqlitePool;
+use enums::by_table::web_scraping_targets::scraping_status::ScrapingStatus;
+
+pub struct Args <'a> {
+  pub canonical_url: &'a str,
+
+  pub scraping_status: ScrapingStatus,
+  pub scrape_attempts: i64,
+
+  pub sqlite_pool: &'a SqlitePool,
+}
+
+pub async fn update_web_scraping_target(args: Args<'_>) -> AnyhowResult<()> {
+  let scraping_status = args.scraping_status.to_str().to_string();
+  let query = sqlx::query!(
+        r#"
+UPDATE web_scraping_targets
+SET
+  scraping_status = ?,
+  scrape_attempts = ?,
+  version = version + 1
+WHERE
+  canonical_url = ?
+        "#,
+        scraping_status,
+        args.scrape_attempts,
+        args.canonical_url,
+    );
+
+  let query_result = query.execute(args.sqlite_pool)
+      .await;
+
+  match query_result {
+    Ok(_) => Ok(()),
+    Err(err) => Err(anyhow!("error updating: {:?}", err)),
+  }
+}

crates/programs/aichatbot-sidecar/_sql/migrations/2023-02-05-064307_web_scraping_targets/up.sql

@@ -31,7 +31,7 @@ CREATE TABLE web_scraping_targets (
 
   created_at DATETIME DEFAULT CURRENT_TIMESTAMP NOT NULL,
   updated_at DATETIME DEFAULT CURRENT_TIMESTAMP NOT NULL,
-  fetched_at DATETIME DEFAULT NULL,
+  -- fetched_at DATETIME DEFAULT NULL,
 
   version INT DEFAULT 0 NOT NULL
 );

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

@@ -19,42 +19,43 @@ use crate::gui::launch_gui::launch_gui;
 use crate::main_loop::main_loop;
 use crate::persistence::save_directory::SaveDirectory;
 use crate::shared_state::app_control_state::AppControlState;
+use crate::shared_state::job_state::JobState;
 use crate::startup_args::get_startup_args;
 use crate::web_server::launch_web_server::{launch_web_server, LaunchWebServerArgs};
+use crate::workers::web_content_scraping::main_loop::web_content_scraping_main_loop;
+use crate::workers::web_index_ingestion::main_loop::web_index_ingestion_main_loop;
+use enums::by_table::web_scraping_targets::web_content_type::WebContentType;
 use errors::AnyhowResult;
 use log::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::sync::Arc;
 use std::thread;
 use std::time::Duration;
-use sqlx::sqlite::SqlitePoolOptions;
 use tokio::runtime::Runtime;
-use enums::by_table::web_scraping_targets::web_content_type::WebContentType;
-use sqlite_queries::queries::by_table::web_scraping_targets::insert_web_scraping_target::{Args, insert_web_scraping_target};
 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 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;
+use workers::web_content_scraping::single_target::ingest_url_scrape_and_save::ingest_url_scrape_and_save;
 
-#[tokio::main]
-pub async fn main2() -> AnyhowResult<()> {
-  let database_url = easyenv::get_env_string_required("DATABASE_URL")?;
-  let pool = SqlitePoolOptions::new()
-      .max_connections(5)
-      .connect(&database_url).await?;
-
-  let _ = dotenv::from_filename(".env-aichatbot-secrets").ok();
-  let startup_args = get_startup_args()?;
-  let save_directory = SaveDirectory::new(&startup_args.save_directory);
-
-  let url = "https://techcrunch.com/2023/02/04/elon-musk-says-twitter-will-provide-a-free-write-only-api-to-bots-providing-good-content/";
-  ingest_url_scrape_and_save(url, WebContentType::TechCrunchArticle, &save_directory).await?;
-
-
-
-  Ok(())
-}
+//#[tokio::main]
+//pub async fn main2() -> AnyhowResult<()> {
+//  let database_url = easyenv::get_env_string_required("DATABASE_URL")?;
+//  let pool = SqlitePoolOptions::new()
+//      .max_connections(5)
+//      .connect(&database_url).await?;
+//
+//  let _ = dotenv::from_filename(".env-aichatbot-secrets").ok();
+//  let startup_args = get_startup_args()?;
+//  let save_directory = SaveDirectory::new(&startup_args.save_directory);
+//
+//  let url = "https://techcrunch.com/2023/02/04/elon-musk-says-twitter-will-provide-a-free-write-only-api-to-bots-providing-good-content/";
+//  ingest_url_scrape_and_save(url, WebContentType::TechCrunchArticle, &save_directory).await?;
+//
+//
+//
+//  Ok(())
+//}
 
 pub const LOG_LEVEL: &'static str = concat!(
   "info,",
@@ -97,6 +98,7 @@ pub async fn main() -> AnyhowResult<()> {
   let app_control_state2 = app_control_state.clone();
   let openai_client2 = openai_client.clone();
   let job_state2 = job_state.clone();
+  let job_state3 = 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)
@@ -114,11 +116,9 @@ pub async fn main() -> AnyhowResult<()> {
       let _r = web_index_ingestion_main_loop(job_state2).await;
     });
 
-    // TODO: Scraping thread (needs Sqlite queries)
-    //tokio_runtime.spawn(async {
-    //  // TODO...
-    //  let _r = web_index_ingestion_main_loop(job_state2.clone()).await;
-    //});
+    tokio_runtime.spawn(async {
+      let _r = web_content_scraping_main_loop(job_state3).await;
+    });
 
     // TODO: GPT Transformation thread
     //tokio_runtime.spawn(async {

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

@@ -0,0 +1,83 @@
+use crate::shared_state::job_state::JobState;
+use enums::by_table::web_scraping_targets::scraping_status::ScrapingStatus;
+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 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::WebScrapingResult;
+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::web_content_scraping::single_target::process_target_record::process_target_record;
+
+/// Follow up on articles tagged to be indexed by downloading and scraping their contents.
+pub async fn web_content_scraping_main_loop(job_state: Arc<JobState>) {
+  loop {
+    info!("web_content_scraping main loop");
+
+    single_job_loop_iteration(&job_state).await;
+
+    info!("web_index_ingestion 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<JobState>) {
+  let statuses = vec![ScrapingStatus::New, ScrapingStatus::Failed];
+  for status in statuses {
+    scrape_jobs_of_status(status, job_state).await;
+  }
+}
+
+async fn scrape_jobs_of_status(status: ScrapingStatus, job_state: &Arc<JobState>) {
+  const BATCH_SIZE : i64 = 10;
+
+  let mut last_id = 0;
+  let mut failure_count = 0;
+
+  loop {
+    let query_result = list_web_scraping_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 {
+      let result = process_target_record(&target, job_state).await;
+      if let Err(err) = result {
+        error!("Error processing target: {:?}", err);
+      }
+      last_id = target.id;
+    }
+  }
+}

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

@@ -1,2 +1,2 @@
-pub mod ingest_url_scrape_and_save;
 pub mod main_loop;
+pub mod single_target;

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

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

crates/programs/aichatbot-sidecar/src/workers/web_content_scraping/single_target/process_target_record.rs

@@ -0,0 +1,44 @@
+use crate::shared_state::job_state::JobState;
+use crate::workers::web_content_scraping::single_target::ingest_url_scrape_and_save::ingest_url_scrape_and_save;
+use enums::by_table::web_scraping_targets::scraping_status::ScrapingStatus;
+use errors::AnyhowResult;
+use log::error;
+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::update_web_scraping_target::{Args, update_web_scraping_target};
+use std::sync::Arc;
+
+pub async fn process_target_record(target: &WebScrapingTargetRecord, job_state: &Arc<JobState>) -> AnyhowResult<()> {
+  let result = ingest_url_scrape_and_save(
+    &target.canonical_url,
+    target.web_content_type,
+    &job_state.save_directory).await;
+
+  if let Err(err) = result {
+    error!("error ingesting url: {:?}", err);
+
+    let next_scraping_status = if target.scrape_attempts >= 2 {
+      ScrapingStatus::PermanentlyFailed
+    } else {
+      ScrapingStatus::Failed
+    };
+    update_web_scraping_target(Args {
+      canonical_url: &target.canonical_url,
+      scraping_status: next_scraping_status,
+      scrape_attempts: target.scrape_attempts + 1,
+      sqlite_pool: &job_state.sqlite_pool,
+    }).await?; // NB: If these queries fail, we could get stuck.
+
+    return Err(err);
+  }
+
+  // TODO: Insert result record in new table.
+
+  update_web_scraping_target(Args {
+    canonical_url: &target.canonical_url,
+    scraping_status: ScrapingStatus::Success,
+    scrape_attempts: target.scrape_attempts + 1,
+    sqlite_pool: &job_state.sqlite_pool,
+  }).await?; // NB: If these queries fail, we could get stuck.
+
+  Ok(())
+}