Suuuryaa/KnowScrap

可靠的 Python 网络抓取：内置真实的反机器人绕过功能。
Python 核心 · Node.js 反检测

KnowScraper 处理网络抓取中所有棘手的部分：机器人检测、重试、并发、去重、存储——因此你只需专注于编写 handler，而不是基础设施。 ``` import asyncio from knowscraper import CheerioCrawler, Router, Dataset router = Router() dataset = Dataset(name="quotes") @router.default_handler async def handler(ctx): for q in ctx.parsed.select(".quote"): await dataset.push_data({ "text": q.select_one(".text").get_text(strip=True), "author": q.select_one(".author").get_text(strip=True), }) await ctx.enqueue_links(selector="li.next a") # follow pagination async def main(): crawler = CheerioCrawler( router=router, dataset=dataset, max_concurrency=5, respect_robots_txt=True, use_anti_detection=True, # Chrome TLS fingerprint via Node.js ) await crawler.run(["https://quotes.toscrape.com"]) await dataset.export_to_csv() asyncio.run(main()) ``` ## 安装说明 ``` pip install knowscraper playwright install chromium # only needed for browser crawlers ``` ## 包含的功能 | | 功能 | |---|---| | 🛡️ | **反机器人检测**：Chrome TLS 指纹、浏览器指纹注入、隐身 JS 补丁、拟人化的鼠标/滚动/输入 | | 🌐 | **6 种爬虫**：HTTP、Cheerio、Playwright、Puppeteer、自适应（自动检测 JS）、AI 驱动 | | 🔄 | **持久化队列**：基于 SQLite，可从崩溃中恢复，从停止处继续执行 | | ⚡ | **智能并发**：根据 CPU 和内存自动缩放 | | 🔀 | **Session 和代理轮换**：每个请求拥有独立的身份和 IP | | 🤖 | **AI 提取**：描述你想要的内容，Claude 会为你提取（无需选择器） | | 🧩 | **插件**：`DedupPlugin`、`RetryPlugin`、`StatsPlugin`、`LoggingPlugin` | | 🔒 | **CAPTCHA 解决**：reCAPTCHA v2/v3、hCaptcha、Cloudflare Turnstile | | 🗺️ | **Sitemap**：发现并抓取 XML sitemap，支持 lastmod 过滤 | | 📦 | **导出**：开箱即用支持 CSV 和 JSON | ## 爬虫 | 爬虫 | JS 渲染 | 反机器人检测 | 适用场景 | |---|---|---|---| | `CheerioCrawler` | ✗ | TLS 指纹 | 静态网站、博客、电子商务 | | `HttpCrawler` | ✗ | TLS 指纹 | API、原始 HTTP | | `PlaywrightCrawler` | ✓ | 完整指纹 + 隐身 | SPA、React/Vue/Angular 应用 | | `PuppeteerCrawler` | ✓ | 完整指纹 | Puppeteer 特定的工作流 | | `AdaptiveCrawler` | 自动 | 两者皆有 | 混合网站，自动升级 | | `AICrawler` | ✓ | 完整指纹 | 无需编写选择器即可提取数据 | ## 示例

PlaywrightCrawler：重度依赖 JavaScript 的网站

``` from knowscraper import PlaywrightCrawler, Router, Dataset router = Router() dataset = Dataset(name="results") @router.default_handler async def handler(ctx): title = await ctx.page.title() await dataset.push_data({"title": title, "url": ctx.request.url}) crawler = PlaywrightCrawler( router=router, dataset=dataset, stealth_mode=True, # removes webdriver flag, spoofs plugins inject_fingerprint=True, # real Chrome fingerprint via Node.js random_interactions=True, # human-like mouse + scroll ) await crawler.run(["https://example.com"]) ```

AICrawler：无需选择器

``` from knowscraper import AICrawler, Router, Dataset router = Router() dataset = Dataset(name="products") @router.default_handler async def handler(ctx): # plain English : Claude reads the page and extracts the data data = await ctx.extract("product name, price, rating, and availability") await dataset.push_data(data) # also works for actions await ctx.act("click the Accept Cookies button if it exists") crawler = AICrawler(router=router, dataset=dataset) # 设置 ANTHROPIC_API_KEY 或传递 api_key="sk-ant-..." await crawler.run(["https://example.com/products"]) ```

基于标签的路由：列表页 + 产品页

``` from knowscraper import CheerioCrawler, Router, Dataset, Request router = Router() dataset = Dataset(name="products") @router.handler("listing") async def on_listing(ctx): await ctx.enqueue_links(selector="a.product-link", label="product") await ctx.enqueue_links(selector="a.next-page", label="listing") @router.handler("product") async def on_product(ctx): await dataset.push_data({ "name": ctx.parsed.select_one("h1").get_text(strip=True), "price": ctx.parsed.select_one(".price").get_text(strip=True), "url": ctx.request.url, }) crawler = CheerioCrawler(router=router, dataset=dataset) await crawler.run([Request(url="https://shop.example.com", label="listing")]) ```

插件：去重、重试、统计

``` from knowscraper import CheerioCrawler, DedupPlugin, RetryPlugin, StatsPlugin stats = StatsPlugin() crawler = CheerioCrawler( router=router, plugins=[ DedupPlugin(key="url"), # drop duplicate records before saving RetryPlugin(base_delay=2.0), # exponential backoff: 2s, 4s, 8s... stats, ], ) await crawler.run(["https://example.com"]) print(stats.report()) # {"requests_by_domain": {...}, "avg_response_time": 0.3, "data_records_saved": 42} ```

代理轮换 + session 池

``` from knowscraper import CheerioCrawler, ProxyConfiguration proxy_config = ProxyConfiguration( proxy_urls=[ "http://user:pass@proxy1.example.com:8080", "http://user:pass@proxy2.example.com:8080", ], rotate="round_robin", ) crawler = CheerioCrawler(router=router, proxy_configuration=proxy_config) ```

恢复崩溃的抓取任务

``` # 首次运行：正常爬取 await crawler.run(["https://example.com"]) # 重启：从停止的确切位置继续（SQLite 队列持久化） await crawler.run(resume=True) ```

Sitemap 发现

``` from knowscraper import fetch_sitemap_urls from datetime import datetime urls = await fetch_sitemap_urls( "https://example.com", modified_after=datetime(2024, 1, 1), max_urls=500, ) await crawler.run(urls) ```

## CLI ``` knowscraper create my-project # new HTTP scraper knowscraper create my-project --type playwright # new browser scraper knowscraper run main.py # run a scraper file knowscraper info # version + environment ``` ## 工作原理 KnowScraper 在 Python 旁运行了一个轻量级的 Node.js 微服务。它会自动启动，并处理所有需要匹配 Chrome 精确网络行为的工作。 ``` Your code (Python) │ ▼ BaseCrawler — queue, retries, rate limiting, robots.txt │ ├── CheerioCrawler / HttpCrawler │ └── Node.js service → got-scraping (Chrome TLS) │ └── PlaywrightCrawler / PuppeteerCrawler ├── Node.js service → fingerprint-generator + injector └── Playwright → stealth patches + human interactions ``` ## 开源协议 MIT © [Suuuryaa](https://github.com/Suuuryaa)

标签：DFIR, MITM代理, Playwright, Python, Web爬虫, 反爬虫对抗, 无后门, 浏览器自动化, 特征检测, 逆向工具