网络爬虫：数据抓取 (Web Scraping: Data Extraction)

章节概述

Python 爬虫是数据获取的核心工具——从网页中提取结构化数据。本章从 C 程序员视角出发，对比 C 的 socket/libcurl 手动拼 HTTP 与 Python 的 requests + BeautifulSoup/Scrapy 声明式抓取。

C 程序员视角：C 写爬虫需要手动管理 TCP 连接、拼接 HTTP 请求、解析 HTML 字符串。Python 的爬虫生态把这一切封装成了几行代码。

1. HTTP 请求基础：requests

pip install requests

import requests
 
# GET 请求
r = requests.get('https://httpbin.org/get')
print(r.status_code)   # 200
print(r.headers['Content-Type'])
print(r.text)          # 响应体（字符串）
print(r.json())        # 解析为 dict
 
# 带参数的 GET
r = requests.get('https://httpbin.org/get', params={'q': 'python', 'page': 1})
 
# POST 请求
r = requests.post('https://httpbin.org/post', data={'username': 'test', 'password': '123'})
 
# 自定义 Headers
headers = {
    'User-Agent': 'Mozilla/5.0 (X11; Linux x86_64) AppleWebKit/537.36',
    'Accept-Language': 'zh-CN,zh;q=0.9',
}
r = requests.get('https://example.com', headers=headers)
 
# Session（保持 Cookie）
session = requests.Session()
session.get('https://example.com/login', params={'user': 'test', 'pass': '123'})
r = session.get('https://example.com/dashboard')  # 自动带 Cookie

2. HTML 解析：BeautifulSoup

pip install beautifulsoup4 lxml

from bs4 import BeautifulSoup
import requests
 
r = requests.get('https://example.com')
soup = BeautifulSoup(r.text, 'lxml')
 
# 查找元素
title = soup.find('h1').text
links = soup.find_all('a')
for link in links:
    print(link.get('href'), link.text)
 
# CSS 选择器
items = soup.select('div.content > p.intro')
items = soup.select('[data-id="123"]')
 
# 提取属性
for img in soup.find_all('img'):
    print(img.get('src'), img.get('alt'))

3. 完整爬虫示例：爬取新闻标题

import requests
from bs4 import BeautifulSoup
import csv
import time
 
def scrape_news(url):
    headers = {'User-Agent': 'Mozilla/5.0'}
    r = requests.get(url, headers=headers, timeout=10)
    soup = BeautifulSoup(r.text, 'lxml')
 
    articles = []
    for item in soup.select('article'):
        title = item.select_one('h2')
        link = item.select_one('a')
        if title and link:
            articles.append({
                'title': title.text.strip(),
                'url': link.get('href'),
            })
    return articles
 
def save_csv(data, filename):
    with open(filename, 'w', newline='', encoding='utf-8') as f:
        writer = csv.DictWriter(f, fieldnames=['title', 'url'])
        writer.writeheader()
        writer.writerows(data)
 
# 使用
articles = scrape_news('https://example.com/news')
save_csv(articles, 'news.csv')
print(f"爬取了 {len(articles)} 篇文章")

4. 处理动态渲染页面：Selenium / Playwright

pip install playwright
playwright install chromium

from playwright.sync_api import sync_playwright
 
with sync_playwright() as p:
    browser = p.chromium.launch(headless=True)
    page = browser.new_page()
    page.goto('https://example.com')
 
    # 等待元素加载
    page.wait_for_selector('div.content')
 
    # 滚动加载更多
    page.evaluate('window.scrollTo(0, document.body.scrollHeight)')
    page.wait_for_timeout(2000)
 
    # 提取数据
    items = page.query_selector_all('.item')
    for item in items:
        print(item.inner_text())
 
    # 截图
    page.screenshot(path='screenshot.png')
 
    browser.close()

5. 爬虫礼仪与反爬对策

robots.txt 遵守

from urllib.robotparser import RobotFileParser
 
rp = RobotFileParser()
rp.set_url('https://example.com/robots.txt')
rp.read()
 
if rp.can_fetch('*', 'https://example.com/page'):
    # 可以爬
    pass

反爬对策

