Compare commits

..

No commits in common. "6f98001d8e430015c1b69629121b57a562a9b01d" and "5850f19cab14def45dc8585ba39dda20168fcb6c" have entirely different histories.

3 changed files with 25 additions and 40 deletions

View File

@ -3,12 +3,12 @@ surreal_url = "localhost:8000"
surreal_username = "root" surreal_username = "root"
surreal_password = "root" surreal_password = "root"
surreal_ns = "test" surreal_ns = "test"
surreal_db = "v1.21.1" surreal_db = "v1.20.3"
# Crawler config # Crawler config
# crawl_filter = "https://ftpgeoinfo.msl.mt.gov/Data/Spatial/MSDI/Imagery/2023_NAIP/UTM_County_Mosaics/" crawl_filter = "https://ftpgeoinfo.msl.mt.gov/Data/Spatial/MSDI/Imagery/2023_NAIP/UTM_County_Mosaics/"
crawl_filter = "https://oliveratkinson.net" # crawl_filter = "https://oliveratkinson.net"
# start_url = "https://ftpgeoinfo.msl.mt.gov/Data/Spatial/MSDI/Imagery/2023_NAIP/UTM_County_Mosaics/" start_url = "https://ftpgeoinfo.msl.mt.gov/Data/Spatial/MSDI/Imagery/2023_NAIP/UTM_County_Mosaics/"
start_url = "https://oliveratkinson.net" # start_url = "https://oliveratkinson.net"
budget = 1000 budget = 100
batch_size = 500 batch_size = 5

View File

