Examples¶
Runnable programs, ordered so that reading them in sequence explains the design. Each
one is self-contained and lives in
examples/.
Three of them need an extra:
pip install "lncrawl-scraper[cdp]" # 03
pip install "lncrawl-scraper[botauth]" # 07
pip install "lncrawl-scraper[image]" # 10, for get_image
01 · Quickstart¶
The shortest useful program.
Nothing is configured. The default transport already reproduces a real browser's TLS and HTTP/2 fingerprint, which is what clears the four transport layers that most protected sites stop at.
uv run python examples/01_quickstart.py
from scraper import Scraper
with Scraper(origin="https://example.com") as scraper:
soup = scraper.get_soup("https://example.com/")
print(soup.select_one("h1").text)
# Selection never returns None, so chained access is always safe.
print(repr(soup.select_one(".does-not-exist").text))
# Links a person could actually click: hidden and nofollow anchors are dropped,
# which is the whole defence against a decoy maze.
for link in scraper.links(soup):
print(link.url, "|", link.text)
02 · The model¶
The model this library reasons with, printed out.
Read this before configuring anything. Which layer is binding decides what is worth changing, and what that layer reads decides whether it can be satisfied at all.
uv run python examples/02_the_model.py
from scraper import LAYERS, Layer, weakest
from scraper.layers import TRANSPORT_LAYERS, marginal_gain
print(f"{'#':>3} {'layer':34} {'reads':8} {'do':11} ")
print("-" * 62)
for layer, info in LAYERS.items():
print(f"{layer.value:>3} {info.title:34} {info.trait.value:8} {info.stance.value:11}")
print()
print("Layers 2-5 are one barrier, not four:", sorted(int(x) for x in TRANSPORT_LAYERS))
print("One impersonation profile passes all four, or none of them.")
print()
print("The bound: admission is limited by the weakest layer.")
odds = {
Layer.IP_REPUTATION: 0.05, # a datacenter address
Layer.TLS_FINGERPRINT: 0.99, # a perfect Chrome profile
Layer.BEHAVIOURAL: 0.60,
}
binding, chance = weakest(odds) # type: ignore[misc]
print(f" binding layer: {binding} at {chance:.0%}")
# The consequence people spend the most time ignoring.
print(f" perfecting TLS gains: {marginal_gain(odds, Layer.TLS_FINGERPRINT, 1.0):.0%}")
print(f" fixing the address gains: {marginal_gain(odds, Layer.IP_REPUTATION, 0.9):.0%}")
03 · Challenged site¶
A site that serves a JavaScript challenge.
The pattern is solve-once-and-reuse, and the reason it works is that a clearance is bound to the address, User-Agent and TLS fingerprint that earned it. So the browser runs once, its exact User-Agent is adopted, and everything after that is an ordinary cheap request on the same identity.
Needs the cdp extra and a Chrome: pip install lncrawl-scraper[cdp]
uv run python examples/03_challenged_site.py
import logging
from scraper import CdpSolver, Scraper, ScraperConfig
from scraper.exceptions import Exhausted
logging.basicConfig(level=logging.INFO, format="%(levelname)s %(name)s: %(message)s")
config = ScraperConfig(
# headless=False is the default, and not because headless cannot clear — it can,
# measured. A visible window is the one a person can reach into and solve by hand,
# which is worth having wherever there is a display. Set headless=True on a server.
browser=CdpSolver(),
)
TARGET = "https://nowsecure.nl/"
with Scraper(config=config) as scraper:
try:
for page in range(3):
response = scraper.get(TARGET)
print(page, response.status_code, len(response.content))
except Exhausted as exc:
# The message names the layer that ended the attempt and what would move it.
print("stopped at", exc.layer, "-", exc.detail)
print()
print(scraper.explain(TARGET))
04 · Addresses¶
Where the packets come from, which is layer 1.
The difficulty here is economic, not technical. The address is chosen freely, but its reputation accrued over time and can be rented, never fabricated. Declare the kind honestly: claiming MOBILE for a datacenter range does not change what the reputation database thinks, it only stops this library from telling you that layer 1 is the reason nothing works.
uv run python examples/04_addresses.py
from scraper import ExitKind, ExitSpec, Scraper, ScraperConfig, TorPoolSpec
from scraper.exceptions import Exhausted
config = ScraperConfig(
exits=[
# Best kind first is not required — the pool sorts by kind.
ExitSpec(
url="http://user:pass@mobile.provider.test:8000",
kind=ExitKind.MOBILE,
label="carrier",
),
ExitSpec(
url="http://user:pass@residential.provider.test:8000",
kind=ExitKind.RESIDENTIAL,
),
# A tor-pool endpoint: many Tor instances behind one sticky port. Reported as
# TOR because exit lists are published, so it clears none of layer 1. Right
# for a site that does not score addresses, wrong for one that does.
TorPoolSpec(api_url="http://127.0.0.1:8080", token="tp_...."),
],
# Concurrent sessions per address is itself a behavioural signal, so this stays
# in the low single digits. Values above 3 are clamped.
max_sessions_per_exit=2,
)
with Scraper(config=config) as scraper:
print("best kind on offer:", scraper.exits.best_kind.value)
print("clears layer 1:", scraper.exits.reach())
# An address is leased per origin and held. Rotation happens on evidence, never
# on a timer, because a clearance and the accumulated history are both bound to
# the address.
lease = scraper.exits.lease("example.com")
print("leased:", lease.exit_id, "->", lease.proxies)
try:
scraper.get("https://example.com/")
except Exhausted as exc:
# With only Tor configured this is the message you get, and it is the useful
# one: rotating between published ranges cannot help.
print(exc.detail)
05 · Behaviour¶
The hardest layer, addressed by not trying to defeat it.
A per-zone behavioural model reads accumulated, non-portable history: timing regularity, navigation chains, session age and depth, concurrent sessions per address. None of that can be presented on demand, so the only thing that works is to behave the way the model expects and let the history accrue.
uv run python examples/05_behaviour.py
import statistics
from scraper import Pacer, PacingPolicy, Scraper, ScraperConfig, SharedState
policy = PacingPolicy(
interval=4.0, # target mean, not a floor
shape=2.5, # low shape = long right tail, which is what browsing looks like
pause_chance=0.06, # occasional reading pauses; a pure stream has none
warmup=True, # a visitor does not land on a deep page first
)
# The gaps are drawn, not set. A fixed interval produces perfectly regular arrivals,
# which is a stronger signal than being fast.
pacer = Pacer(policy, seed=1)
gaps = [pacer.gap("example.com") for _ in range(500)]
print(f"mean {statistics.mean(gaps):.2f}s median {statistics.median(gaps):.2f}s")
print(f"min {min(gaps):.2f}s max {max(gaps):.2f}s sd {statistics.pstdev(gaps):.2f}s")
config = ScraperConfig(pacing=policy)
# Two scrapers on one host must not look like two contradictory visitors: separate
# addresses, separate clocks, one of them always arriving cold. Sharing the site
# state is what keeps them one visitor.
state = SharedState.create(config)
first = Scraper(origin="https://example.com", config=config, state=state)
second = Scraper(origin="https://example.com", config=config, state=state)
try:
first.get("https://example.com/")
print("what the other scraper knows:", second.knows("https://example.com/").successes)
print(second.explain("https://example.com/"))
finally:
first.close()
second.close()
state.close()
06 · Decoy content¶
The one layer that returns no error.
A honeypot inserts hidden nofollow links into a page, leading into a maze of generated decoy pages. Following them poisons the store and flags the session network-wide, and nothing about the responses says so — the scrape looks like it is working.
uv run python examples/06_decoy_content.py
from scraper import Scraper, ScraperConfig, TopicGuard, safe_links
from scraper.links import looks_like_maze
PAGE = """
<html><body>
<a href="/chapter/1">Chapter 1</a>
<a href="/trap/a" rel="nofollow">bait</a>
<a href="/trap/b" style="display:none">bait</a>
<div style="visibility:hidden"><a href="/trap/c">bait</a></div>
<a href="/trap/d"></a>
</body></html>
"""
# include_rejected shows the reasoning, which is otherwise indistinguishable from a
# page that simply had no links.
for link in safe_links(PAGE, "https://example.com/index.html", include_rejected=True):
verdict = "follow" if link.followable else f"skip ({link.rejected})"
print(f"{verdict:32} {link.url}")
print()
# The backstop for having got it wrong anyway. Decoy pages are generated to read as
# plausible prose, so structure will not give them away — but they are not *about*
# what the site is about.
guard = TopicGuard(min_samples=3)
for _ in range(4):
guard.learn("chapter translation novel protagonist cultivation sect elder sword")
print("on topic :", guard.suspect("the protagonist drew his sword and left the sect"))
print("off topic:", guard.suspect("amortisation schedules against municipal bond covenants"))
print()
# A maze is produced, not authored, so its paths share a shape.
print("generated:", looks_like_maze([f"https://example.com/g/{i:04d}/p" for i in range(20)]))
# on_decoy="raise" turns a suspicion into an exception. Right for anything that
# trains on or republishes what it collects; "warn" is the default because the check
# is a heuristic and a false positive should not fail a job.
config = ScraperConfig(guard_topic=True, on_decoy="raise")
with Scraper(config=config) as scraper:
print("guard configured:", scraper.config.on_decoy)
07 · Web bot auth¶
Signing requests: the honest route through the one layer with no bypass.
A verifier checks a signature over the request under a private key, resolved against a directory you publish. There is nothing to imitate, because the check is arithmetic over a secret — which is also why a valid signature is the cheapest tier in the whole stack: no browser, no proxy reputation, no pacing games.
Current deployments fail open, so an unsigned request is not blocked, it just gets scored by everything else. Where a signature is required there is no bypass, only registration.
Needs the botauth extra: pip install lncrawl-scraper[botauth]
uv run python examples/07_web_bot_auth.py
import json
from pathlib import Path
from scraper import BotAuthConfig, BotAuthKey, Scraper, ScraperConfig
from scraper.botauth import DIRECTORY_PATH
key_path = Path("botauth.key")
if key_path.exists():
key = BotAuthKey.load(key_path)
else:
key = BotAuthKey.generate()
key.save(key_path) # written 0600
print(f"generated a new key at {key_path}")
print("key id (JWK thumbprint):", key.key_id)
print()
print(f"serve this at {DIRECTORY_PATH} on a host you control:")
print(json.dumps(key.directory(), indent=2))
print()
signed = key.sign("https://example.com/some/page", agent="https://crawler.example/")
for name, value in signed.as_headers().items():
print(f"{name}: {value}")
# Proof the setup works, without waiting for a site to tell you it does not.
print()
print("verifies:", key.verify("https://example.com/x", signed, agent="https://crawler.example/"))
print("but not for another host:", key.verify("https://elsewhere.test/x", signed))
config = ScraperConfig(
botauth=BotAuthConfig(
key=key,
agent="https://crawler.example/",
# Roll out one site at a time; empty signs everywhere.
only_hosts=("example.com",),
)
)
with Scraper(config=config) as scraper:
scraper.get("https://example.com/")
08 · Archive and managed¶
The cheapest rung and the most expensive one.
The cheapest way past a protected site is not to touch it: an archived snapshot is served from a host with no mitigation stack. It trades freshness for cost, which only you can weigh, so it is off by default.
The last rung is delegation, and it is last for an honest reason: against a per-zone composite model that is actively tuned, maintaining a bypass is a standing engineering cost rather than a piece of work with an end. No provider is bundled — their formats differ and change, and a wrapper that guesses wrong fails in a way that looks like the site blocking you.
uv run python examples/08_archive_and_managed.py
import requests
from scraper import Scraper, ScraperConfig
from scraper.tiers import http_provider
from scraper.tiers.archive import SOURCE_HEADER, ArchiveTier
from scraper.transport import ImpersonateTransport
URL = "https://example.com/"
# What is in the archive, and how stale.
transport = ImpersonateTransport()
tier = ArchiveTier(transport)
for timestamp, original in tier.captures(URL, limit=5):
print(timestamp, original)
transport.close()
config = ScraperConfig(
archive=True,
archive_max_age=90 * 86400, # refuse anything older than 90 days
)
with Scraper(config=config) as scraper:
response = scraper.get(URL)
# The response carries the *original* URL, so relative links resolve against the
# real site rather than redirecting the crawl into the snapshot.
print(response.url, "captured", response.headers.get(SOURCE_HEADER))
def my_provider(method: str, url: str, **options) -> requests.Response:
"""Anything satisfying this signature is a managed tier.
It must return the *origin's* status and body. Returning the provider's own
status instead breaks diagnosis: a 200 wrapping a 403 reads as a successful
scrape of a block page.
"""
return requests.request(method, url, timeout=options.get("timeout") or 60)
with Scraper(config=ScraperConfig(managed=my_provider)) as scraper:
print(scraper.planner.ladder())
# For the several services shaped as "GET the endpoint with the target as a parameter".
provider = http_provider("https://api.provider.test/v1", token="k3y", extra={"render_js": "true"})
print(provider)
09 · Diagnostics¶
Finding out why, which is the point of the whole design.
Two habits make this library usable in anger: read explain() after a run, and read
the exception's layer rather than its status code. "403 after 3 retries" is the
message that sends people to rewrite the part that was already working.
uv run python examples/09_diagnostics.py
import logging
from scraper import Layer, Scraper, ScraperConfig, diagnose
from scraper.exceptions import Blocked, Exhausted, Impassable, Poisoned
# DEBUG prints one line per decision, with the reasoning attached.
logging.basicConfig(level=logging.DEBUG, format="%(levelname)-7s %(message)s")
# Diagnosis is a pure function, so a saved page can be classified with no network.
# This is the fastest way to check what the library thinks of a response you captured.
saved = "<html><head><title>Just a moment...</title></head><body>__cf_chl_</body></html>"
print(diagnose(status=200, body=saved))
print(diagnose(status=429, headers={"retry-after": "30"}))
print(diagnose(status=403, body="<p>Error 1020</p>"))
print(diagnose(status=403, body="<p>Error 1010</p>"))
print(diagnose(status=404))
URL = "https://example.com/deep/page"
with Scraper(config=ScraperConfig(remember=True)) as scraper:
try:
scraper.get(URL)
except Impassable as exc:
# No bypass exists. The message names the only legitimate route.
print("stop:", exc.layer, exc.detail)
except Poisoned as exc:
print("decoy content:", exc.detail)
except Exhausted as exc:
# A bypass may exist; this configuration does not reach it.
print("out of reach:", exc.layer)
print(" ", exc.detail)
if exc.layer is None:
# Worth branching on: the failure is ours, not the site's — a proxy
# credential, an origin that never answered. Nothing about the site's
# defences was learned, and no exit is worth rotating over it.
print(" -> check the configuration, not the target")
elif exc.layer is Layer.IP_REPUTATION:
print(" -> configure a residential or mobile exit")
elif exc.layer in (Layer.MANAGED_CHALLENGE, Layer.TURNSTILE, Layer.CDP):
print(" -> configure a browser solver")
except Blocked as exc:
trait = exc.layer_info.trait.value if exc.layer_info else "unattributed"
print("blocked at", exc.layer, "which reads a", trait, "property")
print()
print(scraper.explain(URL))
# Everything learned is inspectable, and persists between runs.
profile = scraper.knows(URL)
print()
print("binding layer :", profile.binding)
print("working tier :", profile.tier)
print("interval :", round(profile.interval, 2))
print("ledger :", profile.successes, "ok /", profile.failures, "failed")
print("known decoys :", len(profile.decoys))
10 · Files and soup¶
The ergonomic surface: soup, JSON, forms, files, images.
Every one of these goes through the same retrieval loop, so a challenged site is handled identically whether you asked for HTML or a cover image.
uv run python examples/10_files_and_soup.py
from pathlib import Path
from scraper import PageSoup, Scraper
out = Path("downloads")
with Scraper(origin="https://example.com") as scraper:
scraper.headers["accept-language"] = "en-GB,en;q=0.9"
soup = scraper.get_soup("https://example.com/")
print(soup.select_one("h1").text)
print(soup.select_one("p").text[:60])
# Never None: an empty PageSoup is falsy and its accessors return "".
missing = soup.select_one("#nope")
print(bool(missing), repr(missing.text), repr(missing.get_attr("href")))
# Sub-resources are marked as such: different fetch metadata, and they stay out of
# the referrer chain, because a chain threaded through every image is not one a
# browser produces.
scraper.get_file("https://example.com/", out / "page.html")
print("saved", (out / "page.html").stat().st_size, "bytes")
# data = scraper.get_json("https://example.com/api/items")
# response = scraper.submit_form("https://example.com/search", data={"q": "hello"})
# image = scraper.get_image("https://example.com/cover.jpg") # needs the image extra
# Parsing without fetching.
standalone = PageSoup.create("<div class='a'><span>hi</span></div>")
print(standalone.select_one(".a span").text)
print([node.name for node in standalone.select_one(".a")])