Python防爬虫案例如何封装爬虫防护

本文目录导读：

1. 基础IP限制防护
2. 高级指纹识别防护
3. 动态验证码防护
4. 综合防护中间件
5. 使用建议

我来为你介绍几个Python防爬虫的封装实现方案,从简单到复杂，包含完整的代码示例。

基础IP限制防护

from flask import Flask, request, jsonify
from functools import wraps
import time
from collections import defaultdict
import hashlib
class IPRateLimiter:
    def __init__(self, max_requests=10, time_window=60):
        self.max_requests = max_requests
        self.time_window = time_window
        self.requests = defaultdict(list)
    def is_allowed(self, ip):
        now = time.time()
        # 清理过期记录
        self.requests[ip] = [t for t in self.requests[ip] if now - t < self.time_window]
        if len(self.requests[ip]) >= self.max_requests:
            return False
        self.requests[ip].append(now)
        return True
class AntiCrawler:
    def __init__(self):
        self.rate_limiter = IPRateLimiter()
        self.blocked_ips = set()
    def check_request(self, request):
        """综合检测请求是否来自爬虫"""
        ip = request.remote_addr
        # 1. 检查IP是否被封锁
        if ip in self.blocked_ips:
            return False
        # 2. 检查频率限制
        if not self.rate_limiter.is_allowed(ip):
            self.blocked_ips.add(ip)
            return False
        # 3. 检查User-Agent
        user_agent = request.headers.get('User-Agent', '')
        if self._is_suspicious_ua(user_agent):
            self.blocked_ips.add(ip)
            return False
        # 4. 检查请求间隔
        if not self._check_request_interval(request):
            return False
        return True
    def _is_suspicious_ua(self, user_agent):
        """检测可疑的User-Agent"""
        suspicious_patterns = [
            'python-requests',
            'scrapy',
            'curl',
            'wget',
            'bot',
            'spider',
            'crawler'
        ]
        return any(pattern.lower() in user_agent.lower() 
                  for pattern in suspicious_patterns)
    def _check_request_interval(self, request):
        """检查请求间隔时间"""
        # 实现具体的请求间隔检测逻辑
        return True
# 使用装饰器保护API
def anti_crawler_protect(anti_crawler_instance):
    def decorator(f):
        @wraps(f)
        def decorated_function(*args, **kwargs):
            if not anti_crawler_instance.check_request(request):
                return jsonify({'error': 'Too many requests'}), 429
            return f(*args, **kwargs)
        return decorated_function
    return decorator
# Flask应用示例
app = Flask(__name__)
anti_crawler = AntiCrawler()
@app.route('/api/data')
@anti_crawler_protect(anti_crawler)
def get_data():
    return jsonify({'data': 'sensitive data'})

高级指纹识别防护

import jwt
from datetime import datetime, timedelta
import base64
import json
from Crypto.Cipher import AES
import secrets
class AdvancedAntiCrawler:
    def __init__(self, secret_key):
        self.secret_key = secret_key
        self.client_fingerprints = {}
    def generate_client_token(self, request):
        """生成客户端唯一标识令牌"""
        fingerprint = self._create_fingerprint(request)
        payload = {
            'fingerprint': fingerprint,
            'exp': datetime.utcnow() + timedelta(hours=24),
            'iat': datetime.utcnow(),
            'jti': secrets.token_hex(16)
        }
        token = jwt.encode(payload, self.secret_key, algorithm='HS256')
        return token
    def _create_fingerprint(self, request):
        """创建浏览器指纹"""
        fingerprint_data = {
            'ip': request.remote_addr,
            'user_agent': request.headers.get('User-Agent', ''),
            'accept_language': request.headers.get('Accept-Language', ''),
            'accept_encoding': request.headers.get('Accept-Encoding', ''),
            'cookies': str(request.cookies),
            'screen_size': request.headers.get('X-Screen-Size', ''),
            'timezone': request.headers.get('X-Timezone', ''),
            'canvas_fp': request.headers.get('X-Canvas-Fingerprint', '')
        }
        return hashlib.sha256(str(fingerprint_data).encode()).hexdigest()
    def validate_puzzle(self, puzzle_data):
        """验证拼图验证结果"""
        # 实现拼图验证逻辑
        pass
