影刀RPA技术深度：反爬策略应对实战指南——User-Agent轮换Cookie管理与IP代理配置完全解析

影刀RPA技术深度：反爬策略应对实战指南——User-Agent轮换Cookie管理与IP代理配置完全解析

作者：林焱

反爬策略应对是影刀RPA中网页数据采集的核心技术。掌握反爬策略应对技巧，能稳定高效地采集数据。本文深入讲解User-Agent轮换、Cookie管理、验证码识别和IP代理配置，全部附带真实代码和报错案例。

一、反爬策略概述

网站为了防止自动化采集，会采取各种反爬策略。

1.1 常见反爬策略

1.2 在影刀中应对反爬

影刀提供了多种指令应对反爬：

• HTTP请求指令：可以配置请求头、Cookie、代理
• 网页自动化指令：可以执行JavaScript、处理验证码
• Python脚本指令：可以调用第三方库

二、User-Agent轮换实战

2.1 User-Agent原理

User-Agent是HTTP请求头中的一部分，标识客户端类型。

网站可以检测User-Agent，如果发现是自动化工具，就拒绝请求。

2.2 User-Agent轮换策略

准备多个User-Agent，每次请求随机选择一个。

真实User-Agent列表：

user_agents = [
'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/120.0.0.0 Safari/537.36',
'Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/120.0.0.0 Safari/537.36',
'Mozilla/5.0 (Windows NT 10.0; Win64; x64; rv:121.0) Gecko/20100101 Firefox/121.0',
'Mozilla/5.0 (X11; Linux x86_64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/120.0.0.0 Safari/537.36'
]

2.3 在影刀中实现User-Agent轮换

真实配置方案：

指令：Python脚本
代码：
import random

user_agents = [
'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36',
'Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/537.36',
'Mozilla/5.0 (Windows NT 10.0; Win64; x64; rv:121.0) Gecko/20100101 Firefox/121.0'
]

# 随机选择User-Agent
user_agent = random.choice(user_agents)
yingdao.set_variable('current_user_agent', user_agent)

指令：HTTP请求
方法：GET
URL：https://www.example.com
请求头：
User-Agent: {{current_user_agent}}

2.4 在网页自动化中使用User-Agent

在影刀的"打开网页"指令中，可以配置User-Agent。

真实配置参数：

指令：打开网页
URL：https://www.example.com
高级选项：
自定义请求头：User-Agent: Mozilla/5.0 …

2.5 动态User-Agent生成

某些网站会检测User-Agent的格式，需要动态生成。

真实代码示例：

import random
import time

def generate_user_agent():
"""生成随机的User-Agent"""
chrome_version = random.randint(110, 130)
firefox_version = random.randint(110, 130)

templates = [
f'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/{chrome_version}.0.0.0 Safari/537.36',

f'Mozilla/5.0 (Windows NT 10.0; Win64; x64; rv:{firefox_version}.0) Gecko/20100101 Firefox/{firefox_version}.0'
]

return random.choice(templates)

# 使用
user_agent = generate_user_agent()
print(user_agent)

三、Cookie管理完全攻略

3.1 Cookie原理

Cookie是服务器发送给浏览器的小数据片段，浏览器会在后续请求中带上Cookie。

网站可以用Cookie验证用户身份和会话状态。

3.2 Cookie获取与保存

真实代码示例：

import requests
import json

def get_and_save_cookies(login_url, username, password):
"""登录并保存Cookie"""
session = requests.Session()

# 登录
login_data = {
'username': username,
'password': password
}
response = session.post(login_url, data=login_data)

if response.status_code == 200:
# 保存Cookie
cookies = session.cookies.get_dict()
with open('cookies.json', 'w') as f:
json.dump(cookies, f)

print("Cookie保存成功")
return cookies
else:
raise Exception(f"登录失败：{response.status_code}")

# 使用
cookies = get_and_save_cookies(
'https://www.example.com/login',
'myusername',
'mypassword'
)

3.3 Cookie加载与使用

真实代码示例：

import requests
import json

