5分钟搭建国外搜索网站爬虫，源码解析带你搞定数据难题

官方文档往往厚达几百页，读起来让人头晕，重点却藏在角落。很多开发者卡在环境配置上，还没开始写代码就放弃了。今天直接上干货，通过源码解析一个轻量级的国外搜索网站数据获取工具，帮你快速抓住核心逻辑。

项目目标与场景定位

我们要解决的不是简单的页面抓取，而是如何稳定地从国外主流搜索网站获取结构化数据。以 DuckDuckGo 为例，它的接口相对友好，且无强登录限制，适合作为入门实战项目。

核心目标：

获取关键词搜索结果的前 10 条数据（标题、链接、摘要）。
数据清洗与标准化，输出为 JSON 格式。
具备基础的反爬应对能力（请求头伪装、重试机制）。

为什么选这个场景？ 对于转岗开发者或初级后端工程师，掌握非结构化数据的获取与清洗，是理解数据管道（Data Pipeline）第一步的关键。国外网站通常遵循标准 HTML 结构，便于学习 DOM 解析；同时，其网络环境对国内直连有一定挑战，能顺带练习网络调试技巧。

目录结构与依赖管理

保持工程化习惯，从零搭建项目。我们使用 Python 3.9+，依赖库极简，避免过度工程化。

search_scraper/
├── main.py          # 入口文件
├── scraper.py       # 核心抓取逻辑
├── parser.py        # HTML 解析逻辑
├── utils.py         # 工具函数（重试、日志）
├── requirements.txt # 依赖清单
├── config.yaml      # 配置文件
└── output/          # 数据输出目录

requirements.txt:

requests==2.31.0
beautifulsoup4==4.12.2
lxml==4.9.1
PyYAML==6.0

config.yaml 示例：

target:url: "https://duckduckgo.com/html"headers:User-Agent: "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/120.0.0.0 Safari/537.36"Accept-Language: "en-US,en;q=0.9"
request:timeout: 10retry_count: 3delay_min: 1.0delay_max: 3.0

核心代码实现与源码解析

这部分是重点，我们逐行拆解 scraper.py 和 parser.py。

1. 网络请求封装 (utils.py)

不要裸用 requests.get，必须加上重试和随机延迟，否则极易被风控。

import time
import random
import logging
import requests
from requests.adapters import HTTPAdapter
from urllib3.util.retry import Retry# 配置日志
logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s')
logger = logging.getLogger(__name__)class RobustSession:def __init__(self, config):self.config = configself.session = requests.Session()# 重试机制：遇到 5xx 或 429 (Too Many Requests) 自动重试retry_strategy = Retry(total=config['request']['retry_count'],backoff_factor=1,status_forcelist=[429, 500, 502, 503, 504])adapter = HTTPAdapter(max_retries=retry_strategy)self.session.mount("http://", adapter)self.session.mount("https://", adapter)# 设置默认请求头self.session.headers.update(config['target']['headers'])def get(self, url, params=None):"""发送 GET 请求，包含随机延迟"""# 随机延迟，模拟人类行为delay = random.uniform(self.config['request']['delay_min'], self.config['request']['delay_max'])logger.info(f"Sleeping for {delay:.2f}s before request...")time.sleep(delay)try:response = self.session.get(url, params=params, timeout=self.config['request']['timeout'])response.raise_for_status() # 如果状态码不是 2xx，抛出异常return responseexcept requests.RequestException as e:logger.error(f"Request failed: {e}")raise

2. HTML 解析逻辑 (parser.py)

使用 BeautifulSoup 解析 DuckDuckGo 的 HTML 结构。注意，不同网站的 DOM 结构可能变化，这里以当前稳定版本为例。

from bs4 import BeautifulSoup
import reclass DuckDuckGoParser:@staticmethoddef parse(html_content: str) -> list:"""解析 HTML 内容，提取搜索结果:param html_content: 原始 HTML 字符串:return: 包含结果字典的列表"""soup = BeautifulSoup(html_content, 'lxml')results = []# 搜索结果通常在 class 为 'result' 的 div 中# 注意：不同网站选择器不同，需根据实际页面调整result_divs = soup.find_all('div', class_='result')for div in result_divs:# 提取标题和链接title_tag = div.find('a', class_='result__a')if not title_tag:continuetitle = title_tag.get_text(strip=True)url = title_tag.get('href', '')# 提取摘要snippet_tag = div.find('a', class_='result__snippet')snippet = snippet_tag.get_text(strip=True) if snippet_tag else ""# 数据清洗：去除多余空格和特殊字符title = re.sub(r'\s+', ' ', title)snippet = re.sub(r'\s+', ' ', snippet)results.append({'title': title,'url': url,'snippet': snippet})return results

源码解析关键点：

选择器稳定性：class_='result__a' 是 DuckDuckGo 的特定类名。在实际项目中，建议结合 ID 或唯一属性进行定位，避免仅依赖 class，因为前端框架可能会动态生成类名。
异常处理：if not title_tag: continue 确保了即使某个结果块结构异常，也不会中断整个解析过程。
正则清洗：re.sub(r'\s+', ' ', ...) 用于统一空白字符，保证数据质量。

3. 主流程整合 (main.py)

import yaml
import json
import os
from scraper import RobustSession
from parser import DuckDuckGoParserdef load_config(path='config.yaml'):with open(path, 'r', encoding='utf-8') as f:return yaml.safe_load(f)def main():config = load_config()session = RobustSession(config)parser = DuckDuckGoParser()keyword = input("请输入搜索关键词: ")params = {'q': keyword}print(f"正在获取 '{keyword}' 的搜索结果...")response = session.get(config['target']['url'], params=params)if response.status_code == 200:results = parser.parse(response.text)if not results:print("未获取到任何结果，请检查网络或选择器。")return# 输出到 JSON 文件output_dir = 'output'os.makedirs(output_dir, exist_ok=True)output_file = os.path.join(output_dir, f"results_{keyword}.json")with open(output_file, 'w', encoding='utf-8') as f:json.dump(results, f, ensure_ascii=False, indent=2)print(f"成功获取 {len(results)} 条结果，已保存至 {output_file}")else:print(f"请求失败，状态码: {response.status_code}")if __name__ == '__main__':main()

运行与测试

在项目根目录执行：

pip install -r requirements.txt
python main.py

测试场景：

正常场景：输入 "Python tutorial"，应生成 output/results_Python_tutorial.json，包含 10 条左右数据。
边界场景：输入特殊字符如 "C++ & Java"，检查 URL 编码是否正确处理（requests 库会自动处理 params 的编码）。
失败场景：断开网络或修改 URL 为错误地址，观察日志是否记录了重试过程。

常见坑点：

编码问题：务必在读取 YAML 和写入 JSON 时指定 encoding='utf-8'，否则中文关键词会报错。
IP 封禁：如果在短时间内多次运行，DuckDuckGo 可能会返回验证码页面。此时 parser 会返回空列表，需增加对验证码页面的检测逻辑（如检测 HTML 中是否包含 "challenge" 字样）。

优化扩展与避坑指南

基础版能跑通，但离生产环境还有距离。以下是几个关键优化方向：

1. 异步请求提升效率

如果需要批量抓取多个关键词，同步请求会因 time.sleep 而变得极慢。引入 aiohttp 和 asyncio 可显著提升吞吐量。

import asyncio
import aiohttpasync def async_fetch(session, url, params):async with session.get(url, params=params) as response:return await response.text()