From 6a86f43a9201898d07ac8aefa7b4ced080dcfc3e Mon Sep 17 00:00:00 2001 From: Brandon Thomas Date: Tue, 7 Feb 2023 02:18:00 -0500 Subject: [PATCH] scraping appears to work --- .../list_web_scraping_targets.rs | 12 ++- .../by_table/web_scraping_targets/mod.rs | 3 +- .../update_web_scraping_target.rs | 39 +++++++++ .../up.sql | 2 +- crates/programs/aichatbot-sidecar/src/main.rs | 58 ++++++------- .../workers/web_content_scraping/main_loop.rs | 83 +++++++++++++++++++ .../src/workers/web_content_scraping/mod.rs | 2 +- .../ingest_url_scrape_and_save.rs | 0 .../web_content_scraping/single_target/mod.rs | 2 + .../single_target/process_target_record.rs | 44 ++++++++++ 10 files changed, 211 insertions(+), 34 deletions(-) create mode 100644 crates/lib/sqlite_queries/src/queries/by_table/web_scraping_targets/update_web_scraping_target.rs rename crates/programs/aichatbot-sidecar/src/workers/web_content_scraping/{ => single_target}/ingest_url_scrape_and_save.rs (100%) create mode 100644 crates/programs/aichatbot-sidecar/src/workers/web_content_scraping/single_target/mod.rs create mode 100644 crates/programs/aichatbot-sidecar/src/workers/web_content_scraping/single_target/process_target_record.rs diff --git a/crates/lib/sqlite_queries/src/queries/by_table/web_scraping_targets/list_web_scraping_targets.rs b/crates/lib/sqlite_queries/src/queries/by_table/web_scraping_targets/list_web_scraping_targets.rs index b6294856ea..9c99e6cd84 100644 --- a/crates/lib/sqlite_queries/src/queries/by_table/web_scraping_targets/list_web_scraping_targets.rs +++ b/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, @@ -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> { + // 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, diff --git a/crates/lib/sqlite_queries/src/queries/by_table/web_scraping_targets/mod.rs b/crates/lib/sqlite_queries/src/queries/by_table/web_scraping_targets/mod.rs index a0a50cdf31..d480b11249 100644 --- a/crates/lib/sqlite_queries/src/queries/by_table/web_scraping_targets/mod.rs +++ b/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; \ No newline at end of file +pub mod list_web_scraping_targets; +pub mod update_web_scraping_target; \ No newline at end of file diff --git a/crates/lib/sqlite_queries/src/queries/by_table/web_scraping_targets/update_web_scraping_target.rs b/crates/lib/sqlite_queries/src/queries/by_table/web_scraping_targets/update_web_scraping_target.rs new file mode 100644 index 0000000000..4aa6e55294 --- /dev/null +++ b/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)), + } +} diff --git a/crates/programs/aichatbot-sidecar/_sql/migrations/2023-02-05-064307_web_scraping_targets/up.sql b/crates/programs/aichatbot-sidecar/_sql/migrations/2023-02-05-064307_web_scraping_targets/up.sql index af8f15ff2c..a5e9b0cdb7 100644 --- a/crates/programs/aichatbot-sidecar/_sql/migrations/2023-02-05-064307_web_scraping_targets/up.sql +++ b/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 ); diff --git a/crates/programs/aichatbot-sidecar/src/main.rs b/crates/programs/aichatbot-sidecar/src/main.rs index e642feac73..debac671bf 100644 --- a/crates/programs/aichatbot-sidecar/src/main.rs +++ b/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 { diff --git a/crates/programs/aichatbot-sidecar/src/workers/web_content_scraping/main_loop.rs b/crates/programs/aichatbot-sidecar/src/workers/web_content_scraping/main_loop.rs index e69de29bb2..5ab93614de 100644 --- a/crates/programs/aichatbot-sidecar/src/workers/web_content_scraping/main_loop.rs +++ b/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) { + 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) { + 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) { + 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; + } + } +} diff --git a/crates/programs/aichatbot-sidecar/src/workers/web_content_scraping/mod.rs b/crates/programs/aichatbot-sidecar/src/workers/web_content_scraping/mod.rs index 6b1b262316..e98bc899e3 100644 --- a/crates/programs/aichatbot-sidecar/src/workers/web_content_scraping/mod.rs +++ b/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; diff --git a/crates/programs/aichatbot-sidecar/src/workers/web_content_scraping/ingest_url_scrape_and_save.rs b/crates/programs/aichatbot-sidecar/src/workers/web_content_scraping/single_target/ingest_url_scrape_and_save.rs similarity index 100% rename from crates/programs/aichatbot-sidecar/src/workers/web_content_scraping/ingest_url_scrape_and_save.rs rename to crates/programs/aichatbot-sidecar/src/workers/web_content_scraping/single_target/ingest_url_scrape_and_save.rs diff --git a/crates/programs/aichatbot-sidecar/src/workers/web_content_scraping/single_target/mod.rs b/crates/programs/aichatbot-sidecar/src/workers/web_content_scraping/single_target/mod.rs new file mode 100644 index 0000000000..fb93c14f83 --- /dev/null +++ b/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; diff --git a/crates/programs/aichatbot-sidecar/src/workers/web_content_scraping/single_target/process_target_record.rs b/crates/programs/aichatbot-sidecar/src/workers/web_content_scraping/single_target/process_target_record.rs new file mode 100644 index 0000000000..fa0c01b64c --- /dev/null +++ b/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) -> 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(()) +}