Compare commits
No commits in common. "main" and "recursive-crawl" have entirely different histories.
main
...
recursive-
6 changed files with 734 additions and 898 deletions
1340
Cargo.lock
generated
1340
Cargo.lock
generated
File diff suppressed because it is too large
Load diff
21
Cargo.toml
21
Cargo.toml
|
@ -5,18 +5,17 @@ authors = ["Felix Ableitner"]
|
|||
edition = "2018"
|
||||
|
||||
[dependencies]
|
||||
reqwest = { version = "0.11.13", default-features = false, features = ["json", "rustls-tls"] }
|
||||
serde = { version = "1.0.149", features = ["derive"] }
|
||||
anyhow = "1.0.66"
|
||||
tokio = { version = "1.22.0", features = ["macros", "rt-multi-thread"] }
|
||||
futures = "0.3.25"
|
||||
serde_json = "1.0.89"
|
||||
semver = "1.0.14"
|
||||
once_cell = "1.16.0"
|
||||
lemmy_api_common = "=0.16.0"
|
||||
lemmy_db_schema = "=0.16.0"
|
||||
reqwest = { version = "0.11.10", default-features = false, features = ["json", "rustls-tls"] }
|
||||
serde = { version = "1.0.137", features = ["derive"] }
|
||||
anyhow = "1.0.57"
|
||||
tokio = { version = "1.18.1", features = ["macros", "rt-multi-thread"] }
|
||||
futures = "0.3.21"
|
||||
serde_json = "1.0.81"
|
||||
semver = "1.0.9"
|
||||
once_cell = "1.10.0"
|
||||
lemmy_api_common = "0.16.0"
|
||||
async-recursion = "1.0.0"
|
||||
log = "0.4.17"
|
||||
derive-new = "0.5.9"
|
||||
stderrlog = "0.5.4"
|
||||
stderrlog = "0.5.1"
|
||||
structopt = "0.3.26"
|
||||
|
|
90
src/crawl.rs
90
src/crawl.rs
|
@ -1,12 +1,12 @@
|
|||
use crate::node_info::{NodeInfo, NodeInfoWellKnown};
|
||||
use crate::CLIENT;
|
||||
use anyhow::{anyhow, Error};
|
||||
use crate::REQUEST_TIMEOUT;
|
||||
use anyhow::Error;
|
||||
use async_recursion::async_recursion;
|
||||
use futures::future::join_all;
|
||||
use lemmy_api_common::site::GetSiteResponse;
|
||||
use log::debug;
|
||||
use reqwest::Url;
|
||||
use semver::Version;
|
||||
use serde::Serialize;
|
||||
use std::collections::HashSet;
|
||||
use std::ops::Deref;
|
||||
use std::sync::Arc;
|
||||
|
@ -27,16 +27,15 @@ pub struct CrawlParams {
|
|||
crawled_instances: Arc<Mutex<HashSet<String>>>,
|
||||
}
|
||||
|
||||
#[derive(Debug)]
|
||||
pub struct CrawlResult {
|
||||
#[derive(Serialize, Debug)]
|
||||
pub struct InstanceDetails {
|
||||
pub domain: String,
|
||||
pub node_info: NodeInfo,
|
||||
pub site_info: Option<GetSiteResponse>,
|
||||
pub site_info: GetSiteResponse,
|
||||
}
|
||||
|
||||
impl CrawlJob {
|
||||
#[async_recursion]
|
||||
pub async fn crawl(self) -> Vec<Result<CrawlResult, Error>> {
|
||||
pub async fn crawl(self) -> Vec<Result<InstanceDetails, Error>> {
|
||||
// need to acquire and release mutex before recursing, otherwise it will deadlock
|
||||
{
|
||||
let mut crawled_instances = self.params.crawled_instances.deref().lock().await;
|
||||
|
@ -53,83 +52,46 @@ impl CrawlJob {
|
|||
return vec![];
|
||||
}
|
||||
|
||||
debug!(
|
||||
"Starting crawl for {}, distance {}",
|
||||
&self.domain, &self.current_distance
|
||||
);
|
||||
let (node_info, site_info) = match self.fetch_instance_details().await {
|
||||
debug!("Starting crawl for {}, distance {}", &self.domain, &self.current_distance);
|
||||
let site_info = match self.fetch_instance_details().await {
|
||||
Ok(o) => o,
|
||||
Err(e) => return vec![Err(e)],
|
||||
};
|
||||
let mut crawl_result = CrawlResult {
|
||||
domain: self.domain.clone(),
|
||||
node_info,
|
||||
site_info: None,
|
||||
};
|
||||
|
||||
if let Some(site_info) = site_info {
|
||||
match Version::parse(&site_info.version) {
|
||||
Ok(version) => {
|
||||
if version < self.params.min_lemmy_version {
|
||||
return vec![Ok(crawl_result)];
|
||||
}
|
||||
}
|
||||
Err(e) => return vec![Err(e.into())],
|
||||
if site_info.1 < self.params.min_lemmy_version {
|
||||
return vec![];
|
||||
}
|
||||
|
||||
let mut result = vec![];
|
||||
if let Some(federated) = &site_info.federated_instances {
|
||||
if let Some(federated) = &site_info.0.federated_instances {
|
||||
for domain in federated.linked.iter() {
|
||||
let crawl_job = CrawlJob::new(
|
||||
domain.clone(),
|
||||
self.current_distance + 1,
|
||||
self.params.clone(),
|
||||
);
|
||||
let crawl_job =
|
||||
CrawlJob::new(domain.clone(), self.current_distance + 1, self.params.clone());
|
||||
result.push(crawl_job.crawl());
|
||||
}
|
||||
}
|
||||
|
||||
let mut result2: Vec<Result<CrawlResult, Error>> =
|
||||
let mut result2: Vec<Result<InstanceDetails, Error>> =
|
||||
join_all(result).await.into_iter().flatten().collect();
|
||||
debug!("Successfully finished crawl for {}", &self.domain);
|
||||
crawl_result.site_info = Some(site_info);
|
||||
result2.push(Ok(crawl_result));
|
||||
result2.push(Ok(InstanceDetails {
|
||||
domain: self.domain,
|
||||
site_info: site_info.0,
|
||||
}));
|
||||
|
||||
result2
|
||||
} else {
|
||||
vec![Ok(crawl_result)]
|
||||
}
|
||||
}
|
||||
|
||||
async fn fetch_instance_details(&self) -> Result<(NodeInfo, Option<GetSiteResponse>), Error> {
|
||||
let rel_node_info: Url = Url::parse("http://nodeinfo.diaspora.software/ns/schema/2.0")
|
||||
.expect("parse nodeinfo relation url");
|
||||
let node_info_well_known = CLIENT
|
||||
.get(&format!("https://{}/.well-known/nodeinfo", &self.domain))
|
||||
.send()
|
||||
.await?
|
||||
.json::<NodeInfoWellKnown>()
|
||||
.await?;
|
||||
let node_info_url = node_info_well_known
|
||||
.links
|
||||
.into_iter()
|
||||
.find(|l| l.rel == rel_node_info)
|
||||
.ok_or_else(|| anyhow!("failed to find nodeinfo link for {}", &self.domain))?
|
||||
.href;
|
||||
let node_info = CLIENT
|
||||
.get(node_info_url)
|
||||
.send()
|
||||
.await?
|
||||
.json::<NodeInfo>()
|
||||
.await?;
|
||||
|
||||
async fn fetch_instance_details(&self) -> Result<(GetSiteResponse, Version), Error> {
|
||||
let site_info_url = format!("https://{}/api/v3/site", &self.domain);
|
||||
let site_info = CLIENT
|
||||
.get(&format!("https://{}/api/v3/site", &self.domain))
|
||||
.get(&site_info_url)
|
||||
.timeout(REQUEST_TIMEOUT)
|
||||
.send()
|
||||
.await?
|
||||
.json::<GetSiteResponse>()
|
||||
.await
|
||||
.ok();
|
||||
Ok((node_info, site_info))
|
||||
.await?;
|
||||
let version = Version::parse(&site_info.version)?;
|
||||
Ok((site_info, version))
|
||||
}
|
||||
}
|
||||
|
|
90
src/lib.rs
90
src/lib.rs
|
@ -1,46 +1,28 @@
|
|||
#[macro_use]
|
||||
extern crate derive_new;
|
||||
|
||||
use crate::crawl::{CrawlJob, CrawlParams, CrawlResult};
|
||||
use crate::node_info::{NodeInfo, NodeInfoUsage, NodeInfoUsers};
|
||||
use crate::crawl::{CrawlJob, CrawlParams, InstanceDetails};
|
||||
use anyhow::Error;
|
||||
use futures::future::join_all;
|
||||
use lemmy_api_common::site::GetSiteResponse;
|
||||
use log::warn;
|
||||
use once_cell::sync::Lazy;
|
||||
use reqwest::{Client, ClientBuilder};
|
||||
use reqwest::Client;
|
||||
use semver::Version;
|
||||
use serde::Serialize;
|
||||
use std::collections::HashSet;
|
||||
use std::sync::Arc;
|
||||
use std::time::Duration;
|
||||
use tokio::sync::Mutex;
|
||||
|
||||
pub mod crawl;
|
||||
mod node_info;
|
||||
|
||||
const REQUEST_TIMEOUT: Duration = Duration::from_secs(10);
|
||||
pub const REQUEST_TIMEOUT: Duration = Duration::from_secs(10);
|
||||
|
||||
static CLIENT: Lazy<Client> = Lazy::new(|| {
|
||||
ClientBuilder::new()
|
||||
.timeout(REQUEST_TIMEOUT)
|
||||
.user_agent("lemmy-stats-crawler")
|
||||
.build()
|
||||
.expect("build reqwest client")
|
||||
});
|
||||
|
||||
#[derive(Serialize, Debug)]
|
||||
pub struct CrawlResult2 {
|
||||
pub domain: String,
|
||||
pub site_info: GetSiteResponse,
|
||||
pub federated_counts: Option<NodeInfoUsage>,
|
||||
}
|
||||
static CLIENT: Lazy<Client> = Lazy::new(Client::default);
|
||||
|
||||
pub async fn start_crawl(
|
||||
start_instances: Vec<String>,
|
||||
exclude_domains: Vec<String>,
|
||||
max_distance: i32,
|
||||
) -> Result<Vec<CrawlResult2>, Error> {
|
||||
) -> Result<Vec<InstanceDetails>, Error> {
|
||||
let params = Arc::new(CrawlParams::new(
|
||||
min_lemmy_version().await?,
|
||||
exclude_domains,
|
||||
|
@ -53,29 +35,25 @@ pub async fn start_crawl(
|
|||
jobs.push(job.crawl());
|
||||
}
|
||||
|
||||
let crawl_results: Vec<CrawlResult> = join_all(jobs)
|
||||
// TODO: log the errors
|
||||
let mut instance_details: Vec<InstanceDetails> = join_all(jobs)
|
||||
.await
|
||||
.into_iter()
|
||||
.flatten()
|
||||
.inspect(|r| {
|
||||
if let Err(e) = r {
|
||||
warn!("{}", e)
|
||||
}
|
||||
})
|
||||
.filter_map(Result::ok)
|
||||
.filter_map(|r| r.ok())
|
||||
.collect();
|
||||
let mut crawl_results = calculate_federated_site_aggregates(crawl_results)?;
|
||||
|
||||
// Sort by active monthly users descending
|
||||
crawl_results.sort_unstable_by_key(|i| {
|
||||
instance_details.sort_unstable_by_key(|i| {
|
||||
i.site_info
|
||||
.site_view
|
||||
.as_ref()
|
||||
.map(|s| s.counts.users_active_month)
|
||||
.unwrap_or(0)
|
||||
});
|
||||
crawl_results.reverse();
|
||||
Ok(crawl_results)
|
||||
instance_details.reverse();
|
||||
|
||||
Ok(instance_details)
|
||||
}
|
||||
|
||||
/// calculate minimum allowed lemmy version based on current version. in case of current version
|
||||
|
@ -92,47 +70,3 @@ async fn min_lemmy_version() -> Result<Version, Error> {
|
|||
version.minor -= 1;
|
||||
Ok(version)
|
||||
}
|
||||
|
||||
fn calculate_federated_site_aggregates(
|
||||
crawl_results: Vec<CrawlResult>,
|
||||
) -> Result<Vec<CrawlResult2>, Error> {
|
||||
let node_info: Vec<(String, NodeInfo)> = crawl_results
|
||||
.iter()
|
||||
.map(|c| (c.domain.clone(), c.node_info.clone()))
|
||||
.collect();
|
||||
let lemmy_instances: Vec<(String, GetSiteResponse)> = crawl_results
|
||||
.into_iter()
|
||||
.filter_map(|c| {
|
||||
let domain = c.domain;
|
||||
c.site_info.map(|c2| (domain, c2))
|
||||
})
|
||||
.collect();
|
||||
let mut ret = vec![];
|
||||
for instance in &lemmy_instances {
|
||||
let federated_counts = if let Some(federated_instances) = &instance.1.federated_instances {
|
||||
node_info
|
||||
.iter()
|
||||
.filter(|i| federated_instances.linked.contains(&i.0) || i.0 == instance.0)
|
||||
.map(|i| i.1.usage.clone())
|
||||
.reduce(|a, b| NodeInfoUsage {
|
||||
users: NodeInfoUsers {
|
||||
total: a.users.total + b.users.total,
|
||||
active_halfyear: a.users.active_halfyear + b.users.active_halfyear,
|
||||
active_month: a.users.active_month + b.users.active_month,
|
||||
},
|
||||
posts: a.posts + b.posts,
|
||||
comments: a.comments + b.comments,
|
||||
})
|
||||
} else {
|
||||
None
|
||||
};
|
||||
// TODO: workaround because GetSiteResponse doesnt implement clone
|
||||
let site_info = serde_json::from_str(&serde_json::to_string(&instance.1)?)?;
|
||||
ret.push(CrawlResult2 {
|
||||
domain: instance.0.clone(),
|
||||
site_info,
|
||||
federated_counts,
|
||||
});
|
||||
}
|
||||
Ok(ret)
|
||||
}
|
||||
|
|
22
src/main.rs
22
src/main.rs
|
@ -1,25 +1,25 @@
|
|||
use anyhow::Error;
|
||||
use lemmy_stats_crawler::{start_crawl, CrawlResult2};
|
||||
use lemmy_stats_crawler::crawl::InstanceDetails;
|
||||
use lemmy_stats_crawler::start_crawl;
|
||||
use serde::Serialize;
|
||||
use structopt::StructOpt;
|
||||
|
||||
#[derive(StructOpt, Debug)]
|
||||
#[structopt()]
|
||||
struct Parameters {
|
||||
#[structopt(short, long, use_delimiter = true, default_value = "lemmy.ml")]
|
||||
#[structopt(short, long, default_value = "lemmy.ml")]
|
||||
start_instances: Vec<String>,
|
||||
#[structopt(
|
||||
short,
|
||||
long,
|
||||
use_delimiter = true,
|
||||
default_value = "ds9.lemmy.ml,enterprise.lemmy.ml,voyager.lemmy.ml,test.lemmy.ml"
|
||||
)]
|
||||
#[structopt(short, long, default_value = "ds9.lemmy.ml, enterprise.lemmy.ml, voyager.lemmy.ml, test.lemmy.ml")]
|
||||
exclude_instances: Vec<String>,
|
||||
#[structopt(short, long, default_value = "20")]
|
||||
max_crawl_distance: i32,
|
||||
|
||||
/// Silence all output
|
||||
#[structopt(short, long)]
|
||||
quiet: bool,
|
||||
/// Verbose mode (-v, -vv, -vvv, etc)
|
||||
#[structopt(short = "v", long = "verbose", parse(from_occurrences))]
|
||||
verbose: usize,
|
||||
}
|
||||
|
||||
#[tokio::main]
|
||||
|
@ -29,7 +29,7 @@ pub async fn main() -> Result<(), Error> {
|
|||
stderrlog::new()
|
||||
.module(module_path!())
|
||||
.quiet(params.quiet)
|
||||
.verbosity(1)
|
||||
.verbosity(params.verbose)
|
||||
.init()?;
|
||||
|
||||
eprintln!("Crawling...");
|
||||
|
@ -56,10 +56,10 @@ struct TotalStats {
|
|||
users_active_week: i64,
|
||||
users_active_month: i64,
|
||||
users_active_halfyear: i64,
|
||||
instance_details: Vec<CrawlResult2>,
|
||||
instance_details: Vec<InstanceDetails>,
|
||||
}
|
||||
|
||||
fn aggregate(instance_details: Vec<CrawlResult2>) -> TotalStats {
|
||||
fn aggregate(instance_details: Vec<InstanceDetails>) -> TotalStats {
|
||||
let mut online_users = 0;
|
||||
let mut total_users = 0;
|
||||
let mut users_active_day = 0;
|
||||
|
|
|
@ -1,47 +0,0 @@
|
|||
use reqwest::Url;
|
||||
use serde::{Deserialize, Serialize};
|
||||
|
||||
#[derive(Deserialize, Debug)]
|
||||
pub struct NodeInfoWellKnown {
|
||||
pub links: Vec<NodeInfoWellKnownLinks>,
|
||||
}
|
||||
|
||||
#[derive(Deserialize, Debug)]
|
||||
pub struct NodeInfoWellKnownLinks {
|
||||
pub rel: Url,
|
||||
pub href: Url,
|
||||
}
|
||||
|
||||
#[derive(Deserialize, Debug, Clone)]
|
||||
#[serde(rename_all = "camelCase")]
|
||||
pub struct NodeInfo {
|
||||
pub version: String,
|
||||
pub software: NodeInfoSoftware,
|
||||
pub protocols: Vec<String>,
|
||||
pub usage: NodeInfoUsage,
|
||||
pub open_registrations: bool,
|
||||
}
|
||||
|
||||
#[derive(Deserialize, Debug, Clone)]
|
||||
pub struct NodeInfoSoftware {
|
||||
pub name: String,
|
||||
pub version: String,
|
||||
}
|
||||
|
||||
#[derive(Deserialize, Serialize, Debug, Clone, Default)]
|
||||
#[serde(rename_all = "camelCase", default)]
|
||||
pub struct NodeInfoUsage {
|
||||
pub users: NodeInfoUsers,
|
||||
#[serde(rename(deserialize = "localPosts"))]
|
||||
pub posts: i64,
|
||||
#[serde(rename(deserialize = "localComments"))]
|
||||
pub comments: i64,
|
||||
}
|
||||
|
||||
#[derive(Deserialize, Serialize, Debug, Clone, Default)]
|
||||
#[serde(rename_all = "camelCase", default)]
|
||||
pub struct NodeInfoUsers {
|
||||
pub total: i64,
|
||||
pub active_halfyear: i64,
|
||||
pub active_month: i64,
|
||||
}
|
Loading…
Reference in a new issue