用Python 3.11的httpx和BeautifulSoup4高效爬取豆瓣电影Top250在Python爬虫领域技术栈的迭代速度令人目不暇接。十年前流行的urllib2如今已被更现代、更高效的库所取代。本文将带你使用Python 3.11的最新特性结合httpx和BeautifulSoup4这两个强力工具打造一个高效、稳定的豆瓣电影Top250爬虫。1. 为什么选择httpx和BeautifulSoup4组合1.1 httpx vs requests现代HTTP客户端的优势httpx是Python生态中新兴的HTTP客户端库相比传统的requests它带来了几项关键改进原生支持HTTP/2显著提升连接效率特别是在需要大量请求的场景下完整的异步支持与asyncio无缝集成轻松实现高性能并发爬取更智能的连接池自动管理连接复用减少TCP握手开销更严格的类型提示充分利用Python 3.11的类型系统提高代码健壮性import httpx # 同步请求示例 with httpx.Client() as client: response client.get(https://movie.douban.com/top250) # 异步请求示例 async with httpx.AsyncClient() as client: response await client.get(https://movie.douban.com/top250)1.2 BeautifulSoup4的最新解析器选择BeautifulSoup4作为HTML解析的标杆库其性能很大程度上取决于底层解析器。2023年推荐使用以下组合解析器速度内存占用容错性适用场景lxml★★★★★★★★★★★大多数情况首选html.parser★★★★★★★★无外部依赖时使用html5lib★★★★★★★处理极不规范的HTMLfrom bs4 import BeautifulSoup # 推荐使用lxml作为解析器 soup BeautifulSoup(html_content, lxml)2. 豆瓣电影Top250页面结构分析2.1 URL规律与分页处理豆瓣电影Top250采用经典的分页模式每页显示25部电影。通过分析我们发现URL模式非常规律https://movie.douban.com/top250?start{offset}filter其中offset参数遵循以下规律第1页start0显示1-25名第2页start25显示26-50名...第10页start225显示226-250名我们可以利用Python 3.11的math.ceil和生成器表达式高效生成所有页面URLimport math total_movies 250 movies_per_page 25 total_pages math.ceil(total_movies / movies_per_page) urls [ fhttps://movie.douban.com/top250?start{page * movies_per_page}filter for page in range(total_pages) ]2.2 关键数据定位与提取豆瓣页面的电影信息主要包含在div classitem元素中。每个电影条目包含电影名称中英文评分评价人数经典台词可能没有导演和主演信息上映年份和国家电影类型使用BeautifulSoup4提取这些信息的核心思路items soup.select(div.item) for item in items: title item.select_one(span.title).text rating item.select_one(span.rating_num).text # 其他字段类似处理3. 完整爬虫实现与反爬策略3.1 基础爬虫框架搭建我们构建一个面向对象的爬虫框架便于扩展和维护class DoubanTop250Spider: BASE_URL https://movie.douban.com/top250 HEADERS { User-Agent: Mozilla/5.0 (Windows NT 10.0; Win64; x64)..., Accept-Language: zh-CN,zh;q0.9, } def __init__(self): self.client httpx.Client(headersself.HEADERS, timeout10.0) def fetch_page(self, page: int) - str: params {start: (page - 1) * 25, filter: } response self.client.get(self.BASE_URL, paramsparams) response.raise_for_status() return response.text def parse_page(self, html: str) - list[dict]: # 解析逻辑实现 pass def run(self): for page in range(1, 11): html self.fetch_page(page) yield from self.parse_page(html)3.2 应对反爬机制的关键技巧豆瓣对爬虫有一定防护以下是几个实用对策请求头伪装设置合理的User-Agent添加Referer和Accept-Language头请求频率控制import random import time # 在请求间添加随机延迟 time.sleep(random.uniform(1.0, 3.0))IP轮换策略使用httpx的代理支持考虑付费代理服务如需要大规模爬取Cookie处理self.client httpx.Client( headersself.HEADERS, cookies{key: value}, # 从浏览器获取有效cookie follow_redirectsTrue )4. 数据清洗与存储方案4.1 数据清洗与规范化从网页抓取的数据通常需要清洗去除空白字符clean_text .join(text.split()) # 合并多个空白字符处理中英文标题def process_title(title_element): chinese title_element.text english title_element.find_next_sibling(span, class_other) return { chinese: chinese, english: english.text.strip() if english else None }评分数据转换rating float(rating_str) if rating_str else None4.2 存储方案比较与实现根据需求不同可以选择多种存储方式方案一CSV文件存储import csv def save_to_csv(movies: list[dict], filename: str): with open(filename, w, newline, encodingutf-8) as f: writer csv.DictWriter(f, fieldnamesmovies[0].keys()) writer.writeheader() writer.writerows(movies)方案二SQLite数据库存储import sqlite3 def init_db(filename: str): conn sqlite3.connect(filename) conn.execute( CREATE TABLE IF NOT EXISTS movies ( id INTEGER PRIMARY KEY, chinese_title TEXT NOT NULL, english_title TEXT, rating REAL, votes INTEGER, year INTEGER, directors TEXT, actors TEXT ) ) return conn方案三JSON格式存储import json def save_to_json(movies: list[dict], filename: str): with open(filename, w, encodingutf-8) as f: json.dump(movies, f, ensure_asciiFalse, indent2)5. 性能优化与高级技巧5.1 异步并发爬取实现利用httpx的异步特性大幅提升爬取速度async def fetch_all_pages(): async with httpx.AsyncClient(headersHEADERS) as client: tasks [fetch_page(client, page) for page in range(1, 11)] return await asyncio.gather(*tasks) async def fetch_page(client: httpx.AsyncClient, page: int): params {start: (page - 1) * 25, filter: } response await client.get(BASE_URL, paramsparams) response.raise_for_status() return response.text5.2 利用Python 3.11新特性优化代码模式匹配Pattern Matchingmatch movie.get(rating): case float(r) if r 9.0: print(经典电影) case float(r) if r 8.0: print(推荐观看) case _: print(一般电影)TOML配置支持import tomllib with open(config.toml, rb) as f: config tomllib.load(f)更快的错误处理try: response client.get(url) response.raise_for_status() except httpx.HTTPStatusError as e: print(f请求失败: {e.response.status_code})5.3 异常处理与日志记录健壮的爬虫需要完善的错误处理import logging logging.basicConfig( levellogging.INFO, format%(asctime)s - %(name)s - %(levelname)s - %(message)s ) logger logging.getLogger(__name__) def safe_parse(html: str): try: return parse_page(html) except Exception as e: logger.error(f解析失败: {str(e)}, exc_infoTrue) return []6. 项目扩展与实用建议6.1 扩展功能思路定时自动更新使用schedule或APScheduler库设置定时任务比较新旧数据只更新变化的条目数据可视化import matplotlib.pyplot as plt ratings [m[rating] for m in movies] plt.hist(ratings, bins10) plt.title(豆瓣Top250评分分布) plt.show()构建简单的Web界面使用FastAPI或Flask展示数据添加搜索和筛选功能6.2 生产环境部署建议容器化部署FROM python:3.11-slim WORKDIR /app COPY requirements.txt . RUN pip install -r requirements.txt COPY . . CMD [python, main.py]监控与告警使用Prometheus监控爬虫运行状态设置异常告警通知分布式扩展考虑使用Celery或Dask分发爬取任务使用Redis作为任务队列和结果存储在实际项目中我发现豆瓣对频繁请求比较敏感建议将爬取间隔设置为3-5秒并尽量在非高峰时段运行爬虫。对于关键业务数据可以考虑使用官方API如果有的话替代网页爬取。