爬蟲05 - 爬蟲攻防

一：隨機User-Agent爬蟲

1：fake-useragent

當爬蟲請求頭（User-Agent）暴露規律時，目標網站的反爬系統會在?5秒內?識別并封鎖IP。2023年AlexTop百萬網站統計顯示，?68.7%的反爬策略會檢測User-Agent特征?。

檢測項	檢測原理	典型案例
固定特征值	持續相同User-Agent觸發閾值告警	某電商平臺連續10次相同UA即封禁
非常用瀏覽器	識別非常規瀏覽器版本（如過時Chrome 85）	政府網站拒絕服務古董瀏覽器
設備類型沖突	移動端UA訪問PC端網頁觸發異常	新聞APP接口校驗設備一致性
協議完整性	缺失Accept-Encoding/Connection等標準頭	金融數據接口強制校驗完整協議頭

pip install fake-useragent --upgrade # 添加upgrade是為了防止舊版數據源失效的問題

from fake_useragent import UserAgent
import requests# 創建UserAgent對象, 下面將使用ua.random 獲取隨機的 UserAgent
ua = UserAgent(browsers=['chrome', 'firefox', 'edge'], os=['windows', 'mac'])
header = {'User-Agent': ua.random, # 隨機獲取一個UserAgent'Accept-Encoding': 'gzip, deflate, br', # 告訴服務器，我們接受gzip壓縮'Accept-Language': 'zh-CN,zh;q=0.9,en;q=0.8', # 告訴服務器，我們接受中文'Connection': 'keep-alive' # 告訴服務器，我們保持連接
}requests.get('https://www.baidu.com', headers=header)

可以封裝設備的一致性