def load_and_use_cookies(url):
"""加载Cookie并使用"""
# 加载Cookie
with open('cookies.json', 'r') as f:
cookies = json.load(f)

# 使用Cookie发送请求
response = requests.get(url, cookies=cookies)

if response.status_code == 200:
return response.text
else:
raise Exception(f"请求失败：{response.status_code}")

# 使用
html = load_and_use_cookies('https://www.example.com/profile')
print(html)

3.4 在影刀中管理Cookie

拼多多店群自动化报活动上架！

真实配置方案：

指令：HTTP请求
方法：POST
URL：https://www.example.com/login
请求体：{"username": "xxx", "password": "xxx"}
保存响应Cookie到变量：login_cookies

指令：HTTP请求
方法：GET
URL：https://www.example.com/profile
请求Cookie：{{login_cookies}}

3.5 Cookie过期处理

Cookie有过期时间，需要处理过期情况。

真实代码示例：

import requests
import json
import time

def get_cookies_with_retry(login_url, username, password, max_retries=3):
"""获取Cookie，带重试"""
for i in range(max_retries):
try:
cookies = get_and_save_cookies(login_url, username, password)
return cookies
except Exception as e:
print(f"第{i+1}次获取Cookie失败：{e}")
if i == max_retries – 1:
raise
time.sleep(2)

def request_with_cookie_refresh(url, login_url, username, password):
"""使用Cookie请求，如果失败则重新获取Cookie"""
try:
# 加载Cookie
with open('cookies.json', 'r') as f:
cookies = json.load(f)

# 使用Cookie请求
response = requests.get(url, cookies=cookies)

if response.status_code == 200:
return response.text
else:
raise Exception("Cookie可能已过期")
except Exception as e:
print(f"请求失败：{e}，重新获取Cookie")

# 重新获取Cookie
cookies = get_cookies_with_retry(login_url, username, password)

# 使用新Cookie请求
response = requests.get(url, cookies=cookies)
return response.text

四、验证码识别实战

4.1 验证码类型

4.2 使用ddddocr识别验证码

ddddocr是优秀的验证码识别库。

安装：

pip install ddddocr

真实代码示例：

import ddddocr
import requests
import time

def recognize_captcha(captcha_image_url, login_url, username, password):
"""识别验证码并提交"""
# 创建OCR对象
ocr = ddddocr.DdddOcr()

# 下载验证码图像
session = requests.Session()
captcha_response = session.get(captcha_image_url)

# 识别验证码
captcha_text = ocr.classification(captcha_response.content)
print(f"识别到的验证码：{captcha_text}")

# 提交登录
login_data = {
'username': username,
'password': password,
'captcha': captcha_text
}
login_response = session.post(login_url, data=login_data)

if '登录成功' in login_response.text:
print("登录成功")
return session.cookies.get_dict()
else:
print("登录失败")
return None

# 使用
cookies = recognize_captcha(
'https://www.example.com/captcha.jpg',
'https://www.example.com/login',
'myusername',
'mypassword'
)

4.3 滑动验证码处理

滑动验证码需要模拟人类滑动行为。

真实代码示例：

import pyautogui
import time
import random

def simulate_slide_captcha(start_x, start_y, distance):
"""模拟滑动验证码"""
# 移动到起始位置
pyautogui.moveTo(start_x, start_y, duration=0.5)

# 按下鼠标
pyautogui.mouseDown()

# 模拟人类滑动：先快后慢
steps = random.randint(20, 30)
for i in range(steps):
# 计算当前应该滑动的距离
progress = (i + 1) / steps
# 使用缓动函数，模拟先快后慢
eased_progress = progress * progress # 二次缓动
current_distance = int(distance * eased_progress)

# 添加随机抖动
offset_y = random.randint(–3, 3)

# 移动到当前位置
current_x = start_x + current_distance
current_y = start_y + offset_y
pyautogui.moveTo(current_x, current_y, duration=0.05)

# 松开鼠标
pyautogui.mouseUp()

print("滑动完成")

# 使用
# 假设验证码滑块起始位置是(500, 300)，需要滑动300像素
simulate_slide_captcha(500, 300, 300)

