stream_response #6

Merged
Oliver merged 6 commits from stream_response into main 2025-04-17 15:39:50 +00:00
5 changed files with 59 additions and 1496 deletions

View File

@ -43,7 +43,6 @@ impl Website {
} }
pub fn set_crawled(&mut self) { pub fn set_crawled(&mut self) {
trace!("Set crawled to true");
self.crawled = true self.crawled = true
} }

View File

@ -1,30 +1,37 @@
use std::{ffi::OsStr, io::ErrorKind, path::PathBuf}; use std::{io::ErrorKind, path::PathBuf};
use reqwest::header::HeaderValue;
use tokio::fs; use tokio::fs;
use tracing::{error, trace}; use tracing::{error, trace, warn};
use url::Url; use url::Url;
pub fn as_path(url: &Url) -> PathBuf { pub fn as_path(url: &Url, content_type: &HeaderValue) -> PathBuf {
// extract data from url to save it accurately // extract data from url to save it accurately
let url_path = PathBuf::from("./downloaded/".to_string() + url.domain().unwrap_or("UnknownDomain") + url.path()); let mut url_path = PathBuf::from("./downloaded/".to_string() + url.domain().unwrap_or("UnknownDomain") + url.path());
// if it's a file if let Ok(header) = content_type.to_str() {
let (basepath, filename) = if url_path.extension().filter(valid_file_extension).is_some() { // text/html; charset=UTF-8; option=value
// get everything up till the file let ttype = if let Some((t, _)) = header.split_once(';') {
let basepath = url_path.ancestors().skip(1).take(1).collect::<PathBuf>(); t
// get the file name } else {
let filename = url_path.file_name().expect("This should exist").to_string_lossy(); header
trace!("Save path: {:?} and base path: {:?}", &url_path, &basepath); };
(basepath, filename.to_string())
if let Some((ttype, subtype)) = ttype.split_once('/') {
trace!("Found Content-Type to be: {ttype}/{subtype} for {}", url.to_string());
// If the Content-Type header is "*/html" (most likely "text/html") and the path's
// extension is anything but html:
if subtype=="html" && !url_path.extension().is_some_and(|f| f=="html" || f=="htm" ) {
// time to slap a index.html to the end of that path there!
url_path = url_path.join("index.html");
}
}
} else { } else {
(url_path.clone(), "index.html".into()) warn!("Header: {:?} couldn't be parsed into a string!", content_type);
}; }
trace!("Final path for {} is: {:?}", url, url_path);
let mut path = PathBuf::new(); url_path
path = path.join(basepath);
path = path.join(filename);
path
} }
pub async fn init(filename: &PathBuf) -> Option<fs::File> { pub async fn init(filename: &PathBuf) -> Option<fs::File> {
@ -50,29 +57,10 @@ pub async fn init(filename: &PathBuf) -> Option<fs::File> {
error!("Couldn't get file's parents: {:?}", &filename); error!("Couldn't get file's parents: {:?}", &filename);
} }
} else { } else {
error!("File creation: {err} {:?}", filename); error!("File open error: {err} {:?}", filename);
} }
// we don't care about other errors, we can't/shouldn't fix them // we don't care about other errors, we can't/shouldn't fix them
None None
} }
} }
} }
fn valid_file_extension(take: &&OsStr) -> bool {
let los = take.to_string_lossy();
let all = los.split('.');
match all.last() {
Some(s) => {
// FIXME it's worth noting that the dumb tlds like .zip are in here,
// which could cause problems
let all_domains = include_str!("tlds-alpha-by-domain.txt");
// check if it is a domain
match all_domains.lines().map(str::to_lowercase).find(|x| x==s.to_lowercase().as_str()) {
Some(_) => false,
None => true
}
},
None => false,
}
}

View File

@ -19,8 +19,8 @@ use metrics::{counter, gauge};
use metrics_exporter_prometheus::PrometheusBuilder; 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, task::JoinSet}; use tokio::{io::{AsyncWriteExt, BufWriter}, task::JoinSet};
use tracing::{debug, error, info, instrument, level_filters::LevelFilter, trace, trace_span}; 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;
@ -174,21 +174,41 @@ 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 // Get body from response
let headers = response.headers();
let path = filesystem::as_path(&site.site); #[allow(non_snake_case)]
let CT = headers.get("Content-Type");
let ct = headers.get("content-type");
if CT.is_none() && ct.is_none() {
}
let ct = match (CT,ct) {
(None, None) => {
warn!("Server did not respond with Content-Type header. Url: {} Headers: ({:?})", site.site.to_string(), headers);
return
},
(None, Some(a)) => a,
(Some(a), None) => a,
(Some(a), Some(_)) => a,
};
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(mut file) = filesystem::init(&path).await { if let Some(file) = filesystem::init(&path).await {
let should_parse = path.to_string_lossy().ends_with(".html"); let should_parse = path.to_string_lossy().ends_with(".html");
let mut buf: Vec<u8> = Vec::new(); 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();
info!("Writing at: {:?}", path);
while let Some(data) = stream.next().await { while let Some(data) = stream.next().await {
match data { match data {
Ok(data) => { Ok(data) => {
debug!("Writing at: {:?}", path); let _ = writer.write_all(&data).await;
let _ = file.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.
if should_parse { if should_parse {
@ -200,6 +220,7 @@ async fn process(mut site: Website, db: Surreal<Client>, reqwest: reqwest::Clien
}, },
} }
} }
let _ = writer.flush();
if should_parse { if should_parse {
// Parse document and get relationships // Parse document and get relationships

View File

@ -39,7 +39,7 @@ impl TokenSink for Website {
if let Some(mut parsed) = url { if let Some(mut parsed) = url {
parsed.set_query(None); parsed.set_query(None);
parsed.set_fragment(None); parsed.set_fragment(None);
debug!("Final cleaned URL: `{}`", parsed.to_string()); trace!("Final cleaned URL: `{}`", parsed.to_string());
let web = Website::new(&parsed.to_string(), false); let web = Website::new(&parsed.to_string(), false);
links.push(web); links.push(web);
} }
@ -127,16 +127,15 @@ fn try_get_url(parent: &Url, link: &str) -> Option<Url> {
} }
let url = origin.clone() + link; let url = origin.clone() + link;
trace!("Built `{url}` from `{origin} + `{}`", link.to_string());
if let Ok(url) = Url::parse(&url) { if let Ok(url) = Url::parse(&url) {
trace!("Saved relative url `{}` AS: `{}`", link, url); trace!("Built `{url}` from `{origin} + `{}`", link.to_string());
Some(url) Some(url)
} else { } else {
error!( error!(
"Failed to reconstruct a url from relative url: `{}` on site: `{}`", "Failed to reconstruct a url from relative url: `{}` on site: `{}`. Failed url was: {}",
link, link,
parent.to_string() parent.to_string(),
url
); );
None None
} }

File diff suppressed because it is too large Load Diff