爬虫先探测再抓取：Python站点可爬性预检与采集策略设计

这次我们看一个很有意思的爬虫设计思路在承诺“能抓”之前先对目标站点做一次探测。项目标题是 “Show HN: Scraper that probes a site before promising it works”核心观点很直接——很多爬虫工具一上来就告诉你能抓某某网站结果跑两步就遇到登录墙、反爬校验、动态渲染、接口鉴权最后全卡住。与其这样不如让爬虫在正式开始任务之前先“摸一遍”目标站点的状态把能不能抓、怎么抓、抓多快、会踩什么坑提前反馈给你。这篇文章不吹概念直接拆解这个“先探测再爬取”的工程思路并给出一套可落地的 Python 爬虫项目模板。内容包括核心能力设计、探测模块怎么写、动态页面检测、robots 合规判断、批量任务调度、接口 API 封装、性能观察和常见排错。不管你是准备自建爬虫还是想给团队做一个通用的网页采集工具这套设计都能直接参考。1. 核心能力速览能力项说明项目定位站点可爬性探测 智能爬取策略选择核心特点爬取前先探测站点状态再决定是否执行任务主要功能robots 检查、HTTP 状态探测、JS 渲染检测、登录墙识别、反爬策略预判、批量任务管理支持协议HTTP/HTTPS静态 HTML 与动态渲染页面运行环境Python 3.8支持 Windows / Linux / macOS推荐硬件常规 CPU 即可无需 GPU启动方式命令行任务 Web API 服务是否支持 API支持可提交探测任务和爬取任务是否支持批量任务支持通过任务队列管理输出格式JSON / CSV / Markdown按需扩展适合场景采集前评估、定向网页数据提取、站群监控、通用爬虫服务这里不写死某个开源仓库的具体版本因为“probe before promise”本身是一种工程模式。你完全可以基于这个思路用 Requests、Scrapy 或 Playwright 自己实现一套。2. 适用场景与使用边界2.1 适合谁用需要批量采集网页数据的团队先探测站点结构避免写完整套爬虫后发现目标站变更了选择器。做 SEO 或竞品分析的个人开发者对一批 URL 做可爬性检测快速知道哪些页面能拿到有效内容。做数据中台或采集服务的工程师把探测结果作为任务调度前置条件提高采集成功率。需要动态渲染页面的场景探测到页面依赖 JavaScript 渲染后自动切换 Playwright / Selenium 策略。2.2 能解决什么问题避免盲目爬取导致 IP 被封。提前发现需要登录、需要验证码、返回空内容的页面。自动识别站点是否支持分页、是否有列表页结构。减少无效任务对带宽和 CPU 的浪费。2.3 不适合什么场景需要绕过登录认证、验证码、付费墙的采集这类行为不在本设计范围内。对站点造成高并发压力的采集需要严格控制 QPS。涉及个人隐私、版权内容、未授权数据的采集需要先确认合规性。2.4 合规边界爬虫必须遵守目标网站的robots.txt规则、服务条款和相关法律法规。采集公开数据也应当控制频率、注明来源、不用于非法用途。涉及用户个人信息、版权作品、商业机密的内容必须先取得合法授权。批量采集时建议在请求头中标注清晰的User-Agent和联系方式方便站点管理员追踪。3. 本地爬虫框架整体设计先讲整体架构。这个项目可以拆成四个模块探测模块对目标 URL 发请求分析响应状态、响应头、页面内容、是否动态渲染。决策模块根据探测结果决定爬取策略静态请求 / 浏览器渲染 / 放弃任务。执行模块按策略抓取页面并解析结构化数据。调度模块管理任务队列、重试、日志和输出。“先探测再承诺”的关键点在第 1 和第 2 步。不要让爬虫直接进入抓取循环而是先跑一轮“预检”。4. 环境准备与前置条件4.1 操作系统建议使用 Linux 或 macOS 部署服务Windows 也可以运行但路径处理和进程管理需要注意。4.2 Python 环境建议 Python 3.8 及以上版本。使用venv或conda隔离依赖。python3 -m venv venv source venv/bin/activate4.3 依赖库根据实际需求安装pip install requests beautifulsoup4 lxml httpx如果要做动态页面探测再加 Playwrightpip install playwright playwright install chromium4.4 目录结构建议按下面方式组织项目probe_scraper/ ├── main.py # 入口 ├── config.py # 配置文件 ├── requirements.txt # 依赖 ├── core/ │ ├── __init__.py │ ├── prober.py # 探测模块 │ ├── decision.py # 决策模块 │ ├── fetcher.py # 抓取执行模块 │ └── parser.py # 页面解析模块 ├── api/ │ ├── __init__.py │ └── server.py # API 服务 ├── tasks/ │ └── queue.py # 任务队列 └── outputs/ # 输出目录5. 安装部署与启动方式这个项目可以用两种方式启动命令行模式和服务模式。5.1 命令行模式直接运行探测脚本python main.py probe --url https://example.com --timeout 5输出结果会显示站点状态、响应时间、是否支持 JS、是否有登录墙等。5.2 API 服务模式启动一个轻量的 HTTP 服务方便其他系统调用python main.py serve --host 127.0.0.1 --port 8000然后通过 HTTP 接口提交探测任务和爬取任务。5.3 配置文件示例在config.py中维护通用参数# config.py CONFIG { default_timeout: 10, user_agent: Mozilla/5.0 (compatible; ProbeScraper/1.0; https://example.com/bot), max_retries: 3, retry_backoff: 2, output_dir: ./outputs, allowed_domains: [], respect_robots: True, request_interval: 1.0, max_concurrent_tasks: 4, }6. 核心功能测试与效果验证6.1 站点可爬性探测先测试最基本的探测能力。这里给出一段通用探测代码实际项目需要根据目标站点的响应结构调整。# core/prober.py import requests from urllib.parse import urlparse class SiteProber: def __init__(self, timeout10, user_agentNone): self.timeout timeout self.user_agent user_agent or ProbeScraper/1.0 def probe(self, url: str) - dict: result { url: url, status_code: None, final_url: None, content_type: None, server: None, is_redirect: False, uses_js: None, has_login_wall: False, robots_allowed: None, error: None, } headers { User-Agent: self.user_agent, } try: response requests.get(url, headersheaders, timeoutself.timeout, allow_redirectsTrue) result[status_code] response.status_code result[final_url] response.url result[content_type] response.headers.get(Content-Type) result[server] response.headers.get(Server) result[is_redirect] response.history and any(r.status_code in (301, 302) for r in response.history) html response.text result[uses_js] self._detect_js_rendering(html) result[has_login_wall] self._detect_login_wall(html, response.url) result[robots_allowed] self._check_robots(url) except requests.exceptions.Timeout: result[error] TIMEOUT except requests.exceptions.ConnectionError: result[error] CONNECTION_ERROR except Exception as exc: result[error] str(exc) return result def _detect_js_rendering(self, html: str) - bool: markers [ id\root\, id\app\, id\__next\, window.__NUXT__, data-server-rendered, script src, render(.*?), ] return any(marker in html for marker in markers) def _detect_login_wall(self, html: str, url: str) - bool: markers [login, signin, password, 验证码, 登录] lowered html.lower() return any(marker in lowered for marker in markers) def _check_robots(self, url: str) - bool: parsed urlparse(url) base f{parsed.scheme}://{parsed.netloc} robots_url f{base}/robots.txt try: robots requests.get(robots_url, headers{User-Agent: self.user_agent}, timeout5) if robots.status_code ! 200: return True # 没有 robots.txt 时默认允许需自行判断策略 return fDisallow: {parsed.path} not in robots.text except Exception: return True这段代码实现了一个基础探测检查状态码、是否重定向、是否使用 JS 渲染、是否出现登录关键词、是否被 robots 禁止。启动后可以用一个目标 URL 测试python main.py probe --url https://example.com --timeout 5预期结果类似{ url: https://example.com, status_code: 200, final_url: https://example.com, content_type: text/html; charsetUTF-8, server: ECS (dcb/7EC3), is_redirect: false, uses_js: false, has_login_wall: false, robots_allowed: true, error: null }判断成功的标准返回结果中没有error字段status_code为 200并且robots_allowed为True。如果uses_js为True说明这个站点需要切换浏览器渲染策略。你可以继续用 Playwright 做一次补充探测。6.2 动态页面探测很多现代网站骨架是空的内容全部由 JavaScript 渲染。直接使用 Requests 拿不到有效数据此时需要自动切换浏览器模式。# core/prober.py import asyncio from playwright.async_api import async_playwright async def probe_with_js(url: str): async with async_playwright() as p: browser await p.chromium.launch(headlessTrue) page await browser.new_page() await page.goto(url, timeout30000, wait_untilnetworkidle) title await page.title() content await page.content() await browser.close() return { title: title, content_length: len(content), uses_js: True, }在决策模块里你可以这样切换策略# core/decision.py def choose_strategy(probe_result: dict) - str: if probe_result.get(error): return abort if probe_result.get(status_code) ! 200: return abort if probe_result.get(has_login_wall): return auth_required if probe_result.get(uses_js): return browser return static这个模块的价值是在真正发起大规模爬取之前先对每个 URL 做一次低成本探测把任务分成“可静态抓”“需要渲染”“不可抓”三类。这样可以减少很多无效请求。6.3 页面内容解析与字段提取探测通过后进入抓取和解析阶段。用 BeautifulSoup 提取数据时先定义字段映射# core/parser.py from bs4 import BeautifulSoup def parse_article(html: str): soup BeautifulSoup(html, lxml) title soup.find(h1).get_text(stripTrue) if soup.find(h1) else None content soup.find(article) paragraphs [] if content: paragraphs [p.get_text(stripTrue) for p in content.find_all(p)] return { title: title, paragraphs: paragraphs, text_length: sum(len(p) for p in paragraphs), }6.4 批量任务测试批量场景下先读取 URL 列表逐个探测再按策略分组执行。这里提供一个简单的任务队列示例# tasks/queue.py from collections import deque class TaskQueue: def __init__(self, urls): self.urls deque(urls) def next(self): return self.urls.popleft() if self.urls else None批量入口可以这样写# main.py from core.prober import SiteProber from core.decision import choose_strategy from core.fetcher import fetch_static, fetch_browser from tasks.queue import TaskQueue def run_batch(urls): prober SiteProber(timeout10) queue TaskQueue(urls) results [] while True: url queue.next() if not url: break probe_result prober.probe(url) strategy choose_strategy(probe_result) if strategy abort: results.append({url: url, status: aborted, reason: probe_result.get(error)}) continue if strategy browser: page_data fetch_browser(url) else: page_data fetch_static(url) results.append({ url: url, status: success, strategy: strategy, data: page_data, }) return results这个批量流程的关键点在于每个任务都带有策略标记你可以根据策略统计成功率和失败原因。7. 接口 API 与批量任务7.1 API 服务设计为了方便和其他系统对接可以暴露两个接口POST /api/probe提交 URL 探测任务。POST /api/scrape提交抓取任务。用 FastAPI 实现一个简单服务pip install fastapi uvicorn# api/server.py from fastapi import FastAPI from pydantic import BaseModel from core.prober import SiteProber from core.decision import choose_strategy from core.fetcher import fetch_static, fetch_browser app FastAPI() prober SiteProber() class ProbeRequest(BaseModel): url: str timeout: int 10 class ScrapeRequest(BaseModel): url: str use_browser: bool False app.post(/api/probe) def probe_site(req: ProbeRequest): result prober.probe(req.url) result[strategy] choose_strategy(result) return result app.post(/api/scrape) def scrape_site(req: ScrapeRequest): if req.use_browser: data fetch_browser(req.url) else: data fetch_static(req.url) return {url: req.url, data: data}启动服务uvicorn api.server:app --host 127.0.0.1 --port 80007.2 curl 调用示例curl -X POST http://127.0.0.1:8000/api/probe \ -H Content-Type: application/json \ -d {url: https://example.com, timeout: 10}返回{ url: https://example.com, status_code: 200, final_url: https://example.com, content_type: text/html; charsetUTF-8, server: ECS (dcb/7EC3), is_redirect: false, uses_js: false, has_login_wall: false, robots_allowed: true, error: null, strategy: static }7.3 Python 调用示例import requests probe_api http://127.0.0.1:8000/api/probe scrape_api http://127.0.0.1:8000/api/scrape def probe_and_scrape(url: str): resp requests.post(probe_api, json{url: url, timeout: 10}, timeout30) probe_result resp.json() print(探测结果:, probe_result) if probe_result.get(status_code) ! 200 or probe_result.get(error): print(放弃抓取) return None strategy probe_result.get(strategy) scrape_resp requests.post(scrape_api, json{url: url, use_browser: strategy browser}, timeout60) return scrape_resp.json()7.4 批量任务队列扩展真正的生产环境建议引入 Celery 或 APScheduler。任务流程批量 URL 入库。每个任务先执行探测。根据探测结果决定抓取策略。记录探测日志和抓取日志。失败任务进入重试队列最多重试 3 次。重试逻辑示例def run_with_retry(func, *args, max_retries3, backoff2): for attempt in range(1, max_retries 1): try: return func(*args) except Exception as exc: if attempt max_retries: raise time.sleep(backoff * attempt)8. 资源占用与性能观察这个项目不涉及 GPU主要资源消耗在网络请求和页面解析上。重点观察四个维度请求频率控制每秒请求数避免触发反爬。连接池使用requests.Session()复用连接减少 TCP 握手开销。内存大批量列表不要一次全部 load 进内存用生成器或队列控制。CPUBeautifulSoup 解析大页面时比较耗 CPU可考虑 lxml 或正则优化。8.1 如何观察资源占用Linux 下可以用top或htoptop -p pid查看当前进程的 CPU 和内存使用情况。Windows 下用任务管理器或resource_monitor。8.2 降低请求频率的方案在config.py中设置请求间隔import time request_interval CONFIG.get(request_interval, 1.0) def rate_limit(): time.sleep(request_interval)8.3 避免端口冲突API 服务默认使用 8000 端口如果被占用可以指定其他端口uvicorn api.server:app --host 127.0.0.1 --port 8001或者检测端口lsof -i :8000如果发现残留进程先清理再重启。9. 常见问题与排查方法问题现象可能原因排查方式解决方案探测结果一直超时目标站点响应慢或屏蔽了请求头用 curl 测试目标 URL检查 User-Agent增加 timeout换更完整的 User-Agent状态码 403 / 429触发反爬或 QPS 过高检查响应头中的 Server 和 Retry-After降低请求频率增加 Cookies 或代理池uses_js为 True 但静态抓取有数据页面中既有静态内容也有动态内容对比静态 HTML 和浏览器渲染 HTML按实际需求选择解析来源robots 检测失败robots.txt 格式异常或网络问题手动访问 robots.txt容错处理检测失败时默认不允许或按配置决定Playwright 启动失败浏览器内核未安装运行playwright install chromium安装对应浏览器内核批量任务卡住某个 URL 长时间无响应查看任务日志检查超时设置给请求设置硬超时限制最大重试次数解析字段为空页面结构变化或选择器失效检查页面 HTML 结构升级选择器使用更稳健的解析逻辑API 返回 500服务器异常或依赖缺失查看 uvicorn 日志修复对应模块检查依赖10. 最佳实践与使用建议10.1 先小规模探测再全量爬取每次接入新站点不要直接跑 10 万条 URL。先拿 5 个 URL 跑通探测和解析流程确认数据质量后再扩大规模。10.2 维护一份站点配置档案建议用 JSON 或数据库保存每个站点的探测结果{ example.com: { status_code: 200, uses_js: false, has_login_wall: false, strategy: static, last_probed_at: 2025-01-01T10:00:00Z } }这样下次任务可以直接跳过探测阶段节省时间。10.3 分离探测与抓取日志日志中要包含任务 ID、URL、阶段probe / scrape、结果、耗时。方便追溯失败原因。python main.py run --url-file urls.txt --log-dir ./logs10.4 注意合规安全爬取前检查目标站点的robots.txt。设置合法的User-Agent最好包含联系方式。控制并发和请求频率。不采集个人隐私、版权内容或未授权数据。采集结果不用于恶意用途不公开传播敏感数据。10.5 接口服务限制访问范围API 服务不能直接暴露到公网。如果必须提供远端访问加一层鉴权# api/server.py from fastapi import Depends, HTTPException, Header API_TOKEN your-secure-token def verify_token(authorization: str Header(None)): if authorization ! fBearer {API_TOKEN}: raise HTTPException(status_code401, detailUnauthorized) return True然后对需要保护的路由添加dependencies[Depends(verify_token)]。11. 总结与下一步这个项目最值得尝试的点是“先探测再承诺”的决策机制。它把爬虫从“盲抓”变成“先评估后执行”在批量采集场景下能明显减少无效请求。你最先要验证的是SiteProber对不同站点的探测结果是否准确特别是uses_js和has_login_wall这两个字段。最容易踩的坑是目标站点有反爬机制导致探测阶段返回 403后续逻辑全部中断——所以在探测模块里一定要把状态码、响应头、异常类型都记录下来。接下来可以继续扩展的方向加入机器学习分类器根据页面特征自动判断内容类型文章、列表、详情页。支持更多渲染环境如 Chrome DevTools Protocol。把探测结果持久化到数据库做站点可爬性画像。引入 Celery 分布式任务队列处理大规模采集。增加数据质量校验对抓取结果做字段完整性检查。如果你正在搭爬虫服务建议把“探测”作为一个独立模块放到任务最前面。一次探测的成本远低于一次爬取失败带来的 IP 封禁和资源浪费。这个思路也可以迁移到其他场景同步任务前先做预检、API 调用前先做连通性测试本质上都是在降低系统的不确定性。建议收藏备用实际跑一遍之后再根据你自己的站点类型调整探测规则和策略判断逻辑。