from fake_useragent import UserAgent
import requests# 可以封裝設備一致性
def generate_user_agent(device_type="pc"):ua = UserAgent()base_headers = {'Accept-Encoding': 'gzip, deflate, br','Accept-Language': 'zh-CN,zh;q=0.9,en;q=0.8'}if device_type == 'mobile':return {**base_headers,'User-Agent': ua.firefox, # a'X-Requested-With': 'com.android.browser'}else:return {**base_headers,'User-Agent': ua.chrome, # b'Sec-CH-UA-Platform': '"Windows"'}for page in range(1, 11):headers = generate_user_agent('mobile' if page % 2 else 'pc')response = requests.get(f'https://www.zhipin.com/job_detail/?query=python&city=101010100&industry=&position=', headers=headers)while True:if response.status_code == 200:print(response.text)breakprint("=" * 20)

2：高級反反爬策略

方案一：動態版本更新?（解決版本過時檢測）

# 強制使用最新Chrome版本  
ua = UserAgent(min_version=120)  # Chrome 120+  
headers = {'User-Agent': ua.chrome}  

方案二：混合真實瀏覽器指紋?

# 從真實瀏覽器捕獲指紋注入  
real_fingerprint = {  'Sec-CH-UA': '"Chromium";v="118", "Google Chrome";v="118", "Not=A?Brand";v="8"',  'Sec-CH-UA-Mobile': '?0',  'Sec-CH-UA-Platform': '"Windows"'  
}  
headers = {‌**generate_context_headers(), **‌real_fingerprint}  

失敗重試熔斷機制

from tenacity import retry, stop_after_attempt, wait_exponential  @retry(stop=stop_after_attempt(3), wait=wait_exponential(multiplier=1))  
def safe_request(url):  try:  return requests.get(url, headers=generate_context_headers())  except requests.exceptions.RequestException as e:  if e.response.status_code == 403:  # 觸發UA刷新熔斷  UserAgent().update()  raise  safe_request('https://target.com/api')  

3：生產環境建議

定時更新UA數據庫

# 每天自動更新UA數據庫  
0 3 * * * /usr/bin/python3 -c "from fake_useragent import UserAgent; UserAgent().update()"  

可以配置些監控和報警

# 當連續5次403錯誤時觸發警報  
if error_count > 5:  send_alert(f"UA策略失效！當前攔截率：{error_count/request_count*100:.2f}%")  switch_to_backup_proxy()  

在生產環境中最好使用多庫備用

# 當fake_useragent失效時切換至browser_useragent  
try:  from fake_useragent import UserAgent  
except ImportError:  from browswer_useragent import BrowserUserAgent as UserAgent

二：代理IP爬蟲

當爬蟲請求頻率超過?5次/秒?時，目標網站的反爬系統將在?10秒內?封鎖當前IP。據2024年全球反爬技術報告，?83%的網站采用IP指紋檢測?作為核心防御手段

動態代理IP池?，結合智能路由與熔斷機制實現反爬突圍，實測將IP封禁率從?72%降至3%?

檢測維度	反爬策略	典型案例
請求頻率閾值	單IP單位時間內請求次數超限觸發封禁	某社交平臺限制單IP每秒3次請求
IP黑名單庫	識別代理服務器IP段并全局封禁	新聞網站屏蔽已知數據中心IP
地理位置異常	短時間跨國IP跳躍觸發風控	電商平臺攔截中美IP交替訪問行為

1：獲取代理IP

可以使用https://free-proxy-list.net/zh-cn/獲取到免費的代理IP

import requests
import random
from bs4 import BeautifulSoup# 獲取免費代理 -> https://free-proxy-list.net/zh-cn/
def scrape_free_proxies():url = "https://free-proxy-list.net/" # 獲取免費的ip代理response = requests.get(url)soup = BeautifulSoup(response.text, 'html.parser')proxies = [] # 創建一個空列表, 存儲獲取到的代理for row in soup.select("table.table tbody tr"):cols = row.find_all('td')if len(cols) < 7:continueif cols[4].text == 'elite proxy' and cols[6].text == 'yes':proxies.append(f"{cols[0].text}:{cols[1].text}")return proxies# 請求, 如果失敗了就換一個代理IP, 最多嘗試5次
def do_request(proxy):for i in range(5):print(f"Trying proxy: {proxy}")try:response = requests.get('http://www.baidu.com', proxies=proxy)print(response.json())returnexcept:print(f"Failed to get IP, trying again...")proxy = {'http': f'http://{random.choice(proxy_list)}','https': f'http://{random.choice(proxy_list)}'}if __name__ == '__main__':proxy_list = scrape_free_proxies()print(proxy_list)# 隨機選擇代理current_proxy = {'http': f'http://{random.choice(proxy_list)}','https': f'http://{random.choice(proxy_list)}'}do_request(current_proxy)

還可以添加代理IP的智能容錯機制

import requests
import random
from bs4 import BeautifulSoup
from tenacity import retry, stop_after_attempt, wait_fixed# 獲取免費代理 -> https://free-proxy-list.net/zh-cn/
def scrape_free_proxies():url = "https://free-proxy-list.net/" # 獲取免費的ip代理response = requests.get(url)soup = BeautifulSoup(response.text, 'html.parser')proxies = [] # 創建一個空列表, 存儲獲取到的代理for row in soup.select("table.table tbody tr"):cols = row.find_all('td')if len(cols) < 7:continueif cols[4].text == 'elite proxy' and cols[6].text == 'yes':proxies.append(f"{cols[0].text}:{cols[1].text}")return proxies# 代理IP的智能容錯機制
@retry(stop=stop_after_attempt(3), wait=wait_fixed(2))
def robust_request(url, proxy_pool):proxy = random.choice(proxy_pool)try:return requests.get(url,proxies={'http': f'http://{proxy}', 'https': f'http://{proxy}'},timeout=10)except (requests.ProxyError, requests.ConnectTimeout):proxy_pool.remove(proxy)  # 移除失效代理raiseif __name__ == '__main__':proxy_list = scrape_free_proxies()url = "http://www.baidu.com"response = robust_request(url, proxy_list)print(response.status_code)print(response.text)

2：高階攻防

四類代理IP的選型

代理類型	優勢	劣勢	適用場景
數據中心代理	高速度、低延遲	易被識別封禁	快速抓取非敏感數據
住宅代理	高匿名性、真實用戶IP	成本高、速度波動	對抗嚴格反爬（如Cloudflare）
移動4G代理	極高匿名性、IP池龐大	穩定性差、管理復雜	抓取APP接口數據
Socks5代理	支持UDP、加密傳輸	配置復雜、兼容性要求	需要深度匿名場景

由此演化出對應IP黑名單的三種策略：

策略一：協議混淆，將HTTP流量偽裝成Socks5

import socks  
import socket  # 強制使用Socks5協議  
socks.set_default_proxy(socks.SOCKS5, "proxy_ip", 1080)  
socket.socket = socks.socksocket  # 發送請求（網站識別為普通Socks流量）  
requests.get("https://target.com")  

策略二：IP冷啟動?：新代理首次訪問僅采集低風險頁面

策略三：流量染色?：在代理請求中注入真實瀏覽器指紋（如TLS指紋）

3：企業級的代理實戰

redis自建代理池系統

import redis
import json
import requestsclass ProxyPool:def __init__(self):self.redis = redis.Redis(host='127.0.0.1', port=6379, db=0)def add_proxy(self, proxy:str, score:float=100):self.redis.zadd('proxies', {proxy: score})def get_best_proxy(self):return self.redis.zrange('proxies', 0, 0)[0].decode()def refresh_proxy(self, proxy:str, penalty:float=20):self.redis.zincrby('proxies', -penalty, proxy)if __name__ == '__main__':# 添加代理pool = ProxyPool()pool.add_proxy('127.0.0.1:8080')pool.add_proxy('127.0.0.1:8081')pool.add_proxy('127.0.0.1:8082')# 獲取代理best_proxy = pool.get_best_proxy()try:# 請求, 如果失敗了就換一個代理IP, 最多嘗試5次requests.get("https://target.com", proxies={'http': best_proxy})except Exception:pool.refresh_proxy(best_proxy)  

商業代理集成

import hashlib  
import time  def gen_mogu_proxy():  # 生成動態簽名  timestamp = str(int(time.time()))  secret = "your_api_secret"  sign = hashlib.md5(f"timestamp={timestamp}&secret={secret}".encode()).hexdigest()  # 獲取獨享代理（按需切換IP）  api_url = f"http://piping.mogumiao.com/proxy/api/get_ip?count=1&timestamp={timestamp}&sign={sign}"  result = requests.get(api_url).json()  return f"{result['msg'][0]['ip']}:{result['msg'][0]['port']}"  # 獲取高匿名IP  
mogu_proxy = gen_mogu_proxy()  
requests.get("https://target.com", proxies={'http': f'http://{mogu_proxy}'})  

三：動態數據的抓取

當傳統爬蟲遭遇?React/Vue單頁應用?時，?83%的數據請求?通過Ajax/WebSocket動態加載，直接獲取HTML源碼的成功率不足15%。

而如果結合?逆向工程?與?無頭瀏覽器控制技術?，構建覆蓋SPA（單頁應用）、SSR（服務端渲染）、CSR（客戶端渲染）的全場景解決方案，實現動態數據抓取成功率從?12%到98%?的技術躍遷

1：動態頁面技術全景

技術類型	核心原理	典型場景
Ajax/XHR	XMLHttpRequest異步獲取數據	電商商品分頁加載
WebSocket	全雙工通信實時更新	股票行情/在線聊天
SSR	服務端生成動態HTML（如Next.js）	新聞門戶首屏渲染
CSR	客戶端JS動態構建DOM（如React/Vue）	后臺管理系統
JSONP	跨域數據獲取（逐漸被CORS替代）	老舊天氣預報接口

2：動態頁面逆向工程

2.1：XHR請求追蹤與解析

1. 打開?Network面板?并篩選XHR/Fetch請求
2. 定位目標數據的API端點（如/graphql）
3. 解析請求頭認證參數（Authorization/X-API-Key）
4. 復制為Python代碼（Copy as cURL → 轉換為requests代碼）

import requests
from urllib.parse import urlencode# 設置請求頭，包含API版本和授權信息
headers = {'x-api-version': '3.2','authorization': 'Bearer eyJhbGciOiJIUzI1Ni...',
}# 定義請求參數，包括類別ID、排序方式、頁碼和平臺信息
params = {'categoryId': 305,'sort': 'sales_desc','page': 1,'platform': 'web'
}# 直接請求數據接口
response = requests.get('https://api.shop.com/graphql',headers=headers,params=urlencode(params, doseq=True)
)# 解析JSON數據
products = response.json()['data']['products']

2.2：websocket實時數據捕獲

import asyncio
import websockets
import jsonasync def fetch_danmu():uri = "wss://live-api.example.com/ws"  # 替換為實際的 WebSocket 地址while True:try:async with websockets.connect(uri) as websocket:print("成功連接到直播間！")# 可選：發送認證信息auth_message = json.dumps({"user": "test_user","token": "your_token"})await websocket.send(auth_message)print("認證信息已發送")while True:try:# 接收服務器發送的彈幕消息message = await websocket.recv()danmu_data = json.loads(message)print(f"收到彈幕: {danmu_data.get('content', '未知內容')}")except websockets.exceptions.ConnectionClosed:print("連接斷開，重新連接中...")breakexcept Exception as e:print(f"發生錯誤: {e}")break# 運行異步任務
asyncio.get_event_loop().run_until_complete(fetch_danmu())

3：無頭瀏覽器控制技術

無頭瀏覽器(Headless Browser)是指沒有圖形用戶界面的瀏覽器，可以通過編程方式控制，模擬用戶操作，執行JavaScript渲染，是現代爬蟲技術中的重要工具。

1. Puppeteer - Google開發的Node庫，控制Chromium/Chrome
2. Playwright - Microsoft開發的多瀏覽器控制工具
3. Selenium - 傳統的瀏覽器自動化框架(后面介紹)
4. Pyppeteer - Puppeteer的Python版本

3.1：Playwright詳解

Playwright是由Microsoft開發的跨瀏覽器自動化測試工具，支持Chromium、WebKit和Firefox

Playwright有如下的特性：

• 多瀏覽器支持：Chromium (Chrome, Edge)、WebKit (Safari)、Firefox
• 跨平臺能力：Windows、macOS、Linux全平臺支持 & 可本地運行也可CI/CD集成
• 多語言綁定：JavaScript/TypeScript、Python、Java、.NET
• 現代化架構：基于WebSocket的通信協議、自動等待機制、強大的選擇器引擎

pip install playwright
playwright install  # 安裝瀏覽器

基本頁面操作

from playwright.sync_api import sync_playwrightwith sync_playwright() as p:# 啟動瀏覽器(無頭模式)browser = p.chromium.launch(headless=False)# 創建新頁面page = browser.new_page()# 導航到URLpage.goto("https://example.com")# 獲取頁面標題print(page.title())# 截圖page.screenshot(path="example.png")# 關閉瀏覽器browser.close()

元素定位與交互 - Playwright提供多種強大的選擇器：

# CSS選擇器
page.click("button.submit")# 文本選擇器
page.click("text=Login")# XPath
page.click("//button[@id='submit']")# 組合選擇器
page.click("article:has-text('Playwright') >> button")

自動等待機制 -> Playwright內置智能等待，無需手動添加sleep

# 等待元素出現(最多10秒)
page.wait_for_selector("#dynamic-element", timeout=10000)# 等待導航完成
page.click("text=Navigate")
page.wait_for_url("**/new-page")# 等待網絡請求完成
with page.expect_response("**/api/data") as response_info:page.click("button.load-data")
response = response_info.value

網絡請求攔截

# 路由攔截
def handle_route(route):if "ads" in route.request.url:route.abort()  # 阻止廣告請求else:route.continue_()page.route("**/*", handle_route)

文件下載處理

# 等待下載開始
with page.expect_download() as download_info:page.click("a#download-link")
download = download_info.value# 保存下載文件
path = download.path()
download.save_as("/path/to/save")

iframe處理

# 定位iframe
frame = page.frame(name="embedded")# 在iframe內操作
frame.fill("#username", "testuser")
frame.click("#submit")

爬蟲實戰應用：

動態內容抓取

async with async_playwright() as p:browser = await p.chromium.launch()page = await browser.new_page()await page.goto("https://dynamic-ecom-site.com")# 滾動加載所有商品while await page.locator("text=Load More").is_visible():await page.click("text=Load More")await page.wait_for_timeout(2000)  # 適當延遲# 提取所有商品數據products = await page.locator(".product").evaluate_all("""products => products.map(p => ({name: p.querySelector('.name').innerText,price: p.querySelector('.price').innerText}))""")print(products)await browser.close()

登錄會話保持

# 保存登錄狀態
context = browser.new_context()
page = context.new_page()
page.goto("login_url")
page.fill("#username", "user")
page.fill("#password", "pass")
page.click("#login")# 保存cookies
context.storage_state(path="auth.json")# 后續使用保存的狀態
context = browser.new_context(storage_state="auth.json")
page = context.new_page()

性能優化技巧

瀏覽器上下文復用：

context = browser.new_context()
page1 = context.new_page()
page2 = context.new_page()

請求過濾：

await page.route("**/*.{png,jpg,jpeg}", lambda route: route.abort())

并行處理：

async with asyncio.TaskGroup() as tg:tg.create_task(scrape_page(page1, url1))tg.create_task(scrape_page(page2, url2))

常見問題解決方案

檢測規避：

# 修改WebGL供應商信息
await page.add_init_script("""const originalGetParameter = WebGLRenderingContext.prototype.getParameter;WebGLRenderingContext.prototype.getParameter = function(parameter) {if (parameter === 37445) return "Intel Open Source Technology Center";return originalGetParameter.call(this, parameter);};
""")

超時處理：

try:await page.wait_for_selector(".element", timeout=5000)
except TimeoutError:print("元素加載超時")

元素點擊問題：

await page.locator("button").dispatch_event("click")  # 直接觸發事件

與Puppeteer對比

特性	Playwright	Puppeteer
瀏覽器支持	多引擎	僅Chromium
語言支持	多種	主要JS
自動等待	更智能	基礎
選擇器引擎	更強大	標準
移動設備模擬	完善	有限
社區生態	快速增長	成熟穩定

3.2：反反爬蟲策略應對

• 指紋偽裝：修改瀏覽器指紋特征
• 行為模擬：模擬人類操作模式(鼠標移動、隨機延遲)
• 代理輪換：結合代理IP池使用
• WebGL/Canvas指紋處理：定制化渲染參數

3.3：高級技術應用

分布式無頭瀏覽器集群

智能渲染策略

• 按需渲染：根據目標網站特點定制渲染策略
• 資源加載控制：選擇性加載CSS/JS/圖片
• 預渲染緩存：對常見頁面進行預渲染

性能優化技術

1. 瀏覽器實例復用：避免頻繁啟動關閉
2. 頁面池管理：維護多個頁面實例
3. 資源攔截：阻止不必要資源加載
4. CDN緩存利用：合理設置緩存策略

挑戰	解決方案
資源消耗大	瀏覽器實例池化、資源限制
檢測風險高	指紋偽裝、行為模擬
穩定性問題	心跳檢測、自動恢復
性能瓶頸	分布式架構、智能調度