Python爬虫实战：requests+BeautifulSoup+Scrapy全栈爬虫开发指南

Python爬虫实战：requests+BeautifulSoup+Scrapy全栈爬虫开发指南

导语： 从简单的单页抓取到复杂的分布式爬虫，Python爬虫体系覆盖了Web数据采集的完整链路。本文从 requests 基础请求、BeautifulSoup 解析HTML、lxml 高性能解析、Scrapy框架搭建，到反爬绕过、代理池、异步爬取一路讲下来，附带真实可运行代码，是从零上手Python爬虫的完整路线图。

一、HTTP请求基础：requests 实战

1.1 基础GET/POST请求

import requests
from requests.adapters import HTTPAdapter
from urllib3.util.retry import Retry

# 基础 GET 请求
response = requests.get(
'https://httpbin.org/get',
params={'key': 'value', 'page': 1},
headers={'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36'},
timeout=10
)
print(response.status_code) # 200
print(response.json()) # 解析JSON响应

# POST 请求
response = requests.post(
'https://httpbin.org/post',
data={'username': 'test', 'password': '123456'}, # 表单提交
# json={'key': 'value'}, # JSON格式提交
)

# 自动重试封装
def create_session(retries=3, backoff_factor=0.5):
session = requests.Session()
retry = Retry(
total=retries,
backoff_factor=backoff_factor,
status_forcelist=[429, 500, 502, 503, 504]
)
adapter = HTTPAdapter(max_retries=retry)
session.mount('http://', adapter)
session.mount('https://', adapter)
return session

session = create_session()

1.2 会话维持与Cookie处理

# 使用 Session 维持登录状态
session = requests.Session()
session.headers.update({
'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64)',
'Accept-Language': 'zh-CN,zh;q=0.9',
})

# 模拟登录
login_data = {'username': 'testuser', 'password': 'testpass', 'remember': True}
session.post('https://example.com/login', data=login_data)

# 登录后访问需要认证的页面
profile = session.get('https://example.com/profile')

# 导出和恢复 Cookie
import json
cookies = requests.utils.dict_from_cookiejar(session.cookies)
with open('cookies.json', 'w') as f:
json.dump(cookies, f)

二、BeautifulSoup HTML解析

from bs4 import BeautifulSoup
import requests

html = requests.get('https://news.ycombinator.com/').text
soup = BeautifulSoup(html, 'lxml') # 推荐使用lxml解析器，速度更快

# CSS选择器（推荐）
titles = soup.select('.storylink')
for title in titles[:5]:
print(title.get_text(), title.get('href'))

# find/find_all
all_links = soup.find_all('a', href=True)
external_links = [a['href'] for a in all_links if a['href'].startswith('http')]

# 获取属性
img_urls = [img['src'] for img in soup.find_all('img', src=True)]

# 结构化提取新闻列表
def parse_hn_news(html):
soup = BeautifulSoup(html, 'lxml')
items = []
for item in soup.select('.athing'):
title_tag = item.select_one('.titlelink')
score_row = item.find_next_sibling()

if not title_tag:
continue

score_tag = score_row.select_one('.score') if score_row else None
items.append({
'id': item.get('id'),
'title': title_tag.get_text(),
'url': title_tag.get('href'),
'score': score_tag.get_text() if score_tag else '0 points'
})
return items

三、lxml + XPath 高性能解析

from lxml import etree
import requests

html = requests.get('https://example.com').text
tree = etree.HTML(html)

# XPath 基础语法
titles = tree.xpath('//h2[@class="title"]/text()')
links = tree.xpath('//a/@href')

# 相对路径提取
articles = tree.xpath('//article')
for article in articles:
title = article.xpath('.//h2/text()')
date = article.xpath('.//time/@datetime')
print(title, date)

# contains() 模糊匹配
news_links = tree.xpath('//a[contains(@class, "news")]/@href')

四、Scrapy框架搭建与实战

4.1 项目初始化

pip install scrapy
scrapy startproject news_spider
cd news_spider
scrapy genspider hacker_news news.ycombinator.com

4.2 Spider 核心实现

# news_spider/spiders/hacker_news.py
import scrapy
from news_spider.items import NewsItem

class HackerNewsSpider(scrapy.Spider):
name = 'hacker_news'
allowed_domains = ['news.ycombinator.com']
start_urls = ['https://news.ycombinator.com/']

custom_settings = {
'DOWNLOAD_DELAY': 1, # 限速：每次请求间隔1秒
'RANDOMIZE_DOWNLOAD_DELAY': True,
'CONCURRENT_REQUESTS': 2,
'DEFAULT_REQUEST_HEADERS': {
'User-Agent': 'Mozilla/5.0 (compatible; research bot)',
}
}

def parse(self, response):
for item in response.css('.athing'):
title_link = item.css('.titlelink')
news = NewsItem()
news['title'] = title_link.css('::text').get()
news['url'] = title_link.attrib.get('href')
news['item_id'] = item.attrib.get('id')

# 获取分数行（下一个兄弟元素）
score_row = item.xpath('following-sibling::tr[1]')
news['score'] = score_row.css('.score::text').get('0 points')
yield news

# 翻页
next_page = response.css('a.morelink::attr(href)').get()
if next_page:
yield response.follow(next_page, self.parse)

4.3 Item Pipeline 数据持久化

# news_spider/pipelines.py
import sqlite3
from itemadapter import ItemAdapter

class SQLitePipeline:
def open_spider(self, spider):
self.conn = sqlite3.connect('news.db')
self.cursor = self.conn.cursor()
self.cursor.execute('''
CREATE TABLE IF NOT EXISTS news (
id TEXT PRIMARY KEY,
title TEXT,
url TEXT,
score TEXT,
created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
)
''')

def process_item(self, item, spider):
adapter = ItemAdapter(item)
self.cursor.execute(
'INSERT OR REPLACE INTO news (id, title, url, score) VALUES (?, ?, ?, ?)',
(adapter['item_id'], adapter['title'], adapter['url'], adapter['score'])
)
self.conn.commit()
return item

def close_spider(self, spider):
self.conn.close()

五、异步爬虫：aiohttp + asyncio

import asyncio
import aiohttp
from typing import List

async def fetch_url(session: aiohttp.ClientSession, url: str) –> dict:
try:
async with session.get(url, timeout=aiohttp.ClientTimeout(total=10)) as resp:
return {'url': url, 'status': resp.status, 'text': await resp.text()}
except Exception as e:
return {'url': url, 'error': str(e)}

async def fetch_all(urls: List[str], concurrency: int = 10) –> List[dict]:
semaphore = asyncio.Semaphore(concurrency)
async def bounded_fetch(session, url):
async with semaphore:
return await fetch_url(session, url)

connector = aiohttp.TCPConnector(limit=100, force_close=True)
async with aiohttp.ClientSession(connector=connector) as session:
tasks = [bounded_fetch(session, url) for url in urls]
return await asyncio.gather(*tasks)

# 使用
urls = [f'https://httpbin.org/get?n={i}' for i in range(20)]
results = asyncio.run(fetch_all(urls, concurrency=5))

六、开发痛点与报错避坑指南

# 检测并修复编码
import chardet
raw = response.content
detected = chardet.detect(raw)
text = raw.decode(detected['encoding'] or 'utf-8', errors='ignore')

七、全文总结

八、技术进阶展望

• 学习 Scrapy-Redis 实现分布式爬虫
• 研究 Playwright-Python 处理复杂动态渲染
• 探索 Scrapy 中间件开发：代理中间件、UA轮换中间件

参考文献