4.4 在影刀中处理验证码

真实配置方案：

指令：HTTP请求
方法：GET
URL：https://www.example.com/captcha.jpg
保存响应到文件：C:\\captcha\\captcha.jpg

指令：Python脚本
代码：
import ddddocr

ocr = ddddocr.DdddOcr()

with open('C:\\captcha\\captcha.jpg', 'rb') as f:
captcha_image = f.read()

captcha_text = ocr.classification(captcha_image)
yingdao.set_variable('captcha_text', captcha_text)

指令：填写输入框
元素：验证码输入框
内容：{{captcha_text}}

指令：点击元素
元素：提交按钮

五、IP代理配置

5.1 IP代理原理

IP代理通过代理服务器转发请求，隐藏真实IP。

5.2 代理类型

5.3 在影刀中配置代理

真实配置方案：

指令：HTTP请求
方法：GET
URL：https://www.example.com
代理：
类型：HTTP
地址：127.0.0.1
端口：8080
用户名：proxy_user（可选）
密码：proxy_password（可选）

5.4 在Python中使用代理

真实代码示例：

import requests

def request_with_proxy(url, proxy_address):
"""使用代理发送请求"""
proxies = {
'http': proxy_address,
'https': proxy_address
}

try:
response = requests.get(url, proxies=proxies, timeout=10)

if response.status_code == 200:
return response.text
else:
raise Exception(f"请求失败：{response.status_code}")
except Exception as e:
print(f"使用代理{proxy_address}请求失败：{e}")
return None

# 使用
proxy_address = 'http://127.0.0.1:8080'
html = request_with_proxy('https://www.example.com', proxy_address)

5.5 代理池管理

对于大规模数据采集，需要使用代理池。

真实代码示例：

import requests
import random
import time

class ProxyPool:
"""代理池管理"""

def __init__(self, proxy_list):
self.proxy_list = proxy_list
self.current_index = 0
self.failed_proxies = set()

def get_proxy(self):
"""获取代理"""
# 过滤掉失败的代理
available_proxies = [p for p in self.proxy_list if p not in self.failed_proxies]

if not available_proxies:
raise Exception("没有可用的代理")

# 轮询获取代理
proxy = available_proxies[self.current_index % len(available_proxies)]
self.current_index += 1

return proxy

def mark_proxy_failed(self, proxy):
"""标记代理失败"""
self.failed_proxies.add(proxy)
print(f"代理{proxy}已标记为失败")

# 使用
proxy_list = [
'http://proxy1:8080',
'http://proxy2:8080',
'http://proxy3:8080'
]

proxy_pool = ProxyPool(proxy_list)

for i in range(10):
proxy = proxy_pool.get_proxy()

try:
html = request_with_proxy('https://www.example.com', proxy)
if html:
print(f"使用代理{proxy}请求成功")
break
except Exception as e:
print(f"使用代理{proxy}请求失败：{e}")
proxy_pool.mark_proxy_failed(proxy)

真实报错案例：

错误：Connection refused
原因：代理服务器不可用
解决：更换代理服务器

错误：Proxy authentication required
原因：代理需要认证
解决：在代理地址中添加用户名和密码，如 http://user:pass@proxy:8080

六、请求限速与并发控制

6.1 请求限速原理

请求限速是防止被封IP的重要手段。

6.2 固定间隔限速

真实代码示例：

import time
import requests

def request_with_fixed_interval(urls, interval=1.0):
"""固定间隔请求"""
results = []

for url in urls:
response = requests.get(url)
results.append(response.text)

# 等待固定间隔
time.sleep(interval)

return results

6.3 随机间隔限速

更真实的限速策略是使用随机间隔。

真实代码示例：

import time
import random
import requests

def request_with_random_interval(urls, min_interval=1.0, max_interval=3.0):
"""随机间隔请求"""
results = []

for url in urls:
response = requests.get(url)
results.append(response.text)

# 等待随机间隔
interval = random.uniform(min_interval, max_interval)
time.sleep(interval)

return results

