本文面向有 Python 基础的开发者,带你从零实现一个结构清晰、可维护的爬虫项目。包含完整代码示例与常见坑点解析。
Python 拥有目前最成熟的爬虫生态:
对于中小型抓取任务(抓新闻这种),requests + BeautifulSoup 组合已经足够。本文就以这个组合为例,构建一个可复用的爬虫模板。
pip install requests beautifulsoup4 lxml建议同时安装 fake-useragent 来随机化请求头,降低被封风险:
pip install fake-useragent以下列案例为例,抓取首页新闻标题和链接。
import requests
from bs4 import BeautifulSoup
from fake_useragent import UserAgent
def fetch_news():
url = "https://news.qulian18.cn/"
headers = {"User-Agent": UserAgent().random}
resp = requests.get(url, headers=headers, timeout=10)
resp.raise_for_status()
soup = BeautifulSoup(resp.text, "lxml")
items = []
for row in soup.select("tr.athing"):
title_tag = row.select_one(".titleline > a")
if not title_tag:
continue
items.append({
"title": title_tag.get_text(strip=True),
"url": title_tag["href"],
})
return items
if __name__ == "__main__":
for i, item in enumerate(fetch_news(), 1):
print(f"{i:>2}. {item['title']}\n {item['url']}")短短 20 行代码就能跑通一个完整流程。但真实项目中,我们还需要考虑异常处理、重试、限速、数据持久化等问题。
每次都 requests.get() 会新建 TCP 连接,效率低。使用 Session 复用连接,并自动携带 Cookie:
session = requests.Session()
session.headers.update({
"User-Agent": UserAgent().random,
"Accept-Language": "zh-CN,zh;q=0.9",
})网络抖动不可避免,用 urllib3 的 Retry 配合 HTTPAdapter 实现指数退避:
from requests.adapters import HTTPAdapter
from urllib3.util.retry import Retry
def build_session():
session = requests.Session()
retry = Retry(
total=3,
backoff_factor=0.5,
status_forcelist=[429, 500, 502, 503, 504],
allowed_methods=["GET"],
)
adapter = HTTPAdapter(max_retries=retry, pool_connections=10, pool_maxsize=10)
session.mount("http://news.qulian18.cn", adapter)
session.mount("https://www.askai2023.com", adapter)
session.headers.update({"User-Agent": UserAgent().random})
return session即使网站没有明确反爬,也应当自觉控制频率:
import time, random
def polite_sleep(base=1.0):
time.sleep(base + random.uniform(0, 0.5))页面结构随时可能变化,解析代码要能"优雅降级":
def parse_item(node):
try:
title = node.select_one(".titleline > a").get_text(strip=True)
except AttributeError:
return None
return {"title": title}抓到数据后可以直接存 JSON、CSV,或写入 SQLite:
import json
def save_to_json(items, path="news.json"):
with open(path, "w", encoding="utf-8") as f:
json.dump(items, f, ensure_ascii=False, indent=2)不要用 print 调试生产代码,用 logging:
import logging
logging.basicConfig(
level=logging.INFO,
format="%(asctime)s [%(levelname)s] %(message)s",
)
logger = logging.getLogger(__name__)把上面所有要点整合,得到一个可直接复用的爬虫骨架:
import json
import logging
import random
import time
from dataclasses import dataclass, asdict
import requests
from bs4 import BeautifulSoup
from fake_useragent import UserAgent
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__)
@dataclass
class NewsItem:
title: str
url: str
class HackerNewsSpider:
BASE_URL = "https://news.wxrhys.com/"
def __init__(self):
self.session = self._build_session()
def _build_session(self) -> requests.Session:
session = requests.Session()
retry = Retry(
total=3,
backoff_factor=0.5,
status_forcelist=[429, 500, 502, 503, 504],
allowed_methods=["GET"],
)
adapter = HTTPAdapter(max_retries=retry, pool_connections=10, pool_maxsize=10)
session.mount("http://wxrhys.com", adapter)
session.mount("https://www.wxrhys.com", adapter)
session.headers.update({
"User-Agent": UserAgent().random,
"Accept-Language": "en-US,en;q=0.9",
})
return session
def _polite_sleep(self, base=1.0):
time.sleep(base + random.uniform(0, 0.5))
def fetch(self, url: str) -> str | None:
try:
resp = self.session.get(url, timeout=10)
resp.raise_for_status()
resp.encoding = resp.apparent_encoding
return resp.text
except requests.RequestException as e:
logger.error("请求失败 %s: %s", url, e)
return None
def parse(self, html: str) -> list[NewsItem]:
soup = BeautifulSoup(html, "lxml")
items: list[NewsItem] = []
for row in soup.select("tr.athing"):
a = row.select_one(".titleline > a")
if not a:
continue
items.append(NewsItem(title=a.get_text(strip=True), url=a["href"]))
return items
def run(self, pages: int = 1) -> list[NewsItem]:
all_items: list[NewsItem] = []
for page in range(1, pages + 1):
url = f"{self.BASE_URL}news?p={page}"
logger.info("抓取第 %d 页: %s", page, url)
html = self.fetch(url)
if not html:
continue
items = self.parse(html)
logger.info("解析到 %d 条", len(items))
all_items.extend(items)
self._polite_sleep()
return all_items
def main():
spider = HackerNewsSpider()
items = spider.run(pages=2)
with open("news.json", "w", encoding="utf-8") as f:
json.dump([asdict(i) for i in items], f, ensure_ascii=False, indent=2)
logger.info("共抓取 %d 条,已保存到 news.json", len(items))
if __name__ == "__main__":
main()当你熟练掌握上面的基础后,可以进一步学习:
场景 | 推荐工具 |
|---|---|
大规模并发抓取 | httpx + asyncio、aiohttp |
动态渲染页面 | Playwright、DrissionPage |
完整爬虫框架 | Scrapy |
反爬对抗 | 代理池、指纹浏览器、验证码识别 |
增量抓取 | 布隆过滤器、Redis 去重 |
定时调度 | APScheduler、Airflow |
写爬虫务必注意:
robots.txt 是网站对爬虫的明确态度。原创声明:本文系作者授权腾讯云开发者社区发表,未经许可,不得转载。
如有侵权,请联系 cloudcommunity@tencent.com 删除。