class BehaviorAnalyzer:
    """行为分析引擎"""
    def __init__(self):
        self.behavior_patterns = defaultdict(list)
    def record_behavior(self, session_id, action):
        """记录用户行为"""
        self.behavior_patterns[session_id].append({
            'action': action,
            'timestamp': time.time(),
            'mouse_movement': action.get('mouse_movement', []),
            'scroll_depth': action.get('scroll_depth', 0),
            'click_pattern': action.get('click_pattern', [])
        })
    def analyze(self, session_id):
        """分析行为模式是否异常"""
        if session_id not in self.behavior_patterns:
            return False
        actions = self.behavior_patterns[session_id]
        # 检测异常行为模式
        anomalies = []
        # 检查访问速度是否异常
        if len(actions) >= 2:
            time_diffs = [actions[i+1]['timestamp'] - actions[i]['timestamp'] 
                         for i in range(len(actions)-1)]
            avg_interval = sum(time_diffs) / len(time_diffs)
            if avg_interval < 0.1:  # 小于100ms为异常
                anomalies.append('too_fast')
        # 检查点击模式
        click_patterns = [a['click_pattern'] for a in actions if a['click_pattern']]
        if click_patterns and self._is_human_like(click_patterns):
            anomalies.append('non_human_click')
        return len(anomalies) > 0
    def _is_human_like(self, patterns):
        """判断点击模式是否像人类"""
        # 实现更复杂的行为分析
        return False

动态验证码防护

import random
from PIL import Image, ImageDraw, ImageFont
import io
import base64
class DynamicCaptcha:
    def __init__(self):
        self.captcha_store = {}
    def generate(self, session_id, length=6):
        """生成动态验证码"""
        # 生成随机字符
        chars = 'ABCDEFGHJKLMNPQRSTUVWXYZ23456789'
        code = ''.join(random.choice(chars) for _ in range(length))
        # 创建验证码图片
        width = 200
        height = 80
        image = Image.new('RGB', (width, height), (255, 255, 255))
        draw = ImageDraw.Draw(image)
        # 添加干扰元素
        for _ in range(10):
            x = random.randint(0, width)
            y = random.randint(0, height)
            draw.point((x, y), fill=(random.randint(0, 255), 
                                     random.randint(0, 255), 
                                     random.randint(0, 255)))
        # 添加文字
        font_size = 40
        text_width = len(code) * (font_size - 10)
        x_offset = (width - text_width) // 2
        for i, char in enumerate(code):
            # 随机颜色
            color = (random.randint(0, 150), 
                    random.randint(0, 150), 
                    random.randint(0, 150))
            # 随机旋转
            angle = random.randint(-20, 20)
            # 这里简化处理，实际应该使用 ImageFont
            draw.text((x_offset + i * (font_size - 5), 20), 
                     char, fill=color)
        # 添加干扰线
        for _ in range(5):
            x1 = random.randint(0, width)
            y1 = random.randint(0, height)
            x2 = random.randint(0, width)
            y2 = random.randint(0, height)
            draw.line([(x1, y1), (x2, y2)], 
                     fill=(random.randint(0, 200), 
                          random.randint(0, 200), 
                          random.randint(0, 200)),
                     width=1)
        # 保存到内存
        img_buffer = io.BytesIO()
        image.save(img_buffer, format='PNG')
        img_buffer.seek(0)
        # 存储验证码
        self.captcha_store[session_id] = {
            'code': code,
            'timestamp': time.time(),
            'attempts': 0
        }
        return base64.b64encode(img_buffer.getvalue()).decode()
    def verify(self, session_id, user_input, max_attempts=3):
        """验证验证码"""
        if session_id not in self.captcha_store:
            return False, 'Invalid session'
        captcha_data = self.captcha_store[session_id]
        # 检查过期
        if time.time() - captcha_data['timestamp'] > 300:  # 5分钟过期
            del self.captcha_store[session_id]
            return False, 'Captcha expired'
        # 检查尝试次数
        if captcha_data['attempts'] >= max_attempts:
            del self.captcha_store[session_id]
            return False, 'Too many attempts'
        captcha_data['attempts'] += 1
        # 验证
        if user_input.upper() == captcha_data['code']:
            del self.captcha_store[session_id]
            return True, 'Success'
        return False, 'Wrong captcha'
# 使用示例
from flask import session
@app.route('/api/secure-data', methods=['POST'])
def secure_data():
    captcha = DynamicCaptcha()
    # 如果是新会话，要求验证码
    if 'captcha_verified' not in session:
        captcha_image = captcha.generate(session.sid)
        return jsonify({
            'require_captcha': True,
            'captcha_image': captcha_image
        })
    # 验证通过后执行业务逻辑
    return jsonify({'data': 'protected data'})