6.4 并发控制

对于大量请求，需要控制并发数。

真实代码示例：

import requests
from concurrent.futures import ThreadPoolExecutor
import threading

def request_with_concurrency(urls, max_workers=5, interval=1.0):
"""并发请求，带限速"""
results = []
last_request_time = 0
lock = threading.Lock()

def request_with_rate_limit(url):
"""带限速的请求"""
nonlocal last_request_time

with lock:
# 限速
current_time = time.time()
time_since_last_request = current_time – last_request_time

if time_since_last_request < interval:
time.sleep(interval – time_since_last_request)

last_request_time = time.time()

# 发送请求
response = requests.get(url)
return response.text

with ThreadPoolExecutor(max_workers=max_workers) as executor:
results = list(executor.map(request_with_rate_limit, urls))

return results

七、真实报错信息与解决方案

7.1 报错：403 Forbidden

可能原因：

解决方案：

# 1. 更换User-Agent
headers = {'User-Agent': 'Mozilla/5.0 …'}

# 2. 更新Cookie
cookies = get_fresh_cookies()

# 3. 使用代理
proxies = {'http': 'http://proxy:8080'}

response = requests.get(url, headers=headers, cookies=cookies, proxies=proxies)

7.2 报错：验证码识别失败

可能原因：

解决方案：

# 1. 预处理验证码图像
import cv2
import numpy as np

image = cv2.imread('captcha.jpg')
gray = cv2.cvtColor(image, cv2.COLOR_BGR2GRAY)
_, binary = cv2.threshold(gray, 128, 255, cv2.THRESH_BINARY)
cv2.imwrite('captcha_processed.jpg', binary)

# 2. 使用更强大的OCR库
# ddddocr通常已经很好，如果还不行，可以尝试商业OCR

# 3. 人工打码
# 对于特别难的验证码，可以使用人工打码平台

7.3 报错：代理连接失败

可能原因：

解决方案：

# 1. 测试代理是否可用
import requests

proxies = {'http': 'http://proxy:8080'}
try:
response = requests.get('http://www.baidu.com', proxies=proxies, timeout=5)
print("代理可用")
except Exception as e:
print(f"代理不可用：{e}")

# 2. 添加代理认证
proxies = {'http': 'http://user:pass@proxy:8080'}

# 3. 确认代理类型
# HTTP代理：http://proxy:8080
# HTTPS代理：https://proxy:8080
# SOCKS5代理：socks5://proxy:1080

八、12大核心模块中的反爬应用

8.1 网页自动化模块

在网页自动化模块中，反爬策略应用于：

• 模拟人类操作速度
• 处理验证码
• 管理Cookie

8.2 Excel自动化模块

Excel自动化模块不直接使用反爬策略，但可以将采集的数据保存到Excel。

8.3 数据库模块

数据库模块不直接使用反爬策略，但可以将采集的数据保存到数据库。

8.4 Python脚本模块

Python脚本模块可以执行任意反爬代码。

TEMU店群矩阵自动化运营核价报活动

8.5 图像识别模块

图像识别模块可以用于识别验证码。

8.6 文件处理模块

文件处理模块可以用于保存Cookie、代理列表等。

8.7 定时任务模块

定时任务模块可以定时执行数据采集任务，自动更换代理。

8.8 API对接模块

API对接模块不直接使用反爬策略，但某些API也有频率限制，需要限速。

8.9 邮件模块

邮件模块可以用于发送采集异常通知。

8.10 钉钉模块

钉钉模块可以用于发送采集异常通知。

8.11 飞书模块

飞书模块可以用于发送采集异常通知。

8.12 企业微信模块

企业微信模块可以用于发送采集异常通知。

九、高级技巧与最佳实践

9.1 使用浏览器指纹伪装

某些网站会检测浏览器指纹，需要伪装。

真实代码示例：

import requests

