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"
|
edition = "2018"
|
||||||
|
|
||||||
[dependencies]
|
[dependencies]
|
||||||
reqwest = { version = "0.11.13", default-features = false, features = ["json", "rustls-tls"] }
|
reqwest = { version = "0.11.10", default-features = false, features = ["json", "rustls-tls"] }
|
||||||
serde = { version = "1.0.149", features = ["derive"] }
|
serde = { version = "1.0.137", features = ["derive"] }
|
||||||
anyhow = "1.0.66"
|
anyhow = "1.0.57"
|
||||||
tokio = { version = "1.22.0", features = ["macros", "rt-multi-thread"] }
|
tokio = { version = "1.18.1", features = ["macros", "rt-multi-thread"] }
|
||||||
futures = "0.3.25"
|
futures = "0.3.21"
|
||||||
serde_json = "1.0.89"
|
serde_json = "1.0.81"
|
||||||
semver = "1.0.14"
|
semver = "1.0.9"
|
||||||
once_cell = "1.16.0"
|
once_cell = "1.10.0"
|
||||||
lemmy_api_common = "=0.16.0"
|
lemmy_api_common = "0.16.0"
|
||||||
lemmy_db_schema = "=0.16.0"
|
|
||||||
async-recursion = "1.0.0"
|
async-recursion = "1.0.0"
|
||||||
log = "0.4.17"
|
log = "0.4.17"
|
||||||
derive-new = "0.5.9"
|
derive-new = "0.5.9"
|
||||||
stderrlog = "0.5.4"
|
stderrlog = "0.5.1"
|
||||||
structopt = "0.3.26"
|
structopt = "0.3.26"
|
||||||
|
|
112
src/crawl.rs
112
src/crawl.rs
|
@ -1,12 +1,12 @@
|
||||||
use crate::node_info::{NodeInfo, NodeInfoWellKnown};
|
|
||||||
use crate::CLIENT;
|
use crate::CLIENT;
|
||||||
use anyhow::{anyhow, Error};
|
use crate::REQUEST_TIMEOUT;
|
||||||
|
use anyhow::Error;
|
||||||
use async_recursion::async_recursion;
|
use async_recursion::async_recursion;
|
||||||
use futures::future::join_all;
|
use futures::future::join_all;
|
||||||
use lemmy_api_common::site::GetSiteResponse;
|
use lemmy_api_common::site::GetSiteResponse;
|
||||||
use log::debug;
|
use log::debug;
|
||||||
use reqwest::Url;
|
|
||||||
use semver::Version;
|
use semver::Version;
|
||||||
|
use serde::Serialize;
|
||||||
use std::collections::HashSet;
|
use std::collections::HashSet;
|
||||||
use std::ops::Deref;
|
use std::ops::Deref;
|
||||||
use std::sync::Arc;
|
use std::sync::Arc;
|
||||||
|
@ -27,16 +27,15 @@ pub struct CrawlParams {
|
||||||
crawled_instances: Arc<Mutex<HashSet<String>>>,
|
crawled_instances: Arc<Mutex<HashSet<String>>>,
|
||||||
}
|
}
|
||||||
|
|
||||||
#[derive(Debug)]
|
#[derive(Serialize, Debug)]
|
||||||
pub struct CrawlResult {
|
pub struct InstanceDetails {
|
||||||
pub domain: String,
|
pub domain: String,
|
||||||
pub node_info: NodeInfo,
|
pub site_info: GetSiteResponse,
|
||||||
pub site_info: Option<GetSiteResponse>,
|
|
||||||
}
|
}
|
||||||
|
|
||||||
impl CrawlJob {
|
impl CrawlJob {
|
||||||
#[async_recursion]
|
#[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
|
// need to acquire and release mutex before recursing, otherwise it will deadlock
|
||||||
{
|
{
|
||||||
let mut crawled_instances = self.params.crawled_instances.deref().lock().await;
|
let mut crawled_instances = self.params.crawled_instances.deref().lock().await;
|
||||||
|
@ -53,83 +52,46 @@ impl CrawlJob {
|
||||||
return vec![];
|
return vec![];
|
||||||
}
|
}
|
||||||
|
|
||||||
debug!(
|
debug!("Starting crawl for {}, distance {}", &self.domain, &self.current_distance);
|
||||||
"Starting crawl for {}, distance {}",
|
let site_info = match self.fetch_instance_details().await {
|
||||||
&self.domain, &self.current_distance
|
|
||||||
);
|
|
||||||
let (node_info, site_info) = match self.fetch_instance_details().await {
|
|
||||||
Ok(o) => o,
|
Ok(o) => o,
|
||||||
Err(e) => return vec![Err(e)],
|
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 {
|
if site_info.1 < self.params.min_lemmy_version {
|
||||||
match Version::parse(&site_info.version) {
|
return vec![];
|
||||||
Ok(version) => {
|
|
||||||
if version < self.params.min_lemmy_version {
|
|
||||||
return vec![Ok(crawl_result)];
|
|
||||||
}
|
|
||||||
}
|
|
||||||
Err(e) => return vec![Err(e.into())],
|
|
||||||
}
|
|
||||||
|
|
||||||
let mut result = vec![];
|
|
||||||
if let Some(federated) = &site_info.federated_instances {
|
|
||||||
for domain in federated.linked.iter() {
|
|
||||||
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>> =
|
|
||||||
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
|
|
||||||
} else {
|
|
||||||
vec![Ok(crawl_result)]
|
|
||||||
}
|
}
|
||||||
|
|
||||||
|
let mut result = vec![];
|
||||||
|
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());
|
||||||
|
result.push(crawl_job.crawl());
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
let mut result2: Vec<Result<InstanceDetails, Error>> =
|
||||||
|
join_all(result).await.into_iter().flatten().collect();
|
||||||
|
debug!("Successfully finished crawl for {}", &self.domain);
|
||||||
|
result2.push(Ok(InstanceDetails {
|
||||||
|
domain: self.domain,
|
||||||
|
site_info: site_info.0,
|
||||||
|
}));
|
||||||
|
|
||||||
|
result2
|
||||||
}
|
}
|
||||||
|
|
||||||
async fn fetch_instance_details(&self) -> Result<(NodeInfo, Option<GetSiteResponse>), Error> {
|
async fn fetch_instance_details(&self) -> Result<(GetSiteResponse, Version), Error> {
|
||||||
let rel_node_info: Url = Url::parse("http://nodeinfo.diaspora.software/ns/schema/2.0")
|
let site_info_url = format!("https://{}/api/v3/site", &self.domain);
|
||||||
.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?;
|
|
||||||
|
|
||||||
let site_info = CLIENT
|
let site_info = CLIENT
|
||||||
.get(&format!("https://{}/api/v3/site", &self.domain))
|
.get(&site_info_url)
|
||||||
|
.timeout(REQUEST_TIMEOUT)
|
||||||
.send()
|
.send()
|
||||||
.await?
|
.await?
|
||||||
.json::<GetSiteResponse>()
|
.json::<GetSiteResponse>()
|
||||||
.await
|
.await?;
|
||||||
.ok();
|
let version = Version::parse(&site_info.version)?;
|
||||||
Ok((node_info, site_info))
|
Ok((site_info, version))
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
90
src/lib.rs
90
src/lib.rs
|
@ -1,46 +1,28 @@
|
||||||
#[macro_use]
|
#[macro_use]
|
||||||
extern crate derive_new;
|
extern crate derive_new;
|
||||||
|
|
||||||
use crate::crawl::{CrawlJob, CrawlParams, CrawlResult};
|
use crate::crawl::{CrawlJob, CrawlParams, InstanceDetails};
|
||||||
use crate::node_info::{NodeInfo, NodeInfoUsage, NodeInfoUsers};
|
|
||||||
use anyhow::Error;
|
use anyhow::Error;
|
||||||
use futures::future::join_all;
|
use futures::future::join_all;
|
||||||
use lemmy_api_common::site::GetSiteResponse;
|
|
||||||
use log::warn;
|
|
||||||
use once_cell::sync::Lazy;
|
use once_cell::sync::Lazy;
|
||||||
use reqwest::{Client, ClientBuilder};
|
use reqwest::Client;
|
||||||
use semver::Version;
|
use semver::Version;
|
||||||
use serde::Serialize;
|
|
||||||
use std::collections::HashSet;
|
use std::collections::HashSet;
|
||||||
use std::sync::Arc;
|
use std::sync::Arc;
|
||||||
use std::time::Duration;
|
use std::time::Duration;
|
||||||
use tokio::sync::Mutex;
|
use tokio::sync::Mutex;
|
||||||
|
|
||||||
pub mod crawl;
|
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(|| {
|
static CLIENT: Lazy<Client> = Lazy::new(Client::default);
|
||||||
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>,
|
|
||||||
}
|
|
||||||
|
|
||||||
pub async fn start_crawl(
|
pub async fn start_crawl(
|
||||||
start_instances: Vec<String>,
|
start_instances: Vec<String>,
|
||||||
exclude_domains: Vec<String>,
|
exclude_domains: Vec<String>,
|
||||||
max_distance: i32,
|
max_distance: i32,
|
||||||
) -> Result<Vec<CrawlResult2>, Error> {
|
) -> Result<Vec<InstanceDetails>, Error> {
|
||||||
let params = Arc::new(CrawlParams::new(
|
let params = Arc::new(CrawlParams::new(
|
||||||
min_lemmy_version().await?,
|
min_lemmy_version().await?,
|
||||||
exclude_domains,
|
exclude_domains,
|
||||||
|
@ -53,29 +35,25 @@ pub async fn start_crawl(
|
||||||
jobs.push(job.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
|
.await
|
||||||
.into_iter()
|
.into_iter()
|
||||||
.flatten()
|
.flatten()
|
||||||
.inspect(|r| {
|
.filter_map(|r| r.ok())
|
||||||
if let Err(e) = r {
|
|
||||||
warn!("{}", e)
|
|
||||||
}
|
|
||||||
})
|
|
||||||
.filter_map(Result::ok)
|
|
||||||
.collect();
|
.collect();
|
||||||
let mut crawl_results = calculate_federated_site_aggregates(crawl_results)?;
|
|
||||||
|
|
||||||
// Sort by active monthly users descending
|
// Sort by active monthly users descending
|
||||||
crawl_results.sort_unstable_by_key(|i| {
|
instance_details.sort_unstable_by_key(|i| {
|
||||||
i.site_info
|
i.site_info
|
||||||
.site_view
|
.site_view
|
||||||
.as_ref()
|
.as_ref()
|
||||||
.map(|s| s.counts.users_active_month)
|
.map(|s| s.counts.users_active_month)
|
||||||
.unwrap_or(0)
|
.unwrap_or(0)
|
||||||
});
|
});
|
||||||
crawl_results.reverse();
|
instance_details.reverse();
|
||||||
Ok(crawl_results)
|
|
||||||
|
Ok(instance_details)
|
||||||
}
|
}
|
||||||
|
|
||||||
/// calculate minimum allowed lemmy version based on current version. in case of current version
|
/// 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;
|
version.minor -= 1;
|
||||||
Ok(version)
|
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 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 serde::Serialize;
|
||||||
use structopt::StructOpt;
|
use structopt::StructOpt;
|
||||||
|
|
||||||
#[derive(StructOpt, Debug)]
|
#[derive(StructOpt, Debug)]
|
||||||
#[structopt()]
|
#[structopt()]
|
||||||
struct Parameters {
|
struct Parameters {
|
||||||
#[structopt(short, long, use_delimiter = true, default_value = "lemmy.ml")]
|
#[structopt(short, long, default_value = "lemmy.ml")]
|
||||||
start_instances: Vec<String>,
|
start_instances: Vec<String>,
|
||||||
#[structopt(
|
#[structopt(short, long, default_value = "ds9.lemmy.ml, enterprise.lemmy.ml, voyager.lemmy.ml, test.lemmy.ml")]
|
||||||
short,
|
|
||||||
long,
|
|
||||||
use_delimiter = true,
|
|
||||||
default_value = "ds9.lemmy.ml,enterprise.lemmy.ml,voyager.lemmy.ml,test.lemmy.ml"
|
|
||||||
)]
|
|
||||||
exclude_instances: Vec<String>,
|
exclude_instances: Vec<String>,
|
||||||
#[structopt(short, long, default_value = "20")]
|
#[structopt(short, long, default_value = "20")]
|
||||||
max_crawl_distance: i32,
|
max_crawl_distance: i32,
|
||||||
|
|
||||||
/// Silence all output
|
/// Silence all output
|
||||||
#[structopt(short, long)]
|
#[structopt(short, long)]
|
||||||
quiet: bool,
|
quiet: bool,
|
||||||
|
/// Verbose mode (-v, -vv, -vvv, etc)
|
||||||
|
#[structopt(short = "v", long = "verbose", parse(from_occurrences))]
|
||||||
|
verbose: usize,
|
||||||
}
|
}
|
||||||
|
|
||||||
#[tokio::main]
|
#[tokio::main]
|
||||||
|
@ -29,7 +29,7 @@ pub async fn main() -> Result<(), Error> {
|
||||||
stderrlog::new()
|
stderrlog::new()
|
||||||
.module(module_path!())
|
.module(module_path!())
|
||||||
.quiet(params.quiet)
|
.quiet(params.quiet)
|
||||||
.verbosity(1)
|
.verbosity(params.verbose)
|
||||||
.init()?;
|
.init()?;
|
||||||
|
|
||||||
eprintln!("Crawling...");
|
eprintln!("Crawling...");
|
||||||
|
@ -56,10 +56,10 @@ struct TotalStats {
|
||||||
users_active_week: i64,
|
users_active_week: i64,
|
||||||
users_active_month: i64,
|
users_active_month: i64,
|
||||||
users_active_halfyear: 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 online_users = 0;
|
||||||
let mut total_users = 0;
|
let mut total_users = 0;
|
||||||
let mut users_active_day = 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