针对Ajax动态网页的数据爬取,关键在于识别数据接口、分析加密参数以及实现解密算法。以下是完整的动态解密爬虫解决方案:

1.核心解密流程
动态网站通常通过Ajax异步加载数据,这些数据往往经过加密处理,无法直接从网页源码获取。解密过程主要包含以下步骤:

2.数据定位与抓包分析‌
使用浏览器开发者工具的Network面板监控XHR请求,通过触发页面事件或搜索关键词定位目标数据包。重点关注请求头中的User-Agent、Referer等反爬字段,以及POST请求的Form Data参数。

3.加密算法识别‌
通过全局搜索"decrypt"、"JSON.parse"等关键词定位解密函数。常见的加密方式包括AES、MD5、Base64等。AES加密通常采用ECB模式,填充方式为Pkcs7。

4.参数逆向分析‌
分析加密函数的关键参数,包括加密密钥、时间戳、签名等。对于JS加密,可通过打断点、单步调试的方式追踪参数生成过程。

5.完整爬虫实现


import requests
import json
import time
import hashlib
import base64
from Crypto.Cipher import AES
from Crypto.Util.Padding import unpad
import urllib.parse

class AjaxDynamicDecrypt:
    def __init__(self):
        self.session = requests.Session()
        self.headers = {
            'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/119.0.0.0 Safari/537.36',
            'Referer': 'https://target-website.com',
            'X-Requested-With': 'XMLHttpRequest'
        }
    
    def analyze_encryption(self, url):
        """分析加密参数"""
        print("开始分析加密参数...")
        
        # 监控网络请求获取加密接口
        try:
            response = self.session.get(url, headers=self.headers, timeout=10)
            if response.status_code == 200:
                print("页面加载成功,开始分析XHR请求...")
            else:
                print(f"页面加载失败: {response.status_code}")
        except Exception as e:
            print(f"请求异常: {e}")
    
    def aes_decrypt(self, encrypt_str, key_str, mode=AES.MODE_ECB):
        """AES解密函数"""
        try:
            key_str = key_str.encode('utf-8')
            aes = AES.new(key_str, mode)
            encrypt_str = base64.b64decode(encrypt_str)
            decrypt_str = aes.decrypt(encrypt_str).decode('utf-8')
            
            # 处理填充数据
            if decrypt_str.endswith('}'):
                return decrypt_str
            else:
                # 去除填充字符
                length = ord(decrypt_str[-1])
                return decrypt_str[:-length]
        except Exception as e:
            print(f"AES解密失败: {e}")
            return None
    
    def md5_encrypt(self, text):
        """MD5加密"""
        return hashlib.md5(text.encode('utf-8')).hexdigest()
    
    def base64_decode(self, encoded_str):
        """Base64解码"""
        try:
            return base64.b64decode(encoded_str).decode('utf-8')
        except Exception as e:
            print(f"Base64解码失败: {e}")
            return None
    
    def generate_timestamp(self):
        """生成时间戳"""
        return str(int(time.time() * 1000))
    
    def extract_encryption_key(self, js_content):
        """从JS代码中提取加密密钥"""
        # 搜索常见的密钥定义模式
        key_patterns = [
            r"key\s*[:=]\s*['\"]([^'\"]+)['\"]",
            r"password\s*[:=]\s*['\"]([^'\"]+)['\"]",
            r"secret\s*[:=]\s*['\"]([^'\"]+)['\"]"
        ]
        
        import re
        for pattern in key_patterns:
            match = re.search(pattern, js_content)
            if match:
                return match.group(1)
        
        return None
    
    def dynamic_request(self, api_url, params=None):
        """发送动态请求并处理加密响应"""
        try:
            if params is None:
                params = {}
            
            # 添加常见加密参数
            params['timestamp'] = self.generate_timestamp()
            params['sign'] = self.md5_encrypt(json.dumps(params)))
            
            response = self.session.get(api_url, params=params, headers=self.headers, timeout=10)
            
            if response.status_code == 200:
                data = response.json()
                
                # 检查数据是否加密
                if 'data' in data and isinstance(data['data'], str):
                    print("检测到加密数据,开始解密...")
                    # 这里需要根据实际情况调整密钥
                    key = 'your_encryption_key'
                    decrypted_data = self.aes_decrypt(data['data'], key)
                    
                    if decrypted_data:
                        return json.loads(decrypted_data)
                    else:
                        print("解密失败,返回原始数据")
                        return data
                else:
                    return data
            else:
                print(f"API请求失败: {response.status_code}")
                return None
                
        except Exception as e:
            print(f"动态请求异常: {e}")
            return None
    
    def bypass_anti_crawler(self):
        """绕过反爬虫检测"""
        # 使用反检测浏览器驱动
        try:
            import undetected_chromedriver as uc
            options = uc.ChromeOptions()
            driver = uc.Chrome(options=options)
            return driver
        except ImportError:
            print("未安装undetected_chromedriver,使用普通请求方式")
        return None

def main():
    crawler = AjaxDynamicDecrypt()
    
    # 示例:分析目标网站
    target_url = "https://example.com/dynamic-page"
    crawler.analyze_encryption(target_url)
    
    # 绕过反爬虫
    driver = crawler.bypass_anti_crawler()
    
    if driver:
        print("使用浏览器驱动成功绕过反爬虫")
        # 这里可以添加浏览器自动化操作
    else:
        print("使用普通请求方式")

if __name__ == "__main__":
    main()

6.关键技术要点
‌6.1加密算法识别与实现‌
通过全局搜索定位解密函数,识别使用的加密算法类型。AES解密需要密钥和加密模式参数,可通过打断点方式获取。MD5加密常用于生成签名参数,具有不可逆特性。
6‌.2反爬虫绕过策略‌
对于复杂的JS加密,可采用undetected_chromedriver库模拟真实浏览器行为,有效规避反爬检测。同时需要完善请求头信息,模拟正常用户访问。
‌6.3动态参数处理‌
时间戳、随机数、签名等动态参数是Ajax请求的关键。需要分析参数生成逻辑,在Python中重现加密过程。

该解决方案提供了完整的动态解密爬虫框架,包含加密分析、算法实现和反爬绕过等核心功能,适用于大多数Ajax动态网站的爬取需求。

Logo

北京人形旗下天工造物具身智能开源社区,聚焦具身天工与慧思开物两大平台

更多推荐