Python 网络爬虫实战:从零到部署的完整流程

适用人群：初-中级 Python 开发者、数据分析师、运维／测试自动化工程师

工具栈：Python 3.11 + requests + BeautifulSoup / lxml + pandas + (可选) Selenium / Playwright

目录

1. 环境准备
2. 目标网站分析
3. 编写基础爬虫（requests + BS4）
4. 增强：并发爬取 & 反爬绕过
5. 数据持久化（CSV / MySQL / MongoDB）
6. 全流程异常处理与日志
7. 项目打包部署 & 定时任务
8. 合规与反爬道德守则

1 环境准备

python -m venv venv && source venv/bin/activate

pip install requests beautifulsoup4 lxml pandas tqdm

# 如需 JS 渲染：

pip install playwright && playwright install chromium

确保：pip >= 23，系统时间正确，否则 SSL 握手易报错。

2 目标网站分析（以某博客文章列表为例）

1. F12 打开开发者工具 → Network → Doc
2. 找到列表页 URL，观察分页参数：
3. 
   
4. https://example.com/page/1 → 规律 /page/{pageNo}
5. 右键 Copy > Copy selector 确定元素路径：

<h2 class="entry-title"><a href="文章链接">标题</a></h2>

3. 
   
4. 判断是否需要登录／JS 渲染。若纯 HTML，可用 requests；否则使用 Playwright。

3 基础爬虫示例

import requests, time, random

from bs4 import BeautifulSoup

from urllib.parse import urljoin

HEADERS = {

"User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) "

"AppleWebKit/537.36 (KHTML, like Gecko) Chrome/124 Safari/537.36"

}

def fetch_page(url: str) -> str:

resp = requests.get(url, headers=HEADERS, timeout=10)

resp.raise_for_status()

return resp.text

def parse_list(html: str, base: str) -> list[dict]:

soup = BeautifulSoup(html, "lxml")

for h2 in soup.select("h2.entry-title a"):

yield {

"title": h2.text.strip(),

"link": urljoin(base, h2["href"])

}

def main():

base = "https://example.com"

all_posts = []

for page in range(1, 6):

url = f"{base}/page/{page}"

html = fetch_page(url)

all_posts.extend(parse_list(html, base))

time.sleep(random.uniform(1, 3)) # 给服务器喘息

print(f"抓到 {len(all_posts)} 篇文章")

# TODO: 进一步抓取详情页 / 写入文件

if __name__ == "__main__":

main()

4 增强：并发爬取 & 反爬绕过

4-1 异步 + 协程

pip install httpx[http2] asyncio aiofiles

import asyncio, httpx, aiofiles, json

SEM = asyncio.Semaphore(10)

async def fetch(url, client):

async with SEM, client.get(url) as r:

r.raise_for_status()

return r.text

async def crawl(urls):

async with httpx.AsyncClient(headers=HEADERS, http2=True) as client:

tasks = [fetch(u, client) for u in urls]

return await asyncio.gather(*tasks)

# 调用: data = asyncio.run(crawl(url_list))

4-2 常见反爬应对

5 数据持久化

5-1 写 CSV / Excel

import pandas as pd

pd.DataFrame(all_posts).to_csv("posts.csv", index=False, encoding="utf-8-sig")

5-2 写 MySQL

import pymysql

conn = pymysql.connect(host="127.0.0.1", user="root", password="pwd", database="spider", charset="utf8mb4")

with conn.cursor() as cur:

cur.executemany("INSERT IGNORE INTO article(title,link) VALUES(%s,%s)",

[(d["title"], d["link"]) for d in all_posts])

conn.commit()

6 异常处理与日志

import logging, os

logging.basicConfig(

level=logging.INFO,

filename="spider.log",

format="%(asctime)s %(levelname)s %(message)s"

)

try:

html = fetch_page(url)

except (requests.Timeout, requests.HTTPError) as e:

logging.error("抓取失败 %s → %s", url, e)

7 部署与定时任务

Linux 系统 (crontab):

crontab -e

# 每日凌晨 2 点运行

0 2 * * * /usr/bin/python /home/spider/main.py >> /home/spider/cron.log 2>&1

Docker 打包：

FROM python:3.11-slim

COPY . /app

WORKDIR /app

RUN pip install -r requirements.txt

CMD ["python", "main.py"]

8 合法合规 & 道德守则

1. 尊重 robots.txt：若站点明确禁止抓取，请勿爬取。
2. 流量友好：控制并发、限速，避免压垮服务器。
3. 勿爬敏感/隐私信息：遵守 GDPR、网络安全法。
4. 遵照授权协议：对商业站点先取得书面许可。
5. 标注数据来源：二次发布数据时注明原站点。

结论

• 核心流程：确定目标 → 模拟请求 → 解析 → 存储 → 迭代优化
• 充分利用异步、并发、分布式队列 (Redis + RSMQ/Celery) 获得更高抓取速率。
• 安全与合规永远排第一；任何超限操作都可能导致法律风险。