Python脚本爬虫User-Agent如何轮换

wen 实用脚本 2

本文目录导读:

Python脚本爬虫User-Agent如何轮换

  1. 使用fake-useragent库(推荐)
  2. 预定义User-Agent列表轮换
  3. 完整的爬虫类实现
  4. 异步爬虫中的UA轮换(使用aiohttp)
  5. 使用第三方库scrapy轮换UA
  6. 注意事项

在Python爬虫中轮换User-Agent可以有效降低被网站识别为爬虫的风险,以下是几种常用的User-Agent轮换方法:

使用fake-useragent库(推荐)

from fake_useragent import UserAgent
import requests
# 创建UserAgent对象
ua = UserAgent()
# 每次请求随机获取一个User-Agent
headers = {'User-Agent': ua.random}
response = requests.get('https://example.com', headers=headers)

安装方法:

pip install fake-useragent

高级用法:

from fake_useragent import UserAgent
import time
import random
class UserAgentRotator:
    def __init__(self, fallback='Mozilla/5.0'):
        self.ua = UserAgent(fallback=fallback)
    def get_random_ua(self):
        """获取随机User-Agent"""
        return self.ua.random
    def get_specific_browser_ua(self, browser='chrome'):
        """获取特定浏览器类型的UA"""
        if browser.lower() == 'chrome':
            return self.ua.chrome
        elif browser.lower() == 'firefox':
            return self.ua.firefox
        elif browser.lower() == 'safari':
            return self.ua.safari
        elif browser.lower() == 'edge':
            return self.ua.edge
        else:
            return self.ua.random
# 使用示例
rotator = UserAgentRotator()
# 随机UA
headers = {'User-Agent': rotator.get_random_ua()}
# 特定浏览器UA
headers = {'User-Agent': rotator.get_specific_browser_ua('chrome')}

预定义User-Agent列表轮换

import random
import requests
class UserAgentPool:
    def __init__(self):
        self.user_agents = [
            # Chrome
            '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 (X11; Linux x86_64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/120.0.0.0 Safari/537.36',
            # Firefox
            'Mozilla/5.0 (Windows NT 10.0; Win64; x64; rv:121.0) Gecko/20100101 Firefox/121.0',
            'Mozilla/5.0 (Macintosh; Intel Mac OS X 10.15; rv:121.0) Gecko/20100101 Firefox/121.0',
            # Safari
            'Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/605.1.15 (KHTML, like Gecko) Version/17.1 Safari/605.1.15',
            # Edge
            'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/120.0.0.0 Safari/537.36 Edg/120.0.0.0',
            # Opera
            'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/120.0.0.0 Safari/537.36 OPR/106.0.0.0',
        ]
        self.current_index = 0
    def get_random(self):
        """随机获取一个UA"""
        return random.choice(self.user_agents)
    def get_sequential(self):
        """顺序轮换获取UA"""
        ua = self.user_agents[self.current_index]
        self.current_index = (self.current_index + 1) % len(self.user_agents)
        return ua
    def add_ua(self, user_agent):
        """添加自定义UA"""
        self.user_agents.append(user_agent)
# 使用示例
ua_pool = UserAgentPool()
# 随机获取
headers = {'User-Agent': ua_pool.get_random()}
# 顺序获取
headers = {'User-Agent': ua_pool.get_sequential()}

完整的爬虫类实现