import time
import random
 
# 1. 随机 User-Agent
user_agents = [
    'Mozilla/5.0 (Windows NT 10.0; Win64; x64)',
    'Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7)',
    'Mozilla/5.0 (X11; Linux x86_64)',
]
headers = {'User-Agent': random.choice(user_agents)}
 
# 2. 随机延时
time.sleep(random.uniform(1, 3))
 
# 3. 使用代理
proxies = {'http': 'http://proxy:8080', 'https': 'http://proxy:8080'}
r = requests.get(url, headers=headers, proxies=proxies)
 
# 4. Session 保持连接
session = requests.Session()

6. 数据存储

# CSV
import csv
with open('data.csv', 'w', newline='') as f:
    writer = csv.DictWriter(f, fieldnames=['title', 'url'])
    writer.writeheader()
    writer.writerows(data)
 
# JSON
import json
with open('data.json', 'w') as f:
    json.dump(data, f, ensure_ascii=False, indent=2)
 
# SQLite
import sqlite3
conn = sqlite3.connect('data.db')
c = conn.cursor()
c.execute('CREATE TABLE IF NOT EXISTS articles (title TEXT, url TEXT)')
c.executemany('INSERT INTO articles VALUES (?, ?)', [(d['title'], d['url']) for d in data])
conn.commit()
conn.close()
 
# Pandas（最方便）
import pandas as pd
df = pd.DataFrame(data)
df.to_csv('data.csv', index=False)
df.to_json('data.json', force_ascii=False)

7. Scrapy 框架（大规模爬虫）

pip install scrapy
scrapy startproject myspider

# myspider/spiders/news.py
import scrapy
 
class NewsSpider(scrapy.Spider):
    name = 'news'
    start_urls = ['https://example.com/news']
 
    def parse(self, response):
        for article in response.css('article'):
            yield {
                'title': article.css('h2::text').get(),
                'url': article.css('a::attr(href)').get(),
            }
 
        # 翻页
        next_page = response.css('a.next::attr(href)').get()
        if next_page:
            yield response.follow(next_page, self.parse)

# 运行爬虫
scrapy crawl news -o output.json

7. 批量下载文件：下载游戏语音资源

爬虫不只是抓 HTML——批量下载图片、音频、视频等二进制文件也是常见需求。

7.1 使用 urllib 下载单个文件

import urllib.request
 
url = "https://example.com/image.png"
urllib.request.urlretrieve(url, "image.png")  # 一行搞定

7.2 使用 requests 下载（推荐）

import requests
 
url = "https://example.com/video.mp4"
r = requests.get(url, stream=True)  # stream=True 大文件不一次性加载到内存
 
with open("video.mp4", "wb") as f:
    for chunk in r.iter_content(chunk_size=8192):
        f.write(chunk)

7.3 实战：下载明日方舟角色中文语音

场景：从 PRTS Wiki 下载指定干员的全部中文语音 WAV 文件。

#!/usr/bin/env python3
"""下载琳琅诗怀雅所有中文语音 WAV (16-bit, 44.1kHz, mono)"""
 
import os
import urllib.request
 
BASE_URL = "https://torappu.prts.wiki/assets/audio/voice_cn/char_1033_swire2"
 
# (编号, 中文名)
VOICES = [
    (1, "任命助理"),
    (2, "交谈1"),
    (3, "交谈2"),
    (4, "交谈3"),
    (5, "晋升后交谈1"),
    (6, "晋升后交谈2"),
    (7, "信赖提升后交谈1"),
    (8, "信赖提升后交谈2"),
    (9, "信赖提升后交谈3"),
    (10, "闲置"),
    (11, "干员报到"),
    (12, "观看作战记录"),
    (13, "精英化晋升1"),
    (14, "精英化晋升2"),
    (17, "编入队伍"),
    (18, "任命队长"),
    (19, "行动出发"),
    (20, "行动开始"),
    (21, "选中干员1"),
    (22, "选中干员2"),
    (23, "部署1"),
    (24, "部署2"),
    (25, "作战中1"),
    (26, "作战中2"),
    (27, "作战中3"),
    (28, "作战中4"),
    (29, "完成高难行动"),
    (30, "3星结束行动"),
    (31, "非3星结束行动"),
    (32, "行动失败"),
    (33, "进驻设施"),
    (34, "戳一下"),
    (36, "信赖触摸"),
    (37, "标题"),
    (38, "新年祝福"),
    (42, "问候"),
    (43, "生日"),
    (44, "周年庆典"),
]
 