def get_random_fingerprint():
"""生成随机浏览器指纹"""
fingerprints = [
{
'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36',
'Accept': 'text/html,application/xhtml+xml,application/xml;q=0.9,image/webp,*/*;q=0.8',
'Accept-Language': 'zh-CN,zh;q=0.9,en;q=0.8',
'Accept-Encoding': 'gzip, deflate, br',
'Connection': 'keep-alive',
'Upgrade-Insecure-Requests': '1'
},
{
'User-Agent': 'Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/537.36',
'Accept': 'text/html,application/xhtml+xml,application/xml;q=0.9,*/*;q=0.8',
'Accept-Language': 'zh-CN,zh;q=0.9',
'Accept-Encoding': 'gzip, deflate, br',
'Connection': 'keep-alive'
}
]

return random.choice(fingerprints)

# 使用
headers = get_random_fingerprint()
response = requests.get('https://www.example.com', headers=headers)

9.2 使用无头浏览器

对于JavaScript渲染的页面，需要使用无头浏览器。

真实代码示例：

from selenium import webdriver
from selenium.webdriver.chrome.options import Options

def scrape_with_headless_browser(url):
"""使用无头浏览器采集"""
chrome_options = Options()
chrome_options.add_argument('–headless')
chrome_options.add_argument('–disable-gpu')
chrome_options.add_argument('User-Agent=Mozilla/5.0 …')

driver = webdriver.Chrome(options=chrome_options)
driver.get(url)

# 等待页面加载
time.sleep(3)

# 获取页面源码
html = driver.page_source

driver.quit()
return html

9.3 分布式采集

对于超大规模数据采集，需要分布式采集。

架构设计：

主节点：
– 分配任务
– 管理代理池
– 汇总数据

从节点：
– 执行采集任务
– 使用代理
– 保存数据

十、真实案例：京东商品数据采集

场景：采集京东商品数据，需要应对反爬。

解决方案：

import requests
import random
import time
import json
from bs4 import BeautifulSoup

class JDScraper:
"""京东商品数据采集器"""

def __init__(self):
self.user_agents = [
'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36',
'Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/537.36'
]
self.cookies = None
self.proxy_pool = None

def get_headers(self):
"""获取请求头"""
return {
'User-Agent': random.choice(self.user_agents),
'Referer': 'https://www.jd.com/',
'Accept': 'text/html,application/xhtml+xml,application/xml;q=0.9,*/*;q=0.8'
}

def scrape_product(self, product_id):
"""采集商品数据"""
url = f'https://item.jd.com/{product_id}.html'

# 重试3次
for i in range(3):
try:
# 使用代理
proxy = self.proxy_pool.get_proxy() if self.proxy_pool else None

# 发送请求
response = requests.get(
url,
headers=self.get_headers(),
cookies=self.cookies,
proxies={'http': proxy, 'https': proxy} if proxy else None,
timeout=10
)

if response.status_code == 200:
# 解析商品数据
soup = BeautifulSoup(response.text, 'html.parser')

product_data = {
'id': product_id,
'name': soup.select_one('.sku-name').text.strip(),
'price': soup.select_one('.p-price i').text.strip(),
'shop': soup.select_one('.shop-name').text.strip()
}

return product_data
else:
raise Exception(f"请求失败：{response.status_code}")
except Exception as e:
print(f"第{i+1}次采集失败：{e}")

if i < 2:
time.sleep(random.uniform(2, 5))
else:
raise

return None

def scrape_multiple_products(self, product_ids):
"""批量采集商品"""
results = []

for product_id in product_ids:
try:
product_data = self.scrape_product(product_id)

if product_data:
results.append(product_data)

# 随机等待
time.sleep(random.uniform(1, 3))
except Exception as e:
print(f"采集商品{product_id}失败：{e}")

return results

# 使用
scraper = JDScraper()
product_ids = ['123456', '234567', '345678']
products = scraper.scrape_multiple_products(product_ids)
print(products)

十一、总结

反爬策略应对是影刀RPA中网页数据采集的核心技术。掌握反爬策略应对技巧，能稳定高效地采集数据。

关键要点：

下一篇我们将讲解性能优化技巧。

内容标签：#影刀RPA #RPA教程 #反爬策略 #数据采集 #User-Agent

作者：林焱