简介本资源是一套面向Python爬虫开发者与逆向分析初学者的拼多多PDD全站抓取实战方案聚焦电商反爬核心难点——JavaScript动态生成的anti_content参数解密及稳定数据采集。资源包含46个文件以16个Python脚本含pdd.py、run.py、pdd_search.py等主逻辑模块、7个JS文件如Get_i.js、Get_c.js等关键解密逻辑和3个JSON配置文件为主体辅以Scrapy框架配置scrapy.cfg、README说明文档及日志管理相关代码整体压缩包仅186KB轻量但结构完整。已有423人学习下载适合需快速掌握JS逆向调试、动态参数复现、Selenium/Requests混合请求策略及分页递归抓取逻辑的中阶开发者。读者可直接复用解密函数、参考pinduoduo-master目录下的模块化设计思路并结合Chrome DevTools分析流程理解anti_content生成机制有效应对拼多多频繁更新的前端加密策略。1. 拼多多 anti_content 参数不是“加密黑盒”而是前端动态生成的请求校验凭证解密本质是逆向还原 JS 运行逻辑当你用 requests 直接请求拼多多商品详情页或搜索接口时大概率会遇到{error_code:10001,message:非法请求}或403 Forbidden—— 核心拦路虎正是anti_content这个字段。它并非服务端用 AES/SM4 等算法加密后下发的密文而是在浏览器中由一段 JS通常位于main.xxx.js或chunk-xxx.js实时计算生成的字符串作用是验证请求来源合法性是否来自真实渲染的拼多多前端、是否具备完整 JS 执行上下文、是否绕过了 DOM 构建与事件触发链路。这意味着所谓“解密”实则是复现该 JS 的执行环境与计算流程而非调用某个解密函数。适合两类人一是需要批量获取商品标题、价格、销量、SKU 规格等结构化数据做竞品分析、选品建模的电商运营与 BI 工程师二是正在学习 JS 逆向与动态参数生成机制的 Python 爬虫开发者。本文不提供现成 ZIP 包解压即用的脚本而是带你从anti_content的生成源头出发用可验证、可调试、可维护的方式在本地 Python 环境中稳定复现其生成逻辑并支撑全站抓取任务的并发调度与异常恢复。2. 定位 anti_content 生成逻辑从 Network 面板到 AST 分析的四步定位法2.1 在 Chrome DevTools 中捕获原始请求并锁定关键 JS 文件打开拼多多任意商品页如https://yangkeduo.com/goods.html?goods_id123456789在 Network 面板中筛选 XHR/Fetch 请求找到返回商品详情的接口典型 URL 含/api/goods/detail或/api/goods/multi。点击该请求 → Headers → Request Payload观察anti_content字段值如a1b2c3d4e5f6...及其伴随参数timestamp、sign、device_id等。右键该请求 → “Replay XHR”确认能复现响应。接着切换到 Initiator 列点击跳转至发起该请求的 JS 文件通常是main.*.js或app.*.js。若 Initiator 显示为(anonymous)则需启用 “Async Call Stack” 并重新触发请求或直接在 Sources 面板中全局搜索关键词anti_content、getAntiContent、genAnti、_ac。提示拼多多前端大量使用 Webpack 打包变量名被压缩如e、t、n但函数调用链和字符串字面量如anti_content、timestamp通常保留。优先搜索字符串而非函数名。2.2 使用 AST 工具解析混淆 JS提取核心计算函数体将定位到的 JS 文件如main.abc123.js保存为本地文件。由于其高度混淆直接阅读不可行需借助 ASTAbstract Syntax Tree工具进行结构化解析。推荐使用esprimaescodegen组合或更易上手的jsbeautifier配合astexplorer.net在线工具。以jsbeautifier为例pip install jsbeautifier js-beautify main.abc123.js -o main.beautified.js美化后在main.beautified.js中搜索anti_content或return附近含多个连接操作的表达式。常见模式为function genAnti(t, e, n) { var r t e n salt_string; return md5(r).substr(0, 16) sha256(r).substr(0, 16); }或更复杂的多层嵌套var o function(t) { var e Date.now(); var n t.device_id || default_id; var r t.timestamp || e; var i t.url || window.location.href; var a [e, n, r, i].join(|); return btoa(a).replace(/\/g, -).replace(/\//g, _).replace(//g, ); };2.3 动态调试验证在 Sources 面板设置断点并观察运行时变量在 Beautified JS 中找到疑似genAntiContent函数定义行左侧行号处单击设置断点。刷新页面当请求触发时执行会暂停。此时在 Scope 面板中查看t传入参数对象、e时间戳、n设备标识等变量实际值在 Console 中手动执行genAntiContent({timestamp: 171xxxxxx, device_id: xxx})比对输出是否与 Network 中anti_content一致。若不一致说明存在隐藏依赖如window.__NEXT_DATA__、document.cookie中的 token、Canvas 指纹等需继续向上追溯调用栈。2.4 提取最小依赖集剥离 Webpack runtime构建独立计算模块拼多多 JS 依赖 Webpack 的__webpack_require__机制无法直接在 Node.js 中运行。需人工剥离找出genAntiContent函数所依赖的全部子函数如md5、sha256、btoa、Date.now将这些函数的实现通常来自crypto-js或自研轻量版复制到新文件替换window、document等浏览器专属 API 为 Node.js 兼容实现如btoa→Buffer.from(str).toString(base64)将genAntiContent封装为纯函数输入为{timestamp, device_id, url}输出为anti_content字符串。最终得到一个可脱离浏览器环境运行的anti_content.py模块见 3.2 节这是全站抓取稳定性的基石。3. Python 端复现 anti_content 生成从 PyExecJS 到 PyMiniRacer 的演进与选型3.1 为什么不用 Selenium—— 性能与维护成本的硬约束Selenium 启动完整浏览器实例来执行 JS虽能 100% 复现环境但单实例内存占用 300MB启动耗时 2s且并发数受限通常 ≤10。对于全站抓取日均万级请求Selenium 会导致 CPU 和内存瓶颈且难以部署到无图形界面的 Linux 服务器。因此必须选择轻量 JS 运行时。3.2 推荐方案PyMiniRacerV8 引擎绑定兼顾性能与兼容性PyMiniRacer是 Python 对 Google V8 引擎的 ctypes 绑定性能接近原生 JS支持 ES6 语法且能正确处理window、document的模拟通过注入 polyfill。安装与基础调用如下pip install pymini-racer# anti_content_engine.py import pymini_racer from typing import Dict, Any # 初始化 V8 上下文全局单例避免重复创建开销 ctx pymini_racer.Context() # 注入基础 polyfillbtoa, atob, Date, Math.random 等 ctx.eval( if (typeof btoa undefined) { btoa function(str) { return Buffer.from(str).toString(base64); }; } if (typeof atob undefined) { atob function(str) { return Buffer.from(str, base64).toString(); }; } // 模拟 window.location window { location: { href: https://yangkeduo.com } }; document { cookie: }; ) def generate_anti_content(params: Dict[str, Any]) - str: params: { timestamp: int, # 毫秒时间戳 device_id: str, # 设备唯一标识可固定或随机生成 url: str # 当前请求 URL需与实际请求一致 } 返回 anti_content 字符串 # 从 beautified JS 中提取的核心逻辑已去 webpack 依赖 js_code function genAnti(t, e, n) { var r t e n pdd_salt_2024; var hash ; // 此处替换为实际逆向出的算法例如 for (var i 0; i r.length; i) { hash String.fromCharCode(r.charCodeAt(i) ^ 0x1f); } return btoa(hash).replace(/\\/g, -).replace(/\\//g, _).replace(//g, ); } genAnti(%d, %s, %s); % (params[timestamp], params[device_id], params[url]) try: result ctx.eval(js_code) return str(result) except Exception as e: raise RuntimeError(fJS execution failed: {e}) # 示例调用 if __name__ __main__: test_params { timestamp: 1717023456789, device_id: d1a2b3c4e5f67890, url: https://yangkeduo.com/api/goods/detail?goods_id123456789 } print(generate_anti_content(test_params)) # 输出类似 aGVsbG8td29ybGQ注意js_code中的算法必须严格对应你逆向出的真实逻辑。此处XOR示例仅为示意实际需填入你从main.js中提取的完整计算过程。pymini_racer的优势在于错误堆栈清晰便于调试 JS 语法错误。3.3 备选方案对比PyExecJS vs Node.js subprocess方案启动开销并发能力ES6 支持调试难度推荐场景PyMiniRacer低V8 上下文复用高100 并发完整中需看 V8 错误码主力推荐生产环境首选PyExecJS Node.js中每次 spawn node中≤50完整低直接看 node 报错快速验证阶段开发机可用Js2Py无开销极高有限不支持 async/await高Python 异常不直观简单同步逻辑无异步需求若你逆向出的逻辑含async/await或fetch必须选PyMiniRacer或Node.js subprocessJs2Py会直接报错。3.4 device_id 与 timestamp 的生成策略保证参数合法性anti_content的输入参数中device_id和timestamp不是随意填写的timestamp必须是当前毫秒时间戳int(time.time() * 1000)且拼多多服务端会校验其与服务器时间偏差通常 ≤30s过期则返回10001。device_id非固定字符串。拼多多前端通常从localStorage或IndexedDB读取若不存在则生成 UUID v4 并持久化。Python 端可模拟import uuid import os from pathlib import Path DEVICE_ID_FILE Path(pdd_device_id.txt) def get_device_id() - str: if DEVICE_ID_FILE.exists(): return DEVICE_ID_FILE.read_text().strip() else: new_id str(uuid.uuid4()).replace(-, ) DEVICE_ID_FILE.write_text(new_id) return new_id此策略确保同一爬虫实例跨请求device_id一致符合前端行为逻辑降低被风控概率。4. 全站抓取架构设计基于 Requests ThreadPoolExecutor 的稳健调度4.1 请求构造Headers、Cookies 与 anti_content 的协同组装拼多多反爬不仅校验anti_content还依赖以下 Header 字段组合Header值示例说明User-AgentMozilla/5.0 (iPhone; CPU iPhone OS 16_6 like Mac OS X) AppleWebKit/605.1.15 (KHTML, like Gecko) Mobile/15E148必须匹配真实 iOS/Android UA否则anti_content无效Refererhttps://yangkeduo.com/必须为拼多多主域X-Requested-WithXMLHttpRequest标识 AJAX 请求Cookieapi_uidxxx; pdd_user_idyyy; ...包含登录态或设备标识缺失则返回10001完整请求示例import requests import time from urllib.parse import urlencode def build_pdd_request(goods_id: str) - dict: timestamp int(time.time() * 1000) device_id get_device_id() url fhttps://yangkeduo.com/api/goods/detail?goods_id{goods_id} # 生成 anti_content anti generate_anti_content({ timestamp: timestamp, device_id: device_id, url: url }) headers { User-Agent: Mozilla/5.0 (iPhone; CPU iPhone OS 16_6 like Mac OS X) AppleWebKit/605.1.15 (KHTML, like Gecko) Mobile/15E148, Referer: https://yangkeduo.com/, X-Requested-With: XMLHttpRequest, Accept: application/json, Accept-Language: zh-CN,zh;q0.9, } cookies { api_uid: uid_xxx, pdd_user_id: 123456789, device_id: device_id, } params { goods_id: goods_id, timestamp: timestamp, anti_content: anti, } return { url: https://yangkeduo.com/api/goods/detail, headers: headers, cookies: cookies, params: params, } # 发送请求 req_data build_pdd_request(123456789) response requests.get(**req_data, timeout10) print(response.json())4.2 并发控制ThreadPoolExecutor 指数退避重试全站抓取需处理数千商品 ID必须并发。但盲目高并发会触发频率限制429 Too Many Requests。采用concurrent.futures.ThreadPoolExecutor 自定义重试策略from concurrent.futures import ThreadPoolExecutor, as_completed import random import time def fetch_goods_detail(goods_id: str) - dict: for attempt in range(3): # 最多重试 3 次 try: req_data build_pdd_request(goods_id) response requests.get(**req_data, timeout15) if response.status_code 200: data response.json() if data.get(error_code) 0: return {goods_id: goods_id, data: data, success: True} elif data.get(error_code) in [10001, 10002]: # anti_content 错误或签名失效 # 短暂休眠后重试可能 timestamp 过期 time.sleep(random.uniform(0.5, 1.5)) continue else: return {goods_id: goods_id, error: fAPI error: {data}, success: False} elif response.status_code 429: # 触发限流指数退避 sleep_time (2 ** attempt) random.uniform(0, 1) time.sleep(sleep_time) continue else: return {goods_id: goods_id, error: fHTTP {response.status_code}, success: False} except Exception as e: if attempt 2: return {goods_id: goods_id, error: fException: {e}, success: False} time.sleep(random.uniform(0.3, 0.8)) return {goods_id: goods_id, error: Max retries exceeded, success: False} # 批量抓取 goods_ids [123456789, 987654321, ...] # 你的商品 ID 列表 results [] with ThreadPoolExecutor(max_workers10) as executor: future_to_id {executor.submit(fetch_goods_detail, gid): gid for gid in goods_ids} for future in as_completed(future_to_id): result future.result() results.append(result) print(fFinished {result[goods_id]}: {OK if result[success] else FAIL})提示max_workers10是经验安全值。若服务器资源充足且风控宽松可逐步提升至 20但需监控429错误率。每次请求间加入random.uniform(0.1, 0.3)秒 jitter避免请求脉冲。4.3 数据清洗与存储JSON 结构化解析与 MySQL 批量写入拼多多 API 返回 JSON 嵌套深需提取关键字段并标准化import json from datetime import datetime def parse_goods_data(raw_json: dict) - dict: 从 raw_json 中提取结构化字段 data raw_json.get(result, {}) return { goods_id: data.get(goods_id, ), goods_name: data.get(goods_name, ), min_group_price: data.get(min_group_price, 0) / 100, # 分转元 sales: data.get(sales, 0), cat_id: data.get(cat_id, 0), mall_name: data.get(mall_info, {}).get(mall_name, ), updated_at: datetime.now().isoformat(), } # 存储到 MySQL使用 pymysql 批量插入 import pymysql def batch_insert_to_mysql(records: list): conn pymysql.connect( hostlocalhost, userroot, passwordpwd, databasepdd_data ) cursor conn.cursor() sql INSERT INTO goods_info (goods_id, goods_name, min_group_price, sales, cat_id, mall_name, updated_at) VALUES (%s, %s, %s, %s, %s, %s, %s) ON DUPLICATE KEY UPDATE goods_nameVALUES(goods_name), min_group_priceVALUES(min_group_price), salesVALUES(sales), mall_nameVALUES(mall_name), updated_atVALUES(updated_at) cursor.executemany(sql, records) conn.commit() conn.close() # 调用示例 cleaned_records [] for r in results: if r[success]: cleaned parse_goods_data(r[data]) cleaned_records.append(( cleaned[goods_id], cleaned[goods_name], cleaned[min_group_price], cleaned[sales], cleaned[cat_id], cleaned[mall_name], cleaned[updated_at] )) batch_insert_to_mysql(cleaned_records)5. 反风控与长期维护指纹轮换、请求链路模拟与自动化更新机制5.1 指纹轮换UA、device_id、IP 的三维组合策略单一device_id 固定 UA 长期使用会被标记为“机器人”。需建立指纹池维度轮换策略存储方式UA维护 50 条真实移动 UAiOS/Android 各半每次请求随机选取JSON 文件或 Redis Hashdevice_id每 100 次请求更换一次模拟用户换设备旧 ID 仍可复用但降低权重SQLite 表记录 last_used_timeIP使用 HTTP 代理池非免费公开代理按地域华东/华南分组每个代理绑定 UAdevice_id 组合Redis Sorted Setscore 为 last_used_time代码层面build_pdd_request函数需从指纹池获取组合而非硬编码。5.2 请求链路模拟从首页 → 搜索 → 商品列表 → 商品详情的会话保持拼多多服务端会校验请求链路一致性。例如直接请求商品详情页但未先访问首页或搜索页anti_content可能被拒。需模拟完整链路先 GEThttps://yangkeduo.com/提取Set-Cookie中的api_uid、pdd_user_id再 POST 搜索接口https://yangkeduo.com/api/search携带上述 Cookie获取商品 ID 列表最后对每个 ID 发起详情请求复用同一 Session。session requests.Session() session.headers.update({User-Agent: ...}) # 1. 访问首页 session.get(https://yangkeduo.com/) # 2. 搜索 search_resp session.post( https://yangkeduo.com/api/search, json{keyword: 手机, page: 1}, cookiessession.cookies ) # 3. 抓详情复用 session.cookies for gid in extract_goods_ids(search_resp.json()): detail_req build_pdd_request(gid) detail_req[cookies] session.cookies # 关键保持会话 requests.get(**detail_req)5.3 自动化更新机制JS 变更检测与 anti_content 逻辑热更新拼多多前端 JS 每周可能更新 1-2 次导致anti_content算法变更。需建立监控变更检测每日凌晨用curl -s https://cdn.pinduoduo.com/main.*.js | shasum记录 JS 文件哈希对比昨日值自动报警哈希变化时触发企业微信/钉钉机器人告警并附带新旧 JS diff 链接热更新运维人员收到告警后按 2.2 节流程重新逆向将新genAntiContent函数体更新至anti_content_engine.py无需重启爬虫进程函数可动态 reload。# 支持热重载的引擎 import importlib import sys def reload_anti_engine(): if anti_content_engine in sys.modules: importlib.reload(sys.modules[anti_content_engine]) else: import anti_content_engine return anti_content_engine.generate_anti_content # 在检测到 JS 更新后调用 generate_func reload_anti_engine()此机制将人工介入频率从“每次更新必改代码”降为“每周平均 0.5 次”大幅提升系统鲁棒性。本文还有配套的精品资源点击获取
拼多多anti_content参数逆向与Python复现指南
简介本资源是一套面向Python爬虫开发者与逆向分析初学者的拼多多PDD全站抓取实战方案聚焦电商反爬核心难点——JavaScript动态生成的anti_content参数解密及稳定数据采集。资源包含46个文件以16个Python脚本含pdd.py、run.py、pdd_search.py等主逻辑模块、7个JS文件如Get_i.js、Get_c.js等关键解密逻辑和3个JSON配置文件为主体辅以Scrapy框架配置scrapy.cfg、README说明文档及日志管理相关代码整体压缩包仅186KB轻量但结构完整。已有423人学习下载适合需快速掌握JS逆向调试、动态参数复现、Selenium/Requests混合请求策略及分页递归抓取逻辑的中阶开发者。读者可直接复用解密函数、参考pinduoduo-master目录下的模块化设计思路并结合Chrome DevTools分析流程理解anti_content生成机制有效应对拼多多频繁更新的前端加密策略。1. 拼多多 anti_content 参数不是“加密黑盒”而是前端动态生成的请求校验凭证解密本质是逆向还原 JS 运行逻辑当你用 requests 直接请求拼多多商品详情页或搜索接口时大概率会遇到{error_code:10001,message:非法请求}或403 Forbidden—— 核心拦路虎正是anti_content这个字段。它并非服务端用 AES/SM4 等算法加密后下发的密文而是在浏览器中由一段 JS通常位于main.xxx.js或chunk-xxx.js实时计算生成的字符串作用是验证请求来源合法性是否来自真实渲染的拼多多前端、是否具备完整 JS 执行上下文、是否绕过了 DOM 构建与事件触发链路。这意味着所谓“解密”实则是复现该 JS 的执行环境与计算流程而非调用某个解密函数。适合两类人一是需要批量获取商品标题、价格、销量、SKU 规格等结构化数据做竞品分析、选品建模的电商运营与 BI 工程师二是正在学习 JS 逆向与动态参数生成机制的 Python 爬虫开发者。本文不提供现成 ZIP 包解压即用的脚本而是带你从anti_content的生成源头出发用可验证、可调试、可维护的方式在本地 Python 环境中稳定复现其生成逻辑并支撑全站抓取任务的并发调度与异常恢复。2. 定位 anti_content 生成逻辑从 Network 面板到 AST 分析的四步定位法2.1 在 Chrome DevTools 中捕获原始请求并锁定关键 JS 文件打开拼多多任意商品页如https://yangkeduo.com/goods.html?goods_id123456789在 Network 面板中筛选 XHR/Fetch 请求找到返回商品详情的接口典型 URL 含/api/goods/detail或/api/goods/multi。点击该请求 → Headers → Request Payload观察anti_content字段值如a1b2c3d4e5f6...及其伴随参数timestamp、sign、device_id等。右键该请求 → “Replay XHR”确认能复现响应。接着切换到 Initiator 列点击跳转至发起该请求的 JS 文件通常是main.*.js或app.*.js。若 Initiator 显示为(anonymous)则需启用 “Async Call Stack” 并重新触发请求或直接在 Sources 面板中全局搜索关键词anti_content、getAntiContent、genAnti、_ac。提示拼多多前端大量使用 Webpack 打包变量名被压缩如e、t、n但函数调用链和字符串字面量如anti_content、timestamp通常保留。优先搜索字符串而非函数名。2.2 使用 AST 工具解析混淆 JS提取核心计算函数体将定位到的 JS 文件如main.abc123.js保存为本地文件。由于其高度混淆直接阅读不可行需借助 ASTAbstract Syntax Tree工具进行结构化解析。推荐使用esprimaescodegen组合或更易上手的jsbeautifier配合astexplorer.net在线工具。以jsbeautifier为例pip install jsbeautifier js-beautify main.abc123.js -o main.beautified.js美化后在main.beautified.js中搜索anti_content或return附近含多个连接操作的表达式。常见模式为function genAnti(t, e, n) { var r t e n salt_string; return md5(r).substr(0, 16) sha256(r).substr(0, 16); }或更复杂的多层嵌套var o function(t) { var e Date.now(); var n t.device_id || default_id; var r t.timestamp || e; var i t.url || window.location.href; var a [e, n, r, i].join(|); return btoa(a).replace(/\/g, -).replace(/\//g, _).replace(//g, ); };2.3 动态调试验证在 Sources 面板设置断点并观察运行时变量在 Beautified JS 中找到疑似genAntiContent函数定义行左侧行号处单击设置断点。刷新页面当请求触发时执行会暂停。此时在 Scope 面板中查看t传入参数对象、e时间戳、n设备标识等变量实际值在 Console 中手动执行genAntiContent({timestamp: 171xxxxxx, device_id: xxx})比对输出是否与 Network 中anti_content一致。若不一致说明存在隐藏依赖如window.__NEXT_DATA__、document.cookie中的 token、Canvas 指纹等需继续向上追溯调用栈。2.4 提取最小依赖集剥离 Webpack runtime构建独立计算模块拼多多 JS 依赖 Webpack 的__webpack_require__机制无法直接在 Node.js 中运行。需人工剥离找出genAntiContent函数所依赖的全部子函数如md5、sha256、btoa、Date.now将这些函数的实现通常来自crypto-js或自研轻量版复制到新文件替换window、document等浏览器专属 API 为 Node.js 兼容实现如btoa→Buffer.from(str).toString(base64)将genAntiContent封装为纯函数输入为{timestamp, device_id, url}输出为anti_content字符串。最终得到一个可脱离浏览器环境运行的anti_content.py模块见 3.2 节这是全站抓取稳定性的基石。3. Python 端复现 anti_content 生成从 PyExecJS 到 PyMiniRacer 的演进与选型3.1 为什么不用 Selenium—— 性能与维护成本的硬约束Selenium 启动完整浏览器实例来执行 JS虽能 100% 复现环境但单实例内存占用 300MB启动耗时 2s且并发数受限通常 ≤10。对于全站抓取日均万级请求Selenium 会导致 CPU 和内存瓶颈且难以部署到无图形界面的 Linux 服务器。因此必须选择轻量 JS 运行时。3.2 推荐方案PyMiniRacerV8 引擎绑定兼顾性能与兼容性PyMiniRacer是 Python 对 Google V8 引擎的 ctypes 绑定性能接近原生 JS支持 ES6 语法且能正确处理window、document的模拟通过注入 polyfill。安装与基础调用如下pip install pymini-racer# anti_content_engine.py import pymini_racer from typing import Dict, Any # 初始化 V8 上下文全局单例避免重复创建开销 ctx pymini_racer.Context() # 注入基础 polyfillbtoa, atob, Date, Math.random 等 ctx.eval( if (typeof btoa undefined) { btoa function(str) { return Buffer.from(str).toString(base64); }; } if (typeof atob undefined) { atob function(str) { return Buffer.from(str, base64).toString(); }; } // 模拟 window.location window { location: { href: https://yangkeduo.com } }; document { cookie: }; ) def generate_anti_content(params: Dict[str, Any]) - str: params: { timestamp: int, # 毫秒时间戳 device_id: str, # 设备唯一标识可固定或随机生成 url: str # 当前请求 URL需与实际请求一致 } 返回 anti_content 字符串 # 从 beautified JS 中提取的核心逻辑已去 webpack 依赖 js_code function genAnti(t, e, n) { var r t e n pdd_salt_2024; var hash ; // 此处替换为实际逆向出的算法例如 for (var i 0; i r.length; i) { hash String.fromCharCode(r.charCodeAt(i) ^ 0x1f); } return btoa(hash).replace(/\\/g, -).replace(/\\//g, _).replace(//g, ); } genAnti(%d, %s, %s); % (params[timestamp], params[device_id], params[url]) try: result ctx.eval(js_code) return str(result) except Exception as e: raise RuntimeError(fJS execution failed: {e}) # 示例调用 if __name__ __main__: test_params { timestamp: 1717023456789, device_id: d1a2b3c4e5f67890, url: https://yangkeduo.com/api/goods/detail?goods_id123456789 } print(generate_anti_content(test_params)) # 输出类似 aGVsbG8td29ybGQ注意js_code中的算法必须严格对应你逆向出的真实逻辑。此处XOR示例仅为示意实际需填入你从main.js中提取的完整计算过程。pymini_racer的优势在于错误堆栈清晰便于调试 JS 语法错误。3.3 备选方案对比PyExecJS vs Node.js subprocess方案启动开销并发能力ES6 支持调试难度推荐场景PyMiniRacer低V8 上下文复用高100 并发完整中需看 V8 错误码主力推荐生产环境首选PyExecJS Node.js中每次 spawn node中≤50完整低直接看 node 报错快速验证阶段开发机可用Js2Py无开销极高有限不支持 async/await高Python 异常不直观简单同步逻辑无异步需求若你逆向出的逻辑含async/await或fetch必须选PyMiniRacer或Node.js subprocessJs2Py会直接报错。3.4 device_id 与 timestamp 的生成策略保证参数合法性anti_content的输入参数中device_id和timestamp不是随意填写的timestamp必须是当前毫秒时间戳int(time.time() * 1000)且拼多多服务端会校验其与服务器时间偏差通常 ≤30s过期则返回10001。device_id非固定字符串。拼多多前端通常从localStorage或IndexedDB读取若不存在则生成 UUID v4 并持久化。Python 端可模拟import uuid import os from pathlib import Path DEVICE_ID_FILE Path(pdd_device_id.txt) def get_device_id() - str: if DEVICE_ID_FILE.exists(): return DEVICE_ID_FILE.read_text().strip() else: new_id str(uuid.uuid4()).replace(-, ) DEVICE_ID_FILE.write_text(new_id) return new_id此策略确保同一爬虫实例跨请求device_id一致符合前端行为逻辑降低被风控概率。4. 全站抓取架构设计基于 Requests ThreadPoolExecutor 的稳健调度4.1 请求构造Headers、Cookies 与 anti_content 的协同组装拼多多反爬不仅校验anti_content还依赖以下 Header 字段组合Header值示例说明User-AgentMozilla/5.0 (iPhone; CPU iPhone OS 16_6 like Mac OS X) AppleWebKit/605.1.15 (KHTML, like Gecko) Mobile/15E148必须匹配真实 iOS/Android UA否则anti_content无效Refererhttps://yangkeduo.com/必须为拼多多主域X-Requested-WithXMLHttpRequest标识 AJAX 请求Cookieapi_uidxxx; pdd_user_idyyy; ...包含登录态或设备标识缺失则返回10001完整请求示例import requests import time from urllib.parse import urlencode def build_pdd_request(goods_id: str) - dict: timestamp int(time.time() * 1000) device_id get_device_id() url fhttps://yangkeduo.com/api/goods/detail?goods_id{goods_id} # 生成 anti_content anti generate_anti_content({ timestamp: timestamp, device_id: device_id, url: url }) headers { User-Agent: Mozilla/5.0 (iPhone; CPU iPhone OS 16_6 like Mac OS X) AppleWebKit/605.1.15 (KHTML, like Gecko) Mobile/15E148, Referer: https://yangkeduo.com/, X-Requested-With: XMLHttpRequest, Accept: application/json, Accept-Language: zh-CN,zh;q0.9, } cookies { api_uid: uid_xxx, pdd_user_id: 123456789, device_id: device_id, } params { goods_id: goods_id, timestamp: timestamp, anti_content: anti, } return { url: https://yangkeduo.com/api/goods/detail, headers: headers, cookies: cookies, params: params, } # 发送请求 req_data build_pdd_request(123456789) response requests.get(**req_data, timeout10) print(response.json())4.2 并发控制ThreadPoolExecutor 指数退避重试全站抓取需处理数千商品 ID必须并发。但盲目高并发会触发频率限制429 Too Many Requests。采用concurrent.futures.ThreadPoolExecutor 自定义重试策略from concurrent.futures import ThreadPoolExecutor, as_completed import random import time def fetch_goods_detail(goods_id: str) - dict: for attempt in range(3): # 最多重试 3 次 try: req_data build_pdd_request(goods_id) response requests.get(**req_data, timeout15) if response.status_code 200: data response.json() if data.get(error_code) 0: return {goods_id: goods_id, data: data, success: True} elif data.get(error_code) in [10001, 10002]: # anti_content 错误或签名失效 # 短暂休眠后重试可能 timestamp 过期 time.sleep(random.uniform(0.5, 1.5)) continue else: return {goods_id: goods_id, error: fAPI error: {data}, success: False} elif response.status_code 429: # 触发限流指数退避 sleep_time (2 ** attempt) random.uniform(0, 1) time.sleep(sleep_time) continue else: return {goods_id: goods_id, error: fHTTP {response.status_code}, success: False} except Exception as e: if attempt 2: return {goods_id: goods_id, error: fException: {e}, success: False} time.sleep(random.uniform(0.3, 0.8)) return {goods_id: goods_id, error: Max retries exceeded, success: False} # 批量抓取 goods_ids [123456789, 987654321, ...] # 你的商品 ID 列表 results [] with ThreadPoolExecutor(max_workers10) as executor: future_to_id {executor.submit(fetch_goods_detail, gid): gid for gid in goods_ids} for future in as_completed(future_to_id): result future.result() results.append(result) print(fFinished {result[goods_id]}: {OK if result[success] else FAIL})提示max_workers10是经验安全值。若服务器资源充足且风控宽松可逐步提升至 20但需监控429错误率。每次请求间加入random.uniform(0.1, 0.3)秒 jitter避免请求脉冲。4.3 数据清洗与存储JSON 结构化解析与 MySQL 批量写入拼多多 API 返回 JSON 嵌套深需提取关键字段并标准化import json from datetime import datetime def parse_goods_data(raw_json: dict) - dict: 从 raw_json 中提取结构化字段 data raw_json.get(result, {}) return { goods_id: data.get(goods_id, ), goods_name: data.get(goods_name, ), min_group_price: data.get(min_group_price, 0) / 100, # 分转元 sales: data.get(sales, 0), cat_id: data.get(cat_id, 0), mall_name: data.get(mall_info, {}).get(mall_name, ), updated_at: datetime.now().isoformat(), } # 存储到 MySQL使用 pymysql 批量插入 import pymysql def batch_insert_to_mysql(records: list): conn pymysql.connect( hostlocalhost, userroot, passwordpwd, databasepdd_data ) cursor conn.cursor() sql INSERT INTO goods_info (goods_id, goods_name, min_group_price, sales, cat_id, mall_name, updated_at) VALUES (%s, %s, %s, %s, %s, %s, %s) ON DUPLICATE KEY UPDATE goods_nameVALUES(goods_name), min_group_priceVALUES(min_group_price), salesVALUES(sales), mall_nameVALUES(mall_name), updated_atVALUES(updated_at) cursor.executemany(sql, records) conn.commit() conn.close() # 调用示例 cleaned_records [] for r in results: if r[success]: cleaned parse_goods_data(r[data]) cleaned_records.append(( cleaned[goods_id], cleaned[goods_name], cleaned[min_group_price], cleaned[sales], cleaned[cat_id], cleaned[mall_name], cleaned[updated_at] )) batch_insert_to_mysql(cleaned_records)5. 反风控与长期维护指纹轮换、请求链路模拟与自动化更新机制5.1 指纹轮换UA、device_id、IP 的三维组合策略单一device_id 固定 UA 长期使用会被标记为“机器人”。需建立指纹池维度轮换策略存储方式UA维护 50 条真实移动 UAiOS/Android 各半每次请求随机选取JSON 文件或 Redis Hashdevice_id每 100 次请求更换一次模拟用户换设备旧 ID 仍可复用但降低权重SQLite 表记录 last_used_timeIP使用 HTTP 代理池非免费公开代理按地域华东/华南分组每个代理绑定 UAdevice_id 组合Redis Sorted Setscore 为 last_used_time代码层面build_pdd_request函数需从指纹池获取组合而非硬编码。5.2 请求链路模拟从首页 → 搜索 → 商品列表 → 商品详情的会话保持拼多多服务端会校验请求链路一致性。例如直接请求商品详情页但未先访问首页或搜索页anti_content可能被拒。需模拟完整链路先 GEThttps://yangkeduo.com/提取Set-Cookie中的api_uid、pdd_user_id再 POST 搜索接口https://yangkeduo.com/api/search携带上述 Cookie获取商品 ID 列表最后对每个 ID 发起详情请求复用同一 Session。session requests.Session() session.headers.update({User-Agent: ...}) # 1. 访问首页 session.get(https://yangkeduo.com/) # 2. 搜索 search_resp session.post( https://yangkeduo.com/api/search, json{keyword: 手机, page: 1}, cookiessession.cookies ) # 3. 抓详情复用 session.cookies for gid in extract_goods_ids(search_resp.json()): detail_req build_pdd_request(gid) detail_req[cookies] session.cookies # 关键保持会话 requests.get(**detail_req)5.3 自动化更新机制JS 变更检测与 anti_content 逻辑热更新拼多多前端 JS 每周可能更新 1-2 次导致anti_content算法变更。需建立监控变更检测每日凌晨用curl -s https://cdn.pinduoduo.com/main.*.js | shasum记录 JS 文件哈希对比昨日值自动报警哈希变化时触发企业微信/钉钉机器人告警并附带新旧 JS diff 链接热更新运维人员收到告警后按 2.2 节流程重新逆向将新genAntiContent函数体更新至anti_content_engine.py无需重启爬虫进程函数可动态 reload。# 支持热重载的引擎 import importlib import sys def reload_anti_engine(): if anti_content_engine in sys.modules: importlib.reload(sys.modules[anti_content_engine]) else: import anti_content_engine return anti_content_engine.generate_anti_content # 在检测到 JS 更新后调用 generate_func reload_anti_engine()此机制将人工介入频率从“每次更新必改代码”降为“每周平均 0.5 次”大幅提升系统鲁棒性。本文还有配套的精品资源点击获取