SAVE_DIR = os.path.join(os.path.dirname(os.path.abspath(__file__)), "linglang_swire_voice_cn")
os.makedirs(SAVE_DIR, exist_ok=True)
 
headers = {
    "User-Agent": "Mozilla/5.0 (X11; Linux x86_64) AppleWebKit/537.36",
    "Referer": "https://prts.wiki/",
}
 
success = fail = 0
 
for num, name in VOICES:
    filename = f"cn_{num:03d}.wav"
    url = f"{BASE_URL}/{filename}"
    save_path = os.path.join(SAVE_DIR, f"{name}.wav")
 
    if os.path.exists(save_path):
        print(f"[跳过] {name}.wav 已存在")
        success += 1
        continue
 
    try:
        req = urllib.request.Request(url, headers=headers)
        with urllib.request.urlopen(req, timeout=30) as resp:
            data = resp.read()
        with open(save_path, "wb") as f:
            f.write(data)
        print(f"[OK] {name}.wav ({len(data)} bytes)")
        success += 1
    except Exception as e:
        print(f"[失败] {name}.wav: {e}")
        fail += 1
 
print(f"\n完成！成功: {success}，失败: {fail}")
print(f"保存目录: {SAVE_DIR}")

7.4 代码解析

技术点	说明
`urllib.request.Request`	构造请求对象，可设置 headers
`headers["Referer"]`	防盗链——告诉服务器请求来自哪个页面
`headers["User-Agent"]`	伪装浏览器，避免被拦截
`os.path.exists`	跳过已下载的文件（断点续传）
`timeout=30`	超时 30 秒，避免卡死
`resp.read()`	读取全部响应体为 bytes
`os.makedirs(..., exist_ok=True)`	递归创建目录，已存在不报错

7.5 改进方向

# 1. 添加进度显示
import sys
sys.stdout.write(f"\r[{i+1}/{total}] {name}...")
 
# 2. 使用 requests + 流式下载（大文件）
import requests
r = requests.get(url, headers=headers, stream=True)
with open(save_path, "wb") as f:
    for chunk in r.iter_content(chunk_size=8192):
        f.write(chunk)
 
# 3. 并发下载（加速）
from concurrent.futures import ThreadPoolExecutor
with ThreadPoolExecutor(max_workers=5) as pool:
    pool.map(download_one, VOICES)
 
# 4. 添加重试机制
import time
for attempt in range(3):
    try:
        # 下载逻辑
        break
    except Exception:
        time.sleep(2 ** attempt)  # 指数退避

速查卡片

需求	工具/命令
HTTP 请求	`requests.get(url)`
HTML 解析	`BeautifulSoup(html, 'lxml')`
CSS 选择	`soup.select('div.class')`
动态页面	`playwright` / `selenium`
大规模爬虫	`scrapy`
存储 CSV	`df.to_csv('data.csv')`
存储 JSON	`json.dump(data, f)`
存储数据库	`sqlite3` / `SQLAlchemy`

RootStack

探索

01_网络爬虫：数据抓取

网络爬虫：数据抓取 (Web Scraping: Data Extraction)

章节概述

1. HTTP 请求基础：requests

2. HTML 解析：BeautifulSoup

3. 完整爬虫示例：爬取新闻标题

4. 处理动态渲染页面：Selenium / Playwright

5. 爬虫礼仪与反爬对策

robots.txt 遵守

反爬对策

6. 数据存储

7. Scrapy 框架（大规模爬虫）

7. 批量下载文件：下载游戏语音资源

7.1 使用 urllib 下载单个文件

7.2 使用 requests 下载（推荐）

7.3 实战：下载明日方舟角色中文语音

7.4 代码解析

7.5 改进方向

速查卡片

关系图谱

目录

反向链接