@ -20,18 +20,12 @@ pub struct Website {
pub site: Url, pub site: Url,
/// Wether or not this link has been crawled yet /// Wether or not this link has been crawled yet
pub crawled: bool, pub crawled: bool,
/// 200, 404, etc
pub status_code: u16,
} }
// manual impl to make tracing look nicer // manual impl to make tracing look nicer
impl Debug for Website { impl Debug for Website {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
f.debug_struct("Website") f.debug_struct("Website").field("site", &self.site).finish()
.field("host", &self.site.host())
.field("path", &self.site.path())
.field("status_code", &self.status_code)
.finish()
} }
} }
@ -44,11 +38,14 @@ impl Website {
}; };
Self { Self {
crawled, crawled,
site, site
status_code: 0,
} }
} }
pub fn set_crawled(&mut self) {
self.crawled = true
}
// Insert ever item in the vec into surreal, crawled state will be preserved as TRUE // Insert ever item in the vec into surreal, crawled state will be preserved as TRUE
// if already in the database as such or incoming data is TRUE. // if already in the database as such or incoming data is TRUE.
#[instrument(skip(db))] #[instrument(skip(db))]
@ -56,13 +53,11 @@ impl Website {
counter!(STORE).increment(1); counter!(STORE).increment(1);
let mut things = Vec::with_capacity(all.len()); let mut things = Vec::with_capacity(all.len());
// FIXME failes *sometimes* because "Resource Busy"
match db match db
.query( .query(
"INSERT INTO website $array "INSERT INTO website $array
ON DUPLICATE KEY UPDATE ON DUPLICATE KEY UPDATE
accessed_at = time::now(), accessed_at = time::now(),
status_code = $input.status_code,
crawled = crawled OR $input.crawled crawled = crawled OR $input.crawled
RETURN VALUE id; RETURN VALUE id;
", ",

View File

@ -1,4 +1,5 @@
#![feature(ip_from)] #![feature(ip_from)]
#![feature(async_closure)]
#![warn(clippy::expect_used)] #![warn(clippy::expect_used)]
#![deny(clippy::unwrap_used)] #![deny(clippy::unwrap_used)]
@ -19,7 +20,7 @@ use metrics_exporter_prometheus::PrometheusBuilder;
use serde::Deserialize; use serde::Deserialize;
use surrealdb::{engine::remote::ws::Client, Surreal}; use surrealdb::{engine::remote::ws::Client, Surreal};
use tokio::{io::{AsyncWriteExt, BufWriter}, task::JoinSet}; use tokio::{io::{AsyncWriteExt, BufWriter}, task::JoinSet};
use tracing::{debug, debug_span, error, info, instrument, level_filters::LevelFilter, trace, trace_span, warn}; use tracing::{debug, error, info, instrument, level_filters::LevelFilter, trace, trace_span, warn};
use tracing_subscriber::{fmt, layer::SubscriberExt, EnvFilter, Layer, Registry}; use tracing_subscriber::{fmt, layer::SubscriberExt, EnvFilter, Layer, Registry};
mod db; mod db;
@ -172,13 +173,15 @@ async fn process(mut site: Website, db: Surreal<Client>, reqwest: reqwest::Clien
// Send the http request (get) // Send the http request (get)
if let Ok(response) = request_builder.send().await { if let Ok(response) = request_builder.send().await {
// Get body from response
let headers = response.headers(); let headers = response.headers();
let code = response.status();
#[allow(non_snake_case)] #[allow(non_snake_case)]
let CT = headers.get("Content-Type"); let CT = headers.get("Content-Type");
let ct = headers.get("content-type"); let ct = headers.get("content-type");
if CT.is_none() && ct.is_none() {
}
let ct = match (CT,ct) { let ct = match (CT,ct) {
(None, None) => { (None, None) => {
warn!("Server did not respond with Content-Type header. Url: {} Headers: ({:?})", site.site.to_string(), headers); warn!("Server did not respond with Content-Type header. Url: {} Headers: ({:?})", site.site.to_string(), headers);
@ -189,20 +192,18 @@ async fn process(mut site: Website, db: Surreal<Client>, reqwest: reqwest::Clien
(Some(a), Some(_)) => a, (Some(a), Some(_)) => a,
}; };
// create filepath (handles / -> /index.html)
let path = filesystem::as_path(&site.site, ct); let path = filesystem::as_path(&site.site, ct);
// make sure that the file is good to go // make sure that the file is good to go
if let Some(file) = filesystem::init(&path).await { if let Some(file) = filesystem::init(&path).await {
// Get body from response let should_parse = path.to_string_lossy().ends_with(".html");
let mut buf: Vec<u8> = Vec::new();
let mut writer = BufWriter::new(file);
// stream the response onto the disk // stream the response onto the disk
let mut stream = response.bytes_stream(); let mut stream = response.bytes_stream();
let should_parse = path.to_string_lossy().ends_with(".html");
let mut writer = BufWriter::new(file);
let mut buf: Vec<u8> = Vec::new();
// Write file to disk
info!("Writing at: {:?}", path); info!("Writing at: {:?}", path);
while let Some(data) = stream.next().await { while let Some(data) = stream.next().await {
match data { match data {
@ -210,7 +211,6 @@ async fn process(mut site: Website, db: Surreal<Client>, reqwest: reqwest::Clien
let _ = writer.write_all(&data).await; let _ = writer.write_all(&data).await;
// If we are going to parse this file later, we will save it // If we are going to parse this file later, we will save it
// into memory as well as the disk. // into memory as well as the disk.
// We do this because the data here might be incomplete
if should_parse { if should_parse {
data.iter().for_each(|f| buf.push(*f)); data.iter().for_each(|f| buf.push(*f));
} }
@ -222,12 +222,7 @@ async fn process(mut site: Website, db: Surreal<Client>, reqwest: reqwest::Clien
} }
let _ = writer.flush(); let _ = writer.flush();
// (If needed) Parse the file
if should_parse { if should_parse {
let span = debug_span!("Should Parse");
let enter = span.enter();
// Parse document and get relationships // Parse document and get relationships
let sites = parser::parse(&site, &buf).await; let sites = parser::parse(&site, &buf).await;
// De-duplicate this list // De-duplicate this list
@ -241,8 +236,6 @@ async fn process(mut site: Website, db: Surreal<Client>, reqwest: reqwest::Clien
trace!("Saved {diff} from being entered into the db by de-duping"); trace!("Saved {diff} from being entered into the db by de-duping");
// Store all the other sites so that we can link to them. // Store all the other sites so that we can link to them.
let _ = Website::store_all(de_dupe_sites, &db).await; let _ = Website::store_all(de_dupe_sites, &db).await;
drop(enter);
} }
// METRICS // METRICS
@ -250,14 +243,11 @@ async fn process(mut site: Website, db: Surreal<Client>, reqwest: reqwest::Clien
counter!(GET_METRIC).increment(1); counter!(GET_METRIC).increment(1);
// update self in db // update self in db
site.crawled = true; site.set_crawled();
site.status_code = code.as_u16(); Website::store_all(vec![site], &db).await;
Website::store_all(vec![site.clone()], &db).await;
} else { } else {
error!("File failed to cooperate: {:?}", path); error!("File failed to cooperate: {:?}", path);
} }
trace!("Done processing: {}", &site.site);
} else { } else {
error!("Failed to get: {}", &site.site); error!("Failed to get: {}", &site.site);
} }