一、基础概念与网络请求
1.1 请求头与伪装成浏览器
一些网站为了防止爬虫，会有检查机制。我们需要添加请求头，将 Python 伪装成浏览器，常见的键有 User-Agent、Referer、Accept、Cookie 等。
其中的 Accept-Encoding 用于声明客户端支持的压缩方式，如 gzip、deflate、br（Brotli）。告诉服务器后，服务器会返回对应编码的响应内容以减小体积。
response.content 会自动对 gzip、deflate 解码，返回字节类型。response.text 会根据响应头自动解码，返回字符串类型。
注意：requests 默认不支持对 brotli 格式（即 br）进行解码。要么删除 Accept-Encoding 中的 br，要么安装 brotli 包后手动处理：
1
2
3
4
5
6
7
8
|
import brotli
# 先获取响应内容
response = requests.get(...)
# 对 brotli 格式压缩后的二进制数据进行解码（注意是 .content）
if 'Content-Encoding' in response.headers and \
response.headers['Content-Encoding'] == 'br':
data = brotli.decompress(response.content) \
.decode('utf-8')
|
1.2 编码与解码
encode('utf-8') 将 Unicode 字符串按 UTF-8 编码成字节流，用于写入文件或网络传输。decode('utf-8') 将字节流按 UTF-8 解码回 Unicode 字符串，用于文本处理。
在 Python 中，可以使用 u 前缀表示 Unicode 字符串。每个字符对应一个 Unicode 码点，能跨越不同语言和字符集：
1
2
|
# 一个包含英文、中文和表情符号的 Unicode 字符串
unicode_str = u"Hello, 你好，🌟"
|
注意：Python 3 中的字符串默认就是 Unicode 字符串，通常不需要显式使用 u 前缀。
1.3 Cookie 与会话保持
Cookie 由服务器端创建，存储在本地浏览器，下次访问时自动调用，相当于通行证，但存在有效期。可以使用 http.cookiejar 模块与 requests 配合，自动在多次请求间管理 Cookie：
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
|
import requests
import http.cookiejar
# 创建一个 CookieJar 对象
cookie_jar = http.cookiejar.CookieJar()
# 创建一个 HTTPCookieProcessor 对象
cookie_processor = requests.cookies.RequestsCookieJar()
# 将 CookieJar 对象添加到 HTTPCookieProcessor 中
cookie_processor.set_cookiejar(cookie_jar)
# 创建一个 Session 对象
session = requests.Session()
# 将 HTTPCookieProcessor 对象添加到 Session 的 mount 中
session.mount('https://', requests.adapters.HTTPAdapter(max_retries=3))
session.mount('http://', requests.adapters.HTTPAdapter(max_retries=3))
session.mount('ftp://', requests.adapters.HTTPAdapter(max_retries=3))
session.mount('file://', requests.adapters.HTTPAdapter(max_retries=3))
session.mount('data:', requests.adapters.HTTPAdapter(max_retries=3))
session.mount('ws://', requests.adapters.HTTPAdapter(max_retries=3))
session.mount('wss://', requests.adapters.HTTPAdapter(max_retries=3))
session.mount('ws:', requests.adapters.HTTPAdapter(max_retries=3))
session.mount('wss:', requests.adapters.HTTPAdapter(max_retries=3))
session.mount('ftp:', requests.adapters.HTTPAdapter(max_retries=3))
session.mount('sftp:', requests.adapters.HTTPAdapter(max_retries=3))
session.mount('http:', requests.adapters.HTTPAdapter(max_retries=3))
session.mount('https:', requests.adapters.HTTPAdapter(max_retries=3))
session.mount('file:', requests.adapters.HTTPAdapter(max_retries=3))
# 发送请求
response = session.get('https://www.example.com')
# 打印响应内容
print(response.text)
|
优点：保持会话状态，保证每个请求携带正确的 cookie，简化处理过程。也可以通过工具（如 uutool.cn/header2json）将浏览器中的 Cookie 字符串转成 JSON 手动携带。
1.4 IP 代理与状态码
为避免单一 IP 被封禁或限流，需配置代理并在 headers 中随机切换 User-Agent。若直接请求被反爬拦截，可能收到 418 状态码；若收到 403，说明 IP 可能被拉黑，需更换 IP 或降低速度。
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
|
user_agent=[
"Mozilla/5.0 (Macintosh; U; Intel Mac OS X 10_6_8; en-us) AppleWebKit/534.50 (KHTML, like Gecko) Version/5.1 Safari/534.50",
"Mozilla/5.0 (Windows; U; Windows NT 6.1; en-us) AppleWebKit/534.50 (KHTML, like Gecko) Version/5.1 Safari/534.50",
"Mozilla/5.0 (Windows NT 10.0; WOW64; rv:38.0) Gecko/20100101 Firefox/38.0",
"Mozilla/5.0 (Windows NT 10.0; WOW64; Trident/7.0; .NET4.0C; .NET4.0E; .NET CLR 2.0.50727; .NET CLR 3.0.30729; .NET CLR 3.5.30729; InfoPath.3; rv:11.0) like Gecko",
"Mozilla/5.0 (compatible; MSIE 9.0; Windows NT 6.1; Trident/5.0)",
"Mozilla/4.0 (compatible; MSIE 8.0; Windows NT 6.0; Trident/4.0)",
"Mozilla/4.0 (compatible; MSIE 7.0; Windows NT 6.0)",
"Mozilla/4.0 (compatible; MSIE 6.0; Windows NT 5.1)",
"Mozilla/5.0 (Macintosh; Intel Mac OS X 10.6; rv:2.0.1) Gecko/20100101 Firefox/4.0.1",
"Mozilla/5.0 (Windows NT 6.1; rv:2.0.1) Gecko/20100101 Firefox/4.0.1",
"Opera/9.80 (Macintosh; Intel Mac OS X 10.6.8; U; en) Presto/2.8.131 Version/11.11",
"Opera/9.80 (Windows NT 6.1; U; en) Presto/2.8.131 Version/11.11",
"Mozilla/5.0 (Macintosh; Intel Mac OS X 10_7_0) AppleWebKit/535.11 (KHTML, like Gecko) Chrome/17.0.963.56 Safari/535.11",
"Mozilla/4.0 (compatible; MSIE 7.0; Windows NT 5.1; Maxthon 2.0)",
"Mozilla/4.0 (compatible; MSIE 7.0; Windows NT 5.1; TencentTraveler 4.0)",
"Mozilla/4.0 (compatible; MSIE 7.0; Windows NT 5.1)",
"Mozilla/4.0 (compatible; MSIE 7.0; Windows NT 5.1; The World)",
"Mozilla/4.0 (compatible; MSIE 7.0; Windows NT 5.1; Trident/4.0; SE 2.X MetaSr 1.0; SE 2.X MetaSr 1.0; .NET CLR 2.0.50727; SE 2.X MetaSr 1.0)",
"Mozilla/4.0 (compatible; MSIE 7.0; Windows NT 5.1; 360SE)",
"Mozilla/4.0 (compatible; MSIE 7.0; Windows NT 5.1; Avant Browser)",
"Mozilla/4.0 (compatible; MSIE 7.0; Windows NT 5.1)",
"Mozilla/5.0 (iPhone; U; CPU iPhone OS 4_3_3 like Mac OS X; en-us) AppleWebKit/533.17.9 (KHTML, like Gecko) Version/5.0.2 Mobile/8J2 Safari/6533.18.5",
"Mozilla/5.0 (iPod; U; CPU iPhone OS 4_3_3 like Mac OS X; en-us) AppleWebKit/533.17.9 (KHTML, like Gecko) Version/5.0.2 Mobile/8J2 Safari/6533.18.5",
"Mozilla/5.0 (iPad; U; CPU OS 4_3_3 like Mac OS X; en-us) AppleWebKit/533.17.9 (KHTML, like Gecko) Version/4.0.2 Mobile/8J2 Safari/6533.18.5",
"Mozilla/5.0 (Linux; U; Android 2.3.7; en-us; Nexus One Build/FRF91) AppleWebKit/533.1 (KHTML, like Gecko) Version/4.0 Mobile Safari/533.1",
"MQQBrowser/26 Mozilla/5.0 (Linux; U; Android 2.3.7; zh-cn; MB200 Build/GRJ22; CyanogenMod-7) AppleWebKit/533.1 (KHTML, like Gecko) Version/4.0 Mobile Safari/533.1",
"Opera/9.80 (Android 2.3.4; Linux; Opera Mobi/build-1107180945; U; en-GB) Presto/2.8.149 Version/11.10",
"Mozilla/5.0 (Linux; U; Android 3.0; en-us; Xoom Build/HRI39) AppleWebKit/534.13 (KHTML, like Gecko) Version/4.0 Safari/534.13",
"Mozilla/5.0 (BlackBerry; U; BlackBerry 9800; en) AppleWebKit/534.1+ (KHTML, like Gecko) Version/6.0.0.337 Mobile Safari/534.1+",
"Mozilla/5.0 (hp-tablet; Linux; hpwOS/3.0.0; U; en-US) AppleWebKit/534.6 (KHTML, like Gecko) wOSBrowser/233.70 Safari/534.6 TouchPad/1.0",
"Mozilla/5.0 (SymbianOS/9.4; Series60/5.0 NokiaN97-1/20.0.019; Profile/MIDP-2.1 Configuration/CLDC-1.1) AppleWebKit/525 (KHTML, like Gecko) BrowserNG/7.1.18124",
"Mozilla/5.0 (compatible; MSIE 9.0; Windows Phone OS 7.5; Trident/5.0; IEMobile/9.0; HTC; Titan)",
"UCWEB7.0.2.37/28/999",
"NOKIA5700/ UCWEB7.0.2.37/28/999",
"Openwave/ UCWEB7.0.2.37/28/999",
"Mozilla/4.0 (compatible; MSIE 6.0; ) Opera/UCWEB7.0.2.37/28/999",
"Mozilla/6.0 (iPhone; CPU iPhone OS 8_0 like Mac OS X) AppleWebKit/536.26 (KHTML, like Gecko) Version/8.0 Mobile/10A5376e Safari/8536.25",
"Mozilla/5.0 (Windows NT 10.0; WOW64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/63.0.3239.132 Safari/537.36 QIHU 360SE",
]
headers={
'Cookie': '',
'Host': '',
'Referer': '',
'User-Agent': random.choice(user_agent)
}
response = requests.get(
url = '......',
headers = headers,
# 添加代理 proxy，你的代理服务器 IP
proxies = {
'https': 'https://{}'.format(proxy),
'http': 'http://{}'.format(proxy)
},
timeout = 30
)
|
1.5 请求超时
timeout 参数用于控制等待时间，常见为元组形式 (connect_timeout, read_timeout)。
- 连接超时：从发起请求到建立连接的最大等待时间。
- 读取超时：连接建立后，等待服务端返回数据的最大时间。
示例：
timeout=(10, 15)。
二、数据提取与解析
2.1 BeautifulSoup 解析库
BeautifulSoup 是常用的 HTML 解析库。lxml 解析速度快但需安装；html.parser 自带无需安装。
soup.find(name, attrs) 只返回第一个匹配对象，可搭配 .get_text() 获取文本。soup.find_all() 返回所有符合条件的列表，可添加 limit 参数。soup.select('class路径') 返回迭代器，可结合 for 循环与 .find() 使用。
1
2
3
4
5
6
7
8
9
10
|
# 打印 table 标签（id等于giftList）下的子节点
for child in bsObj.find("table",{"id":"giftList"}).children:
print(child)
# 同理有处理父标签的方法 .parent
# table#giftList -> tr 的后一兄弟标签
bsObj.find("table",{"id":"giftList"}).tr.next_siblings
# 同理有 previous_siblings() 前一兄弟标签
# 获取标签的属性
# .attrs 返回一个字典
Tag.attrs["src"]
|
2.2 XPath 与 lxml
XPath 常用组件：tag(标签)、*(所有)、.(当前)、//(所有后代)、..(父级)、[@attrib='value'](属性匹配)、[position](位置，如 last())。
1
2
3
4
5
|
import requests
from lxml import etree
r = requests.get(url= , headers=)
tree = etree.HTML(r.text)
tree.xpath()
|
2.3 正则表达式与原始字符串
正则中 group() 返回整个匹配或指定子组，groups() 以元组返回所有子组。r 表示原始字符串，不对反斜杠进行转义处理。
可以识别邮箱地址的正则（字母数字点加下划线 + @ + 字母 + 点 + 顶级域名）：
[A-Za-z0-9\.+_]+@[A-Za-z]+\.(com|org|edu|net)
三、JS 渲染与自动化
3.1 基础用法
Selenium 多用于需要执行 JS、模拟交互等自动化任务，能模拟鼠标移动、点击和键盘输入：
1
2
3
4
5
6
7
8
|
from selenium import webdriver
from selenium.webdriver.common.by import By
from selenium.webdriver.common.keys import Keys
from selenium.webdriver.support.wait import WebDriverWait
from selenium.webdriver.support import expected_conditions as EC
from selenium.webdriver import ActionChains
import pyautogui
from lxml import html
|
1
2
3
4
5
6
7
8
9
10
11
12
13
14
|
driver = webdriver.Edge("driver的路径")
# 最大化窗口
driver.maximize_window()
# 自定义窗口大小
driver.set_window_size(1280, 720)
# 设置等待时间（最长20秒，每1秒检测一次）
wait = WebDriverWait(driver, 20, 1)
# 发送 get 请求
driver.get(url)
# 获取当前窗口 / 所有窗口
currentwindow = driver.current_window_handle
all_window = driver.window_handles
# 切换窗口
driver.switch_to.window('某个窗口对象')
|
1
2
3
4
5
6
7
8
9
10
11
12
|
# 查找元素（等到元素加载完）
article_loc = wait.until(EC.presence_of_element_located((By.XPATH, article_string)))
# 查找元素（直接查找）
article_loc = driver.find_element(by = By.XPATH, value="//*[@id='tbody']/tr[1]/td[1]/a")
# 获取属性与模拟动作
article_link = article_loc.get_attribute(name='href')
ActionChains(driver).move_to_element(article_loc).perform()
article_loc.click()
ActionChains(driver).send_keys(Keys.SPACE).perform() #传入空格键
# 控制鼠标点击相应坐标
pyautogui.moveTo(1380, 650)
pyautogui.click()
|
3.2 花瓣网图片爬取
对于 JS 动态渲染的页面，可通过 selenium 获取加载后的 html，再利用 BeautifulSoup 解析：
1
2
3
4
5
6
|
from selenium import webdriver
from selenium.webdriver.common.keys import Keys
driver = webdriver.Edge()
driver.get(url)
html = driver.page_source
soup = BeautifulSoup(html, 'html.parser')
|
四、文本与列表型实战
4.1 教务处课程成绩爬取
流程：准备 headers (含 useragent, referer 等) -> 设置 cookie -> requests.get (设置 verify=False) -> BeautifulSoup 解析提取数据 -> 存入列表转 pd.DataFrame -> 导出表格 df.to_excel(filename, index=False, encoding='utf-8')。
4.2 豆瓣电影榜 Top 250
目标网站含反爬，直接 get 会返回 418。需带上浏览器 Cookie 后得到 200。若遇到乱码，多为压缩数据，可注释掉 header 中的 Accept-Encoding。使用 tqdm 显示进度：
1
2
3
4
5
6
7
|
from tqdm import tqdm # 进度条库
# 每页 25 部电影，故爬取 10 页即可
for i in tqdm(range(10)):
url = 'https://movie.douban.com/top250?start={}&filter='.format(num)
response = requests.get(url, headers=header)
# 爬取过程略。。。
num += 25
|
4.3 亚马逊商品爬取
利用请求头带上 cookie，再通过美味汤进行内容解析。
4.4 POST 表单提交
页面表单本质是 POST 请求，无论字段多复杂（单选、下拉等），只关注字段名称和值：
1
2
3
4
5
6
7
8
9
|
import requests
params = {'firstname': 'Ryan', 'lastname': 'Mitchell'}
r = requests.post("http://pythonscraping.com/files/processing.php", data=params)
print(r.text)
# 将编码形式赋值为从响应正文分析出的编码方式
r.encoding = r.apparent_encoding #或者直接 'utf-8'
# 转为 json 数据
result = r.json()
result['...']['...']['...']
|
4.5 CSV 文件写入
newline='' 用于告诉 Python 写入文件时不要自动插入额外的换行符（避免 Windows 下出现空行），编码用 utf-8-sig 防止 Excel 乱码：
1
2
3
4
5
6
7
8
9
|
import csv
# 创建文件并写入表头
with open('中国社会组织_疫情防控.csv', 'w+', newline='', encoding='utf-8-sig') as fp:
writer = csv.writer(fp)
writer.writerow(("标题", "时间", "URL", "正文内容", "来源"))
# 追加写入数据
with open('中国社会组织_疫情防控.csv', 'a+', newline='', encoding='utf-8-sig') as fp:
writer = csv.writer(fp)
writer.writerow((title, publish_time, article_link, article_text, source))
|
五、多媒体文件下载
5.1 B 站视频下载
最好调用 you-get。也可利用 re 模块提取标题和播放信息，分块下载视频和音频，最后用 moviepy 合并。注意以二进制形式(wb)写入不会进行编码转换，适用于音视频等文件。
1
2
3
4
5
6
|
import re
import requests
from tqdm import tqdm
response = requests.get(url, headers=headers)
title = re.findall(r'<h1 title="(.*?)" class="video-title tit">', response.text)[0]
playinfo = re.findall(r'<script>window.__playinfo__=(.*?)</script>', response.text)[0]
|
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
|
# 分块写入
response = requests.get(url, headers=headers)
print('HTTP 状态码：', response.status_code)
chunk_size = 1024
file_size = int(response.headers['content-length']) # 单位字节 Byte
file_size_MB = file_size/1024/1024 # 转换为 MB
print('文件大小：{:.2f} MB'.format(file_size_MB))
temp = '待写入的文件名'
file_name = sys.path[0] + '\\' + temp
# wb 表示二进制形式覆盖写入
with open(file_name, mode='wb') as f, tqdm (
desc = file_name,
total = file_size_MB,
unit = 'iB',
unit_scale = True,
unit_divisor = chunk_size,
) as bar:
for chunk in response.iter_content(chunk_size=chunk_size):
mysize = f.write(chunk)
bar.update(mysize)
|
1
2
3
4
5
6
7
8
9
|
# 音视频合并 (pip3 install moviepy 并重启 vscode)
from moviepy import *
from moviepy.editor import *
video_path = '视频所在路径'
audio_path = '音频所在路径'
video = VideoFileClip(video_path)
audio = AudioFileClip(audio_path)
video = video.set_audio(audio)
video.write_videofile('新文件名', fps=video.fps)
|
5.2 樱花动漫下载 (m3u8 与多线程)
流程：获取 iframe 中的 src -> 提取 m3u8_1 -> 提取包含所有 ts 的 m3u8_2 -> 多线程下载 ts -> ffmpeg 拼接 mp4 (需加 -safe 0 参数)。
前期准备与配置：
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
|
import os
import re
import pathlib
import shutil
import random
import threading
import subprocess
from queue import Queue
import requests
import urllib.parse
from selenium import webdriver
from selenium.webdriver.edge.options import Options
from selenium.webdriver.common.by import By
from selenium.webdriver.support.ui import WebDriverWait
from selenium.webdriver.support import expected_conditions as EC
DRIVER_PATH = "D:/python/msedgedriver.exe"
USERAGENT_ARR = [
'Mozilla/5.0 (Windows NT 6.1; Win64; x64; rv:47.0) Gecko/20100101 Firefox/47.0',
'Mozilla/5.0 (Macintosh; Intel Mac OS X 10.13; rv:63.0) Gecko/20100101 Firefox/63.0',
'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/78.0.3904.97 Safari/537.36',
]
|
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
|
def create_dir(dirname):
if dirname not in os.listdir():
cache_path = os.path.join(os.getcwd(), dirname)
pathlib.Path(cache_path).mkdir(parents=True, exist_ok=True)
print(f'文件夹 {dirname} 创建成功（或已存在）')
def remove_dir_files(dirname):
if dirname in os.listdir():
shutil.rmtree(dirname)
print(f'文件夹 {dirname} 删除成功')
def generate_headers():
headers = {
'Accept': 'text/html,application/xhtml+xml,application/xml;q=0.9,image/webp,image/apng,*/*;q=0.8,application/signed-exchange;v=b3;q=0.7',
'Accept-Encoding': 'gzip, deflate, br',
'Connection': 'keep-alive',
'user-agent': random.choice(USERAGENT_ARR),
}
return headers
def add_driver_options(options):
browser_options = Options()
for opt in options:
browser_options.add_argument(opt)
return browser_options
def initialize_driver():
driver_config = {"options": ['--headless', '--allow-insecure-localhost', 'blink-settings=imagesEnabled=false']}
options = add_driver_options(driver_config["options"])
useragent = random.choice(USERAGENT_ARR)
options.add_argument('user-agent=' + useragent)
driver = webdriver.Edge(DRIVER_PATH, options=options)
return driver
|
提取 m3u8 链接：
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
|
import requests
import urllib.parse
from selenium import webdriver
from selenium.webdriver.edge.options import Options
from selenium.webdriver.common.by import By
from selenium.webdriver.support.ui import WebDriverWait
from selenium.webdriver.support import expected_conditions as EC
def get_iframe_src(driver):
src_url = None
try:
WebDriverWait(driver, 10).until(EC.presence_of_element_located((By.ID, 'yh_playfram')))
iframe_tag = driver.find_elements(By.ID, 'yh_playfram')
src_url = iframe_tag[0].get_attribute('src')
except Exception as e:
print('driver出现错误: ', e)
finally:
driver.quit()
print('driver已退出')
return src_url
def get_m3u8_1(src_url):
decoded_url = urllib.parse.unquote(src_url)
m3u8_url1 = urllib.parse.parse_qs(decoded_url)['url'][0]
return m3u8_url1
def get_m3u8_2(m3u8_url1):
pattern = 'https://[a-zA-Z0-9.-]+/[0-9]+/[a-zA-Z0-9_]+/'
front_url = re.search(pattern, m3u8_url1).group()
resp = requests.get(url=m3u8_url1, headers=generate_headers())
for line in resp.text.split('\n'):
if 'm3u8' in line:
m3u8_url2 = line
break
return front_url + m3u8_url2
|
多线程下载：
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
|
import threading
import subprocess
from queue import Queue
def generate_ts_txt(m3u8_url2):
"""
通过m3u8_url2中的内容将ts文件的顺序记录在 tsfiles.txt 中
创建队列，保存ts文件的链接 ts_queue = Queue(10000)
通过逐行判断 .ts 和 http 是否在其中
"""
pass
def multi_threads_dl(ts_queue):
num = ts_queue.qsize()
t_num = num//5 if num>5 else 1
if t_num > 30: t_num = 30 #最大线程数为30
threads = []
for i in range(t_num):
t = threading.Thread(target = run, name = 'th-'+str(i), kwargs = {'ts_queue': ts_queue})
t.setDaemon(True)
threads.append(t)
for t in threads: t.start()
for t in threads: t.join()
def run(ts_queue):
while not ts_queue.empty():
url = ts_queue.get()
filename = re.search('([a-zA-Z0-9_-]+.ts)', url).group(1).strip()
try:
if os.path.exists( os.path.join(os.getcwd(), 'cache', filename) ):
print(f'{url} 已下载')
else:
requests.packages.urllib3.disable_warnings()
# 注意这里的分块下载
resp = requests.get(url, stream=True, headers=generate_headers(), verify=False, timeout=(10, 15))
with open('.\\cache\\' + filename, 'wb') as fp:
for chunk in resp.iter_content(2048):
if chunk:
fp.write(chunk)
print(f'{url} 下载成功')
except Exception as e:
print('下载失败: ', e)
ts_queue.put(url)
|
ffmpeg 合并：
1
2
3
4
5
6
7
8
9
|
def merge_tsfiles(concatfile, videoname, episode):
try:
create_dir(videoname)
path = f'.\\{videoname}\\{videoname}-第{episode}集.mp4'
command = ['ffmpeg', '-y', '-f', 'concat', '-safe', '0', '-i', concatfile, '-bsf:a', 'aac_adtstoasc', '-c', 'copy', path]
res = subprocess.run(command)
print('视频合并完成\r', res)
except Exception as e:
print('合并出现问题: ', e)
|
六、反爬机制与对抗进阶
6.1 反调试机制
打开 F12 时不断触发 debugger 导致无法正常观察 Network。
应对思路：添加条件断点输入 false；或下载 js 文件去掉 debugger 后用 Fiddler 替换。
6.2 JS 逆向与加密参数
本质是通过开发者工具检查访问所需参数的生成逻辑。例如使用 hashlib 计算 sign 值：
1
2
3
4
5
6
|
import hashlib
t = int(time.time())
sign = hashlib.sha1(f'Xr0Z-javascript-obfuscation-1{t}'.encode('utf-8')).hexdigest()
url = url + f'&t={t}&sign={sign}'
response = requests.get(url, headers=headers).json()
rows = response.get('items')
|
6.3 CSS 反爬
利用 CSS 的 :before 伪元素和 position:relative/left 偏移来隐藏真实数字顺序，需用正则提取样式还原：
1
2
3
4
5
|
css_name = div.get('class')[0]
value = div.text
relative = re.findall(f'\.{css_name} \{{ position:relative \}}', response_text)
left = re.findall(f'\.{css_name} \{{ left:(.*?)em \}}', response_text)
before = re.findall(f'\.{css_name}:before \{{ content:"(\d+)" \}}', response_text)
|
6.4 字体反爬
服务端使用自定义 WebFont 将字符映射为其他字形。需提取 base64 字体解码，用 TTFont 解析字形顺序并建立映射表替换：
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
|
import base64
from fontTools.ttLib import TTFont
from tempfile import TemporaryFile
font_face = re.findall('base64,(.*?)\)', response.text)
with TemporaryFile() as f:
f.write(base64.b64decode(font_face[0]))
f.seek(0)
font = TTFont(f)
font_map = {str(number_map.get(value)): str(px) for px, value in enumerate(font.getGlyphOrder()[1:])}
table = str.maketrans(font_map)
scores = []
for row in rows:
score = int(row.text.translate(table))
scores.append(score)
# unicode 编码和 16 进制转换
unicode_str = font.encode('unicode-escape').decode()
sixteen_str = unicode_str.replace('\\u', '0x')
|
6.5 雪碧图反爬
将数字拼在一张长图里，通过 CSS 偏移显示。需获取 base64 图片，二值化后按列像素判断，复杂情况可引入机器学习模型：
1
2
3
4
5
6
7
8
9
10
11
12
13
|
# 找到图像的 base64 编码并解码
img_str = re.findall('base64,(.*?)"', text)[0]
img_fp = BytesIO(base64.b64decode(img_str.encode('utf-8')))
img = Image.open(img_fp).convert('1') # 图像二值化处理
# 使用机器学习模型预测图像的数字
from tensorflow import keras
model = keras.models.load_model('model.h5')
def pic2num(img, box):
img = img.crop(box).resize((20, 20)).convert('L')
img_arr = 1 - np.reshape(img, (20, 20, 1)) / 255.0
x = np.array([img_arr])
y = model.predict(x)
return str(np.argmax(y[0]))
|
6.6 滑动解锁验证码
流程：输入账号密码 -> 获取缺口图(灰度/二值化) -> 获取缺口横坐标 -> 生成模拟人类加速/减速的轨迹 -> 移动滑块。
1
2
3
4
5
6
7
8
9
|
# 点击并按住滑块
ActionChains(self.browser).click_and_hold(button).perform()
# 根据生成的位移轨迹来滑动滑块
for i in track:
ActionChains(self.browser).move_by_offset(xoffset=i,yoffset=0).perform()
time.sleep(0.0005)
time.sleep(0.5)
# 释放鼠标
ActionChains(self.browser).release().perform()
|
七、工程化实践：Glided Sky 爬虫靶场
7.1 环境配置
在根目录创建 env.py 存放通用内容（含阿布云代理配置等）：
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
|
cookies = ''
headers = {
'Accept': 'text/html,application/xhtml+xml,application/xml;q=0.9,image/webp,image/apng,*/*;q=0.8,application/signed-exchange;v=b3;q=0.9',
'Accept-Encoding': 'gzip, deflate', 'Accept-Language': 'zh-CN,zh;q=0.9', 'Cache-Control': 'max-age=0',
'Connection': 'keep-alive',
'Cookie': cookies,
'Host': 'www.glidedsky.com',
'Referer': 'http://www.glidedsky.com/level/web/crawler-basic-2?page=1',
'Upgrade-Insecure-Requests': '1',
'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/81.0.4044.92 Safari/537.36'
}
proxyHost = "http-dyn.abuyun.com"
proxyPort = "9020"
proxyUser = "your proxyUser"
proxyPass = "your proxyPass"
proxyMeta = 'http://%(user)s:%(pass)s@%(host)s:%(port)s'.format({
"host": proxyHost, "port": proxyPort, "user": proxyUser, "pass": proxyPass,
})
proxies = {"http": proxyMeta, "https": proxyMeta}
|
7.2 基础 1 与 基础 2 (并发)
基础 1 用 BeautifulSoup 查询求和；基础 2 用线程池提速：
1
2
3
4
5
6
|
import requests
from bs4 import BeautifulSoup
from env import headers
response = requests.get(url, headers=headers)
rows = BeautifulSoup(response.text, 'lxml').find_all('div', class_="col-md-1")
score = sum(int(row.text) for row in rows)
|
1
2
3
4
5
6
7
8
9
|
import requests
from bs4 import BeautifulSoup
from concurrent.futures import ThreadPoolExecutor
from env import headers
urls = [...]
pool = ThreadPoolExecutor(max_workers=20)
score = 0
for result in pool.map(crawler, urls):
score += result
|
7.3 滑块验证码实战 (含进程清理与断点续爬)
使用 OpenCV 模板匹配定位缺口，计算缩放比例，用物理运动公式生成轨迹并驱动 Selenium：
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
|
import urllib3
import numpy as np
import time
import os
import cv2
import psutil
import json
import shutil
from PIL import Image
from bs4 import BeautifulSoup
from selenium import webdriver
from selenium.webdriver import ActionChains
from random import randint
from env import cookies
def kill_process(name):
all_pids = psutil.pids()
pids = []
for pid in all_pids:
p = psutil.Process(pid)
if p.name() == name:
pids.append(pid)
for pid in pids:
p = psutil.Process(pid)
for son in p.children(recursive=True):
son.terminate()
p.terminate()
class Crawler:
def __init__(self):
urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning)
chrome_options = webdriver.ChromeOptions()
chrome_options.add_argument('--test-type --ignore-certificate-errors')
chrome_options.add_experimental_option('useAutomationExtension', False)
chrome_options.add_experimental_option("excludeSwitches", ['enable-automation'])
self.driver = webdriver.Chrome(chrome_options=chrome_options)
self.driver.maximize_window()
self.login(cookies)
self.img_dir = 'image'
os.makedirs(self.img_dir, exist_ok=True)
@staticmethod
def get_postion(img_dir, chunk, canves):
otemp = chunk
oblk = canves
target = cv2.imread(otemp, 0)
template = cv2.imread(oblk, 0)
temp = f'{img_dir}/temp.jpg'
targ = f'{img_dir}/targ.jpg'
cv2.imwrite(temp, template)
cv2.imwrite(targ, target)
target = cv2.imread(targ)
target = cv2.cvtColor(target, cv2.COLOR_BGR2GRAY)
target = abs(255 - target)
cv2.imwrite(targ, target)
target = cv2.imread(targ)
template = cv2.imread(temp)
result = cv2.matchTemplate(target, template, cv2.TM_CCOEFF_NORMED)
x, y = np.unravel_index(result.argmax(), result.shape)
return x, y
@staticmethod
def urllib_download(imgurl, imgsavepath):
from urllib.request import urlretrieve
urlretrieve(imgurl, imgsavepath)
@staticmethod
def get_track(distance):
v = 10
t = 0.2
tracks = []
current = 0
mid = distance * 7 / 8
distance += 10
while current < distance:
if current < mid:
a = randint(6, 9)
else:
a = -randint(7, 10)
v0 = v
s = v0 * t + 0.5 * a * (t ** 2)
current += s
tracks.append(round(s))
v = v0 + a * t
for i in range(4):
tracks.append(-randint(2, 3))
for i in range(4):
tracks.append(-randint(1, 3))
return tracks
def crawler(self, url):
self.driver.get(url)
time.sleep(1)
self.driver.switch_to.frame(self.driver.find_element_by_id('tcaptcha_iframe'))
time.sleep(0.5)
bk_block = self.driver.find_element_by_xpath('//img[@id="slideBg"]')
web_image_width = bk_block.size['width']
bk_block_x = bk_block.location['x']
slide_block = self.driver.find_element_by_xpath('//img[@id="slideBlock"]')
slide_block_x = slide_block.location['x']
bk_block = self.driver.find_element_by_xpath('//img[@id="slideBg"]').get_attribute('src')
slide_block = self.driver.find_element_by_xpath('//img[@id="slideBlock"]').get_attribute('src')
slid_ing = self.driver.find_element_by_xpath('//div[@id="tcaptcha_drag_thumb"]')
self.urllib_download(bk_block, f'{self.img_dir}/bkBlock.png')
self.urllib_download(slide_block, f'{self.img_dir}/slideBlock.png')
time.sleep(0.5)
img_bkblock = Image.open(f'{self.img_dir}/bkBlock.png')
real_width = img_bkblock.size[0]
width_scale = float(real_width) / float(web_image_width)
position = self.get_postion(self.img_dir, f'{self.img_dir}/bkBlock.png', f'{self.img_dir}/slideBlock.png')
real_position = position[1] / width_scale
real_position = real_position - (slide_block_x - bk_block_x)
ActionChains(self.driver).click_and_hold(on_element=slid_ing).perform()
time.sleep(0.5)
if randint(1, 10) < 8:
track_list = self.get_track(real_position + 4)
for track in track_list:
ActionChains(self.driver).move_by_offset(xoffset=track, yoffset=0).perform()
time.sleep(0.002)
else:
ActionChains(self.driver).move_by_offset(xoffset=real_position, yoffset=0).perform()
time.sleep(1)
ActionChains(self.driver).release(on_element=slid_ing).perform()
time.sleep(1)
window = self.driver.current_window_handle
self.driver.switch_to.window(window)
time.sleep(1)
rows = BeautifulSoup(self.driver.page_source, 'lxml').find_all('div', class_="col-md-1")
nums = [int(row.text) for row in rows]
return nums
def login(self, cookies):
loginurl = "http://www.glidedsky.com/login"
target_url = "http://www.glidedsky.com/"
if isinstance(cookies, str):
cookies = dict([i.strip().split('=', 1) for i in cookies.split(';')])
self.driver.get(loginurl)
self.driver.delete_all_cookies()
time.sleep(2)
for name, value in cookies.items():
kv = {'name': name, 'value': value}
self.driver.add_cookie(kv)
time.sleep(2)
self.driver.get(target_url)
def main(self, urls, result_path):
with open(result_path, 'r') as f:
dt = json.load(f)
for url in urls:
for _ in range(20):
try:
scores = self.crawler(url)
except:
continue
if scores:
print(f"第{url.split('=')[-1]}页 合计:{sum(scores)} 明细:{scores}")
dt[url] = sum(scores)
with open(result_path, 'w') as f:
json.dump(dt, f)
break
def __del__(self):
self.driver.quit()
shutil.rmtree(self.img_dir)
if __name__ == '__main__':
kill_process('chromedriver.exe')
result_path = f'crawler-captcha-1.json'
if not os.path.exists(result_path):
with open(result_path, 'w') as f:
json.dump({}, f)
with open(result_path, 'r') as f:
dt = json.load(f)
urls = []
for i in range(1, 1001):
url = f'http://www.glidedsky.com/level/web/crawler-captcha-1?page={i}'
urls.append(url)
for key, value in dt.items():
if value > 0:
urls.remove(key)
if not urls:
print(sum(dt.values()))
else:
print(f"剩余待采集页数:{len(urls)}")
Crawler().main(urls, result_path)
|
八、道德规范与注意事项
- 知识产权包含商标(TM/R)、版权(C)与专利。
- 爬虫应在深夜运行并控制速度，不控制速度会过度消耗服务器资源甚至导致小站宕机，属于不道德或非法行为。
robots.txt 没有标准语法且非强制约束，大型网站用它排除机器人以推广自家的 API。- 常见被封杀原因：JS 未执行致页面空白；Cookie 过期致登录异常；IP 被拉黑致 403（应对：换 IP 或降速）。
- 其他工具参考：DrissionPage (https://www.drissionpage.cn/)。
拓展与后续学习建议
- 协议与工程化
- 深入学习 HTTP/HTTPS 协议与常见响应头（如 Cache-Control、Set-Cookie、RateLimit 等），以便更好地分析接口与调试。
- 掌握 Session、CookieJar、连接池与重试机制，构建更健壮的抓取客户端。
- 反爬与对抗升级
- 系统学习常见反爬策略：频率限制、验证码（滑块/点选/九宫格）、JS 混淆与加密、WebRTC 泄露、指纹识别等。
- 熟悉浏览器指纹（Canvas、WebGL、AudioContext 等）及对抗方案。
- 性能与工程实践
- 并发模型：线程池、异步 IO（asyncio + aiohttp）、协程与消息队列（RabbitMQ/Kafka）的组合使用。
- 分布式爬虫框架：Scrapy 与 Scrapy-Redis，实现任务分发与去重。
- 存储与去重：数据库（MySQL、MongoDB、PostgreSQL）、布隆过滤器、URL 指纹等。
- 法律与伦理
- 学习与爬虫相关的法律法规（如个人信息保护法、网络安全法）以及网站的 Terms of Service。
- 养成“尊重 robots.txt、控制频率、最小必要采集、不对外分发敏感数据”的习惯。
记录生活