Ryan ColeMost commercial scrapers break the second a target site switches to dynamic DOM hydration, updates...
Most commercial scrapers break the second a target site switches to dynamic DOM hydration, updates its client-side JavaScript, or deploys Cloudflare Turnstile. Cloud scraping platforms charge $49-199 every month for limited lead credits and arbitrary rate limits.
I got tired of paying that, so I built a local dual-engine scraper. Here is the exact architecture, with working code you can run today.
Static pages (sitemaps, REST endpoints, server-rendered catalogs) do not need a browser. An async HTTP client harvests 1,000+ URLs in under 30 seconds. But React/Vue/Next.js SPAs, infinite scroll, and shadow DOM need a real headless browser. The trick is routing each URL to the cheapest engine that can handle it.
import asyncio, httpx, re
EMAIL_RE = re.compile(r"[a-zA-Z0-9_.+-]+@[a-zA-Z0-9-]+\.[a-zA-Z0-9-.]+")
PHONE_RE = re.compile(r"\+?[\d][\d\s\-().]{7,}\d")
async def fetch_static(client, url):
r = await client.get(url, timeout=20)
r.raise_for_status()
return r.text
async def harvest(urls):
async with httpx.AsyncClient(
headers={"User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36"},
follow_redirects=True,
) as client:
sem = asyncio.Semaphore(10)
async def one(u):
async with sem:
await asyncio.sleep(1.5) # polite jitter
try:
return u, await fetch_static(client, u)
except Exception as e:
return u, f"ERROR: {e}"
return await asyncio.gather(*[one(u) for u in urls])
The rule is simple: if the static fetch returns almost no text (JS-rendered shell), escalate that URL to Playwright with stealth settings.
from playwright.async_api import async_playwright
async def fetch_dynamic(url):
async with async_playwright() as p:
browser = await p.chromium.launch(headless=True)
ctx = await browser.new_context(
user_agent="Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36",
viewport={"width": 1366, "height": 768},
)
page = await ctx.new_page()
await page.goto(url, wait_until="networkidle", timeout=30000)
# human-like scroll to trigger lazy hydration
await page.mouse.wheel(0, 2000)
await page.wait_for_timeout(2000)
html = await page.content()
await browser.close()
return html
Add fingerprint rotation, request jitter (1.5-4.5s), and proxy support, and you have the anti-detection layer that keeps this running without bans.
import hashlib
seen = set()
def dedup_key(record: str) -> bool:
h = hashlib.sha256(record.encode()).hexdigest()
if h in seen:
return False
seen.add(h)
return True
def extract_contacts(html: str):
return {
"emails": sorted(set(EMAIL_RE.findall(html))),
"phones": sorted(set(PHONE_RE.findall(html))),
}
Export to Excel/CSV/JSON and you have a CRM-ready lead list from any directory, store, or registry.
I turned this exact pattern into a finished CLI — async HTTPX engine + stealth Playwright fallback, 10 pre-configured presets (B2B directories, e-commerce stores, review catalogs, local registries), contact auto-extraction, Bloom-filter dedup, and xlsx/csv/json export. One command, first dataset in under 60 seconds:
python omniscraper.py --target "https://example-directory.com/companies" --depth 2 --output leads.xlsx
# [SUCCESS] Extracted 482 verified company records in 24.8s. Saved to leads.xlsx.
It just passed QC and launched on Gumroad: OmniScraper AI — Headless Lead Intelligence CLI. Pay once, run forever locally — no monthly subscription, no credit limits.
Launch deal: code SCRAPER20 takes 20% off (first 50 uses):
https://ancuboy.gumroad.com/l/omniscraper-ai/SCRAPER20
But you do not need it to use the two-engine pattern above. Start with the static harvester — it covers most directories for free.