use super::*;
impl Scraper {
pub fn new(config: ScraperConfig) -> Result<Self, ScraperError> {
cpu::install();
let browser_private_network = config.fetcher.ssrf.allow_private_network;
let browser_allow = browser_allow_list(&config.fetcher.ssrf)?;
let mut egress = Vec::new();
for (name, proxy) in &config.egress {
let mut fc = config.fetcher.clone();
if !proxy.is_empty() {
fc.proxy = Some(proxy.clone());
}
let class = config.egress_classes.get(name).copied().unwrap_or_default();
egress.push((
name.clone(),
class,
Arc::new(Fetcher::new_egress(fc).map_err(ScraperError::Fetcher)?),
));
}
let fetcher = Arc::new(Fetcher::new(config.fetcher).map_err(ScraperError::Fetcher)?);
let llm = config
.llm
.map(pagesift_llm::ChatClient::new)
.transpose()
.map_err(ScraperError::Llm)?;
let rate = &config.policy.rate;
let initial_conc = rate
.global_ceiling
.map_or(rate.global_concurrency, |c| rate.global_concurrency.min(c))
.max(1);
let crawl_slots = pagesift_policy::AdaptiveConcurrencyGate::new(initial_conc);
Ok(Self {
crawl_slots,
policy: Arc::new({
let wait_price = fetcher.config().wait_price.clone();
let mut p = Policy::new(fetcher, config.policy);
for (name, class, f) in egress {
p.add_egress_class(name, class, f);
}
p.limiter().publish_wait_price(wait_price);
p
}),
store: config.store,
audit: config.audit,
llm,
search: config.search,
browsers: {
let chain = pagesift_cdp::engine_chain_with_allow_list(
&config.browser,
browser_private_network,
&browser_allow,
);
log_engine_chain(&chain);
chain.into_iter().map(|c| (c, OnceCell::new())).collect()
},
launch_failed: Vec::new(),
launch_notes: Vec::new(),
launch_broken: Default::default(),
browser_private_network,
browser_allow,
learner: crate::learn::HostLearner::default(),
planner: super::planner::RenderPlanner::default(),
render_stats: super::render_stats::RenderStats::new(),
pages: std::sync::OnceLock::new(),
})
.map(|mut s| {
s.launch_failed = (0..s.browsers.len()).map(|_| Default::default()).collect();
s.launch_notes = (0..s.browsers.len()).map(|_| Default::default()).collect();
if let Some(store) = &s.store {
super::planner::load(&s.planner, store);
}
s
})
}
pub(super) async fn browser(&self, i: usize) -> Option<Arc<BrowserPool>> {
let (cfg, cell) = self.browsers.get(i)?;
cell.get_or_init(|| async {
let guard: Arc<dyn RequestGuard> = if self.browser_private_network {
Arc::new(SchemeGuard)
} else if self.browser_allow.is_empty() {
Arc::new(PrivateNetworkGuard)
} else {
Arc::new(AllowListGuard::new(self.browser_allow.clone()))
};
match BrowserPool::new(cfg.clone(), guard).await {
Ok(pool) => Some(Arc::new(pool)),
Err(e) => {
let engine = cfg.engine.map_or("browser", |k| k.name());
*self.launch_notes[i].lock().unwrap() = Some(format!("{engine}: {e}"));
if !matches!(e, pagesift_cdp::Error::NoBrowserFound) {
self.launch_broken
.store(true, std::sync::atomic::Ordering::Relaxed);
}
tracing::warn!("{engine} rendering unavailable: {e}");
None
}
}
})
.await
.clone()
}
pub(super) async fn has_browser(&self) -> bool {
for i in 0..self.browsers.len() {
if self.browser(i).await.is_some() {
return true;
}
}
false
}
pub(super) fn browser_launch_failed(&self) -> bool {
self.launch_broken
.load(std::sync::atomic::Ordering::Relaxed)
}
pub(super) fn browser_unavailable_reason(&self) -> String {
let notes: Vec<String> = self
.launch_notes
.iter()
.filter_map(|n| n.lock().unwrap().clone())
.collect();
if notes.is_empty() {
return "no browser engine found (run `pagesift doctor`)".into();
}
let root = notes.iter().any(|n| n.contains("as root"));
let mut reason = format!(
"no browser engine could start ({}); run `pagesift doctor` for details",
notes.join("; ")
);
if root && !notes.iter().any(|n| n.contains("unprivileged user")) {
reason.push_str(
" — pagesift is running as root: the Chrome sandbox does not work as root and pagesift will not disable it; run as an unprivileged user",
);
}
reason
}
pub async fn probe_engines(&self) -> Vec<EngineProbe> {
let mut out = Vec::new();
for i in 0..self.browsers.len() {
let cfg = &self.browsers[i].0;
let engine = cfg.engine.map_or("chrome", |k| k.name());
let location = match &cfg.source {
pagesift_cdp::BrowserSource::Executable(p) => p.display().to_string(),
pagesift_cdp::BrowserSource::Remote(u) => u.to_string(),
pagesift_cdp::BrowserSource::Auto => "auto".into(),
};
let result = match self.browser(i).await {
Some(pool) => pool.probe().await.map_err(|e| e.to_string()),
None => Err(self.launch_notes[i]
.lock()
.unwrap()
.clone()
.unwrap_or_else(|| "could not start".into())),
};
out.push(EngineProbe {
engine,
location,
result,
});
}
out
}
pub async fn probe_browsers(&self) {
for i in 0..self.browsers.len() {
let Some(pool) = self.browser(i).await else {
continue;
};
let engine = pool.engine().name();
match pool.probe().await {
Ok(()) => tracing::info!(engine, "browser engine ready"),
Err(e) => self.note_launch_failure(i, engine, &e),
}
}
}
pub(super) fn note_launch_failure(&self, i: usize, engine: &str, e: &pagesift_cdp::Error) {
let first = self.launch_failed[i]
.lock()
.unwrap()
.replace(Instant::now())
.is_none();
if first {
let msg = e.to_string();
*self.launch_notes[i].lock().unwrap() = Some(format!("{engine}: {msg}"));
let hint = if msg.contains("sandbox") {
": the browser sandbox needs unprivileged user namespaces, which this host or container \
does not allow (check kernel.apparmor_restrict_unprivileged_userns and the container's \
seccomp/AppArmor profile); pagesift does not disable the sandbox"
} else {
""
};
tracing::warn!(
engine,
retry_after_secs = LAUNCH_RETRY.as_secs(),
"{engine} fallback unavailable: {msg}{hint}"
);
}
}
pub async fn shutdown(&self) {
if let Some(store) = &self.store
&& let Err(e) = super::planner::save(&self.planner, store)
{
tracing::warn!("saving the render planner failed: {e}");
}
for (_, cell) in &self.browsers {
if let Some(Some(pool)) = cell.get() {
pool.shutdown().await;
}
}
}
pub async fn scrape(&self, url: &str, opts: &ScrapeOptions) -> Result<Document, ApiError> {
let mode = self
.policy
.robots_mode(RequestKind::Scrape, opts.ignore_robots_txt)
.map_err(policy_error)?;
self.scrape_with(url, opts, mode).await
}
pub fn audit_robots_ignored(&self, kind: RequestKind, mode: RobotsMode, url: &str) {
if mode == RobotsMode::Ignore
&& let Some(audit) = &self.audit
{
let request = match kind {
RequestKind::Scrape => "scrape",
RequestKind::Crawl => "crawl",
RequestKind::Map => "map",
};
audit.record_robots_ignored(request, url, self.policy.robots_ignore_reason(kind));
}
}
pub fn audit(&self) -> Option<&Arc<crate::audit::Audit>> {
self.audit.as_ref()
}
pub fn policy(&self) -> &Policy {
&self.policy
}
pub fn route_pages(&self, runner: Arc<dyn crate::PageRunner>) -> bool {
self.pages.set(runner).is_ok()
}
pub fn page_runner(&self) -> Option<&Arc<dyn crate::PageRunner>> {
self.pages.get()
}
pub fn planner(&self) -> &super::planner::RenderPlanner {
&self.planner
}
pub fn learner(&self) -> &crate::learn::HostLearner {
&self.learner
}
pub fn engine_count(&self) -> usize {
self.browsers.len()
}
pub fn network_stage_snapshots(&self) -> Vec<pagesift_net::StageSnapshot> {
self.policy.fetcher().stage_snapshots()
}
pub fn transport_phase_snapshots(&self) -> Vec<pagesift_net::PhaseSnapshot> {
self.policy.fetcher().phase_snapshots()
}
pub fn crawl_slots(&self) -> &pagesift_policy::AdaptiveConcurrencyGate {
self.sync_crawl_slots();
&self.crawl_slots
}
pub fn concurrency(&self) -> usize {
if let Some(n) = self.pages.get().and_then(|r| r.capacity()) {
return n.max(1);
}
self.policy.limiter().global_limit()
}
pub fn observe_crawl(&self, _queue_wait: Duration, _failed: bool) {
self.sync_crawl_slots();
}
fn sync_crawl_slots(&self) {
let target = self.concurrency();
if self.crawl_slots.limit() != target {
self.crawl_slots.set_limit(target);
}
}
pub async fn scrape_with(
&self,
url: &str,
opts: &ScrapeOptions,
mode: RobotsMode,
) -> Result<Document, ApiError> {
self.scrape_audited("scrape", url, opts, mode).await
}
pub async fn scrape_audited(
&self,
kind: &str,
url: &str,
opts: &ScrapeOptions,
mode: RobotsMode,
) -> Result<Document, ApiError> {
let Some(audit) = &self.audit else {
return self.scrape_inner(Source::Url(url, mode), opts).await;
};
let started = Instant::now();
let mut result = self.scrape_inner(Source::Url(url, mode), opts).await;
let raw = result.as_ref().ok().and_then(|d| d.raw_html.as_deref());
if mode == RobotsMode::Ignore {
let request = if kind == "scrape" {
RequestKind::Scrape
} else {
RequestKind::Crawl
};
let reason = self.policy.robots_ignore_reason(request);
audit.record_scrape_ignoring_robots(
kind,
url,
opts,
&result,
raw,
started.elapsed(),
reason,
);
} else {
audit.record_scrape(kind, url, opts, &result, raw, started.elapsed());
}
if !opts.wants("rawHtml")
&& let Ok(doc) = &mut result
{
doc.raw_html = None;
}
result
}
pub(super) async fn scrape_inner(
&self,
source: Source<'_>,
opts: &ScrapeOptions,
) -> Result<Document, ApiError> {
tag_filter(opts)?;
if opts.export_images {
return self.scrape_exporting_images(source, opts).await;
}
let mut result = self.scrape_document(source, opts).await;
if let Ok(doc) = &mut result {
crate::notice::fill_warning(doc);
}
result
}
async fn scrape_exporting_images(
&self,
source: Source<'_>,
opts: &ScrapeOptions,
) -> Result<Document, ApiError> {
let mode = match &source {
Source::Url(_, mode) => *mode,
Source::Upload(_) => RobotsMode::Respect,
};
let mut inner = opts.clone();
if !inner.wants("images") {
inner
.formats
.push(pagesift_api::types::Format::new("images"));
}
inner.max_age = Some(0);
inner.store_in_cache = false;
let mut result = self.scrape_document(source, &inner).await;
if let Ok(doc) = &mut result {
self.export_web_images(doc, mode).await;
if !opts.wants("images") {
doc.images = None;
}
crate::notice::fill_warning(doc);
}
result
}
async fn scrape_document(
&self,
source: Source<'_>,
opts: &ScrapeOptions,
) -> Result<Document, ApiError> {
let wants_llm = opts.wants("summary") || opts.wants("json");
let tracking = opts
.formats
.iter()
.find(|f| f.kind == "changeTracking")
.map(crate::change::Tracking::parse)
.transpose()?;
if !wants_llm
&& tracking.is_none()
&& opts.query_filter.is_none()
&& opts.chunking.is_none()
{
return self.base_document(source, opts).await;
}
let url = match &source {
Source::Url(url, _) => Some(*url),
Source::Upload(_) => None,
};
if tracking.is_some() && url.is_none() {
return Err(ApiError::unsupported(
"changeTracking needs a URL: it is not available for uploaded files",
));
}
if (wants_llm || tracking.as_ref().is_some_and(|t| t.json)) && self.llm.is_none() {
return Err(ApiError::unsupported(
"summary and json formats (and the json mode of changeTracking) need an LLM: set [llm] in pagesift.toml",
));
}
if tracking.is_some() && self.store.is_none() {
return Err(ApiError::unsupported(
"changeTracking keeps history in the store: enable [cache] in pagesift.toml",
));
}
let mut inner = opts.clone();
inner
.formats
.retain(|f| f.kind != "summary" && f.kind != "json");
let wants_md = inner.wants("markdown");
if !wants_md {
inner
.formats
.push(pagesift_api::types::Format::new("markdown"));
}
if tracking.is_some() && inner.max_age.is_none() {
inner.max_age = Some(0);
}
let mut doc = self.base_document(source, &inner).await?;
if let Some(q) = &opts.query_filter {
apply_query_filter(&mut doc, q).await?;
}
if let Some(c) = &opts.chunking {
doc.chunks = Some(chunk_markdown(
doc.markdown.as_deref().unwrap_or_default(),
c,
)?);
}
let md = doc.markdown.clone().unwrap_or_default();
if let Some(llm) = self.llm.as_ref().filter(|_| wants_llm) {
if opts.wants("summary") {
let summary = pagesift_llm::summarize(llm, &md, &Default::default())
.await
.map_err(llm_error)?;
doc.summary = Some(summary);
}
if let Some(f) = opts.formats.iter().find(|f| f.kind == "json") {
let schema = f
.options
.get("schema")
.cloned()
.unwrap_or_else(|| serde_json::json!({ "type": "object" }));
let eo = pagesift_llm::ExtractOptions {
prompt: f
.options
.get("prompt")
.and_then(|p| p.as_str())
.map(String::from),
};
doc.json = Some(
pagesift_llm::extract_json(llm, &md, &schema, &eo)
.await
.map_err(llm_error)?,
);
}
}
if let (Some(t), Some(store), Some(url)) = (&tracking, &self.store, url) {
self.track_changes(store, t, url, &mut doc).await?;
}
if !wants_md {
doc.markdown = None;
}
Ok(doc)
}
async fn base_document(
&self,
source: Source<'_>,
opts: &ScrapeOptions,
) -> Result<Document, ApiError> {
match source {
Source::Url(url, mode) => self.scrape_cached(url, opts, mode).await,
Source::Upload(upload) => self.parse_upload(upload, opts).await,
}
}
pub(super) async fn track_changes(
&self,
store: &pagesift_store::Store,
t: &crate::change::Tracking,
url: &str,
doc: &mut Document,
) -> Result<(), ApiError> {
use crate::change::{Current, Previous};
let ext = doc.metadata.pagesift.get_or_insert_with(Default::default);
let current = if ext.not_modified {
Current::NotModified
} else if matches!(doc.metadata.status_code, 404 | 410) {
Current::Gone
} else {
Current::Page(doc.markdown.as_deref().unwrap_or_default())
};
let mut outcome = crate::change::track(store, t, url, current).await?;
ext.warnings.append(&mut outcome.warnings);
if let (true, "changed", Previous::Page(prev), Some(llm)) = (
t.json,
outcome.tracking.change_status.as_str(),
&outcome.previous,
&self.llm,
) {
let schema = t
.schema
.clone()
.unwrap_or_else(|| serde_json::json!({ "type": "object" }));
let eo = pagesift_llm::ExtractOptions {
prompt: t.prompt.clone(),
};
let cur = doc.markdown.as_deref().unwrap_or_default();
let (before, after) = futures_util::future::try_join(
pagesift_llm::extract_json(llm, prev, &schema, &eo),
pagesift_llm::extract_json(llm, cur, &schema, &eo),
)
.await
.map_err(llm_error)?;
outcome.tracking.json = Some(crate::change::json_changes(&before, &after));
}
doc.change_tracking = Some(outcome.tracking);
Ok(())
}
}
fn browser_allow_list(ssrf: &pagesift_net::SsrfPolicy) -> Result<NetworkAllowList, ScraperError> {
let mut allow = NetworkAllowList::default();
for host in &ssrf.allow_hosts {
allow.allow_host(host.clone());
}
for ip in &ssrf.allow_ips {
allow.allow_ip(*ip);
}
for cidr in &ssrf.allow_cidrs {
let (net, prefix) = match cidr {
pagesift_net::IpCidr::V4 { net, prefix } => (std::net::IpAddr::V4(*net), *prefix),
pagesift_net::IpCidr::V6 { net, prefix } => (std::net::IpAddr::V6(*net), *prefix),
};
allow
.allow_cidr_parts(net, prefix)
.map_err(|e| ScraperError::Fetcher(pagesift_net::Error::Config(e)))?;
}
Ok(allow)
}