@app.route('/api/verify-captcha', methods=['POST'])
def verify_captcha():
    captcha = DynamicCaptcha()
    data = request.json
    success, message = captcha.verify(
        session.sid, 
        data.get('captcha_input', '')
    )
    if success:
        session['captcha_verified'] = True
        return jsonify({'success': True})
    return jsonify({'success': False, 'message': message})

综合防护中间件

import functools
from flask import g, current_app
class ComprehensiveAntiCrawlerMiddleware:
    """综合反爬虫中间件"""
    def __init__(self, app=None):
        self.app = app
        self.protection_layers = []
    def init_app(self, app):
        self.app = app
        app.before_request(self.check_all_layers)
    def add_layer(self, layer):
        """添加防护层"""
        self.protection_layers.append({
            'check': layer['check'],
            'name': layer.get('name', 'unnamed'),
            'severity': layer.get('severity', 1),
            'block_action': layer.get('block_action', self.default_block_action)
        })
    def default_block_action(self, severity, ip):
        """默认的封锁操作"""
        current_app.logger.warning(f"Blocking IP {ip} with severity {severity}")
    def check_all_layers(self):
        """检查所有防护层"""
        ip = request.remote_addr
        for layer in self.protection_layers:
            if not layer['check'](request):
                layer['block_action'](layer['severity'], ip)
                return jsonify({
                    'error': 'Access denied',
                    'code': 403
                }), 403
    def __call__(self, environ, start_response):
        return self.app(environ, start_response)
# 配置示例
def create_protected_app():
    app = Flask(__name__)
    anti_crawler = ComprehensiveAntiCrawlerMiddleware()
    anti_crawler.init_app(app)
    # 添加各防护层
    anti_crawler.add_layer({
        'name': 'rate_limit',
        'check': lambda r: check_rate_limit(r.remote_addr),
        'severity': 1
    })
    anti_crawler.add_layer({
        'name': 'fingerprint_check',
        'check': lambda r: check_fingerprint(r),
        'severity': 2,
        'block_action': lambda sev, ip: block_ip_permanently(ip)
    })
    return app
def check_rate_limit(ip):
    """检查速率限制"""
    # 实现速率限制逻辑
    return True
def check_fingerprint(request):
    """检查浏览器指纹"""
    # 实现指纹检查
    return True
def block_ip_permanently(ip):
    """永久封锁IP"""
    # 实现IP封锁
    pass

使用建议

# 配置文件示例
ANTI_CRAWLER_CONFIG = {
    'rate_limiting': {
        'max_requests_per_minute': 30,
        'max_requests_per_hour': 500,
        'ban_on_exceed': True,
        'ban_duration': 3600  # 秒
    },
    'fingerprint': {
        'required_headers': [
            'User-Agent',
            'Accept',
            'Accept-Language',
            'Accept-Encoding'
        ],
        'javascript_check': {
            'enabled': True,
            'challenge_difficulty': 'medium'
        }
    },
    'captcha': {
        'enabled': True,
        'difficulty_levels': {
            'easy': 4,
            'medium': 6,
            'hard': 8
        },
        'expire_time': 300
    },
    'behavior_analysis': {
        'mouse_movement': True,
        'scroll_pattern': True,
        'click_interval': True,
        'human_likeness_threshold': 0.8
    },
    'logging': {
        'log_all_attempts': True,
        'log_blocked_requests': True,
        'alert_on_suspicious': True
    }
}
# 使用配置
class ConfigurableAntiCrawler:
    def __init__(self, config=None):
        self.config = config or ANTI_CRAWLER_CONFIG
        self.setup_protection()
    def setup_protection(self):
        """根据配置设置防护"""
        if self.config['rate_limiting']['enabled']:
            self.enable_rate_limiting()
        if self.config['captcha']['enabled']:
            self.enable_captcha()
        if self.config['behavior_analysis']['enabled']:
            self.enable_behavior_analysis()

这些示例覆盖了从基础到高级的反爬虫防护方案,你可以根据实际需求选择并组合使用。

1. 渐进式防护：先使用简单的速率限制，逐步添加更复杂的检测
2. 平衡用户体验：过于严格的防护可能影响正常用户
3. 及时更新：爬虫技术也在发展，需要持续改进防护策略
4. 记录日志：记录所有防护行为，便于分析和优化