import requests
import random
import time
from fake_useragent import UserAgent
from requests.adapters import HTTPAdapter
from urllib3.util.retry import Retry
class SmartCrawler:
    def __init__(self, proxies=None, retry_times=3, timeout=10):
        # User-Agent管理器
        self.ua_manager = UserAgent()
        # 代理列表
        self.proxies = proxies or []
        # 创建会话
        self.session = self._create_session(retry_times, timeout)
        # 请求计数器
        self.request_count = 0
    def _create_session(self, retry_times, timeout):
        """创建带重试机制的会话"""
        session = requests.Session()
        # 配置重试策略
        retry_strategy = Retry(
            total=retry_times,
            backoff_factor=1,
            status_forcelist=[429, 500, 502, 503, 504],
        )
        adapter = HTTPAdapter(max_retries=retry_strategy)
        session.mount('http://', adapter)
        session.mount('https://', adapter)
        return session
    def _get_headers(self, referer=None):
        """生成请求头"""
        headers = {
            'User-Agent': self.ua_manager.random,
            'Accept': 'text/html,application/xhtml+xml,application/xml;q=0.9,*/*;q=0.8',
            'Accept-Language': 'en-US,en;q=0.5',
            'Accept-Encoding': 'gzip, deflate',
            'Connection': 'keep-alive',
            'Upgrade-Insecure-Requests': '1',
        }
        if referer:
            headers['Referer'] = referer
        return headers
    def _get_proxy(self):
        """获取随机代理"""
        if self.proxies:
            return random.choice(self.proxies)
        return None
    def _delay(self, min_delay=1, max_delay=3):
        """随机延迟"""
        delay = random.uniform(min_delay, max_delay)
        time.sleep(delay)
    def request(self, url, method='GET', params=None, data=None, 
                referer=None, min_delay=1, max_delay=3):
        """发送请求"""
        # 更新请求计数器
        self.request_count += 1
        # 定期更换会话(每100次请求)
        if self.request_count % 100 == 0:
            self.session = self._create_session(3, 10)
        # 获取请求头和代理
        headers = self._get_headers(referer)
        proxy = self._get_proxy()
        # 构建请求参数
        request_kwargs = {
            'headers': headers,
            'timeout': 10,
        }
        if proxy:
            request_kwargs['proxies'] = {'http': proxy, 'https': proxy}
        if method.upper() == 'GET':
            request_kwargs['params'] = params
            response = self.session.get(url, **request_kwargs)
        elif method.upper() == 'POST':
            request_kwargs['data'] = data
            response = self.session.post(url, **request_kwargs)
        # 随机延迟
        self._delay(min_delay, max_delay)
        return response
# 使用示例
crawler = SmartCrawler()
# 单个请求
response = crawler.request('https://example.com')
print(response.headers.get('User-Agent'))
# 批量请求示例
urls = ['https://example1.com', 'https://example2.com', 'https://example3.com']
for url in urls:
    try:
        response = crawler.request(url)
        print(f"请求 {url} 成功,状态码: {response.status_code}")
    except Exception as e:
        print(f"请求 {url} 失败: {e}")

异步爬虫中的UA轮换(使用aiohttp)

import aiohttp
import asyncio
import random
from fake_useragent import UserAgent
class AsyncCrawler:
    def __init__(self, max_concurrent=10):
        self.ua = UserAgent()
        self.semaphore = asyncio.Semaphore(max_concurrent)
        self.cookies = {}
    async def fetch(self, url, session):
        async with self.semaphore:
            headers = {'User-Agent': self.ua.random}
            try:
                async with session.get(url, headers=headers, 
                                     cookies=self.cookies, timeout=10) as response:
                    return await response.text()
            except Exception as e:
                print(f"请求失败: {e}")
                return None
    async def run(self, urls):
        async with aiohttp.ClientSession() as session:
            tasks = [self.fetch(url, session) for url in urls]
            results = await asyncio.gather(*tasks)
            return results
# 使用示例
async def main():
    crawler = AsyncCrawler(max_concurrent=5)
    urls = ['https://example.com' for _ in range(10)]
    results = await crawler.run(urls)
    print(f"成功获取 {len([r for r in results if r])} 个页面")
# asyncio.run(main())

使用第三方库scrapy轮换UA

# settings.py
DOWNLOADER_MIDDLEWARES = {
    'scrapy.downloadermiddlewares.useragent.UserAgentMiddleware': None,
    'myproject.middlewares.RotateUserAgentMiddleware': 400,
}
# middlewares.py
from fake_useragent import UserAgent
class RotateUserAgentMiddleware:
    def __init__(self):
        self.ua = UserAgent()
    @classmethod
    def from_crawler(cls, crawler):
        return cls()
    def process_request(self, request, spider):
        request.headers['User-Agent'] = self.ua.random

注意事项

  1. 不要频繁更换UA:每次请求都更换UA反而容易引起注意,建议每3-5次请求更换一次

  2. 配合其他反爬措施

    • 使用代理IP轮换
    • 添加随机延迟
    • 模拟浏览器行为(cookies、headers等)
  3. 定期更新UA列表:浏览器版本更新后,旧UA会被识别为异常

  4. 测试UA有效性

    def test_ua(user_agent):
     """测试UA是否被目标网站接受"""
     headers = {'User-Agent': user_agent}
     url = 'https://httpbin.org/user-agent'
     response = requests.get(url, headers=headers)
     return response.json()['user-agent'] == user_agent

选择哪种方法取决于你的具体需求:

  • 简单场景:使用fake-useragent库
  • 生产环境:预定义+动态更新的UA池
  • 高并发场景:使用异步+UA池
  • 框架集成:使用Scrapy等框架的中间件

抱歉,评论功能暂时关闭!