简介本资源是一套面向Python爬虫开发者与逆向分析初学者的拼多多PDD全站抓取实战方案聚焦电商反爬核心难点——JavaScript动态生成的anti_content参数解密及稳定数据采集。资源提供从JS逻辑逆向、参数复现到Scrapy框架集成的完整技术路径覆盖动态加密分析、Selenium模拟、请求签名还原等关键能力适用于竞品监控、价格分析、商品舆情等实际业务场景。压缩包共46个文件含16个核心Python脚本如pdd.py、run.py、pdd_search.py、7个JS逆向辅助文件Get_i.js、Get_c.js等、3个JSON配置与元数据文件以及scrapy.cfg、README.md等工程化支撑文件整体仅186KB轻量易部署。已有423人学习下载内容结构清晰主目录pinduoduo-master下模块分层明确包含请求封装、JS解密函数、搜索与详情页解析逻辑及Sanic轻量服务接口附带详细注释与可直接调试的运行入口是理解电商平台反爬机制与落地实践的高价值参考样本。1. 拼多多 anti_content 参数不是“加密”而是运行时上下文快照它本质是 JS 执行环境的一次性签名解密失败往往源于没还原出正确的 window 对象状态很多人一看到anti_content就去翻 OpenSSL 或 AES 密钥结果卡在 base64 解码后乱码、MD5 校验不通过、时间戳对不上——这不是算法错了是根本没理解它的生成逻辑。anti_content在拼多多最新版本中2024 Q2 主流流量路径并非服务端下发的密文而是前端 JS 在特定 DOM 状态、特定 timing、特定浏览器环境含 canvas fingerprint、navigator.plugins、performance.now() 采样点下调用window.__anti__函数动态生成的字符串快照。它不依赖固定密钥而依赖执行时刻的「环境熵」。这意味着用js2py直接 eval 一段剥离后的 JS 代码必然失败用 Selenium 启动空白页加载脚本也大概率失效——因为缺失了页面初始化阶段的document.documentElement属性注入、__NEXT_DATA__全局变量挂载、以及window._pdd_初始化链中的 3 个关键副作用函数。真正能复现的路径只有一条在真实页面上下文中触发getAntiContent()调用并捕获其返回值。本资源包里的Get_i.js、Get_c.js、Get_u.js并非独立解密器而是三组用于不同请求场景商品详情 / 搜索列表 / 用户行为上报的「环境适配钩子」它们共同指向一个核心事实anti_content 的生成必须与页面生命周期强绑定。适合正在调试requests execjs组合却始终返回{error:invalid anti}的中级爬虫工程师也适合刚接触 JS 逆向、误以为「找到加密函数就能硬解」的新手——本文将带你绕过所有「伪解密」陷阱直击真实执行链。2. 反编译与环境重建从Get_c.js入口定位__anti__函数并还原其依赖的 window 上下文2.1 定位 anti_content 生成函数的真实位置与调用链打开pinduoduo-master/Get_c.js第一行即为关键线索const anti window.__anti__(window._pdd_.data, window._pdd_.config)。这说明__anti__是全局函数且接收两个参数_pdd_.data页面数据快照和_pdd_.config运行时配置。但直接搜索window.__anti__在源码中找不到定义——它被 webpack 打包进app.xxx.js的闭包内。此时需借助 Chrome DevTools 的Sources → Page → Breakpoints → XHR/fetch breakpoints功能在发起含anti_content的请求如/api/search/items前设置断点。刷新页面后在 Call Stack 中向上追溯最终停在app.xxx.js:12345的某处匿名函数内该函数体包含return e(t,n,r,i)形式调用其中e即__anti__的实际引用。右键 →Show function definition即可跳转到其完整实现。典型结构如下window.__anti__ function(e, t) { var n e || {}, r t || {}, i window.performance.now(), a window.screen.width x window.screen.height, o navigator.platform, s document.documentElement.scrollHeight, u (new Date).getTimezoneOffset(), c window._pdd_.version || 1.0.0, l [i, a, o, s, u, c].join(|); return btoa(sha256(l r.salt n.token)); };提示r.salt和n.token并非常量而是由window._pdd_.config.salt和window._pdd_.data.token动态提供二者均在页面加载早期由script注入或 AJAX 返回后赋值。若未等待document.readyState complete且_pdd_已挂载r.salt会是undefined导致sha256(undefined)报错。2.2 构建可复用的 JS 执行环境用 Puppeteer 注入并捕获 anti_content单纯用execjs或js2py无法模拟performance.now()的高精度时间戳、screen.width/height的真实分辨率、document.documentElement.scrollHeight的渲染后高度。必须使用真实浏览器上下文。pdd.py中的get_anti_content()方法采用 Puppeteer但原始代码存在致命缺陷它在page.goto()后立即执行page.evaluate()此时页面可能尚未完成_pdd_初始化。修正后的流程如下# pdd_sanic/pdd.py from pyppeteer import launch import asyncio async def get_anti_content(url: str) - str: browser await launch(headlessTrue, args[--no-sandbox, --disable-setuid-sandbox]) page await browser.newPage() # 关键等待 _pdd_ 对象完全就绪 await page.goto(url, waitUntilnetworkidle0) # 等待网络空闲 await page.waitForFunction(window._pdd_ window._pdd_.data window._pdd_.config, timeout10000) # 注入 Get_c.js 中的 anti 生成逻辑需提前读取并拼接 anti_js function generateAnti() { if (!window.__anti__) return null; const data window._pdd_.data || {}; const config window._pdd_.config || {}; return window.__anti__(data, config); } generateAnti(); result await page.evaluate(anti_js) await browser.close() return result2.2.1 参数说明与关键校验点waitUntilnetworkidle0确保所有网络请求包括异步 JS 加载完成比load更可靠waitForFunction显式等待_pdd_对象及其子属性存在避免Cannot read property data of undefined错误generateAnti()函数封装将Get_c.js中的调用逻辑包裹为自执行函数规避作用域污染result类型返回str或None需在调用方做空值判断否则后续请求会因anti_contentnull被拦截。2.3 验证环境还原是否成功对比真实请求与模拟生成的 anti_content 值仅生成字符串还不够必须验证其与真实请求一致。抓取一次拼多多搜索页如https://yangkeduo.com/search_result.html?search_key手机在 Network 面板中找到/api/search/items请求复制其anti_content参数值形如ZmFsc2U。再运行上述get_anti_content()函数传入相同 URL打印返回值。若两者完全一致注意 base64 编码无换行说明环境重建成功若差异按以下顺序排查检查 Puppeteer 启动参数是否包含--user-agent拼多多会校验 UA 是否匹配主流 Chrome 版本如Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/124.0.0.0 Safari/537.36查看page.content()中是否包含scriptwindow._pdd_ {...}/script片段若缺失说明页面未正确加载 JS 初始化脚本在page.evaluate()中添加console.log(window._pdd_.config.salt, window._pdd_.data.token)确认两字段非空且格式合理salt 通常为 16 位 hex 字符串token 为 JWT-like 结构。3. 全站抓取架构设计基于 Scrapy 的分层调度与 anti_content 动态注入策略3.1 Scrapy 项目结构优化分离 anti_content 获取与业务逻辑scrapy.cfg和pinduoduo-master目录下的scrapy.cfg表明项目已采用 Scrapy 框架但原始pdd.py将 anti_content 获取硬编码在 Spider 内导致每次请求都启动新浏览器效率极低。应重构为中间件 异步池模式# middlewares.py from scrapy import signals from scrapy.downloadermiddlewares.retry import RetryMiddleware from twisted.internet.defer import inlineCallbacks from pdd_sanic.anti_pool import AntiPool # 自定义 anti_content 复用池 class AntiContentMiddleware: def __init__(self): self.anti_pool AntiPool(max_size5) # 最多缓存 5 个不同域名的 anti 实例 def process_request(self, request, spider): # 从 URL 提取 domain复用对应 anti 实例 domain request.url.split(/)[2] anti_value self.anti_pool.get(domain) if not anti_value: # 异步获取避免阻塞 d self.anti_pool.fetch_async(domain) d.addCallback(lambda v: setattr(request.headers, anti_content, v)) return d request.headers[anti_content] anti_value3.1.1 AntiPool 的核心实现逻辑AntiPool不是简单字典缓存而是维护一组长期存活的 Puppeteer 页面实例每个实例对应一个域名定期如每 30 秒重新生成anti_content并更新缓存。其fetch_async()方法返回 Twisted Deferred与 Scrapy 的异步机制无缝集成# pdd_sanic/anti_pool.py from twisted.internet import reactor, defer from pyppeteer import launch class AntiPool: def __init__(self, max_size5): self.pool {} self.max_size max_size self.launcher None inlineCallbacks def fetch_async(self, domain): if domain not in self.pool: if len(self.pool) self.max_size: # 清理最旧的实例 oldest min(self.pool.keys(), keylambda k: self.pool[k][ts]) yield self.pool[oldest][page].close() del self.pool[oldest] # 启动新页面并预热 if not self.launcher: self.launcher yield launch(headlessTrue, args[--no-sandbox]) page yield self.launcher.newPage() yield page.goto(fhttps://{domain}, waitUntilnetworkidle0) yield page.waitForFunction(window._pdd_) self.pool[domain] {page: page, ts: time.time()} # 复用页面生成 anti page self.pool[domain][page] anti yield page.evaluate(() { if (window.__anti__) { return window.__anti__(window._pdd_.data, window._pdd_.config); } return null; }) self.pool[domain][ts] time.time() defer.returnValue(anti)注意yield page.goto()必须指定waitUntilnetworkidle0否则window._pdd_可能未就绪page.evaluate()返回null时需触发重试不能直接设为 header。3.2 商品列表页解析XPath 与 CSS 选择器的混合使用策略拼多多搜索结果页/search_result.html采用瀑布流懒加载初始 HTML 仅含首屏商品后续由fetch(/api/search/items)动态填充。pdd_search.py中的parse_search()方法需处理两种数据源# pdd_search.py def parse_search(self, response): # 1. 解析静态 HTML 中的首屏商品用于快速验证 anti_content 有效性 static_items response.css(div.item-wrap) for item in static_items: yield { title: item.css(div.goods-name::text).get(), price: item.css(span.price::text).re_first(r¥(\d\.\d)), url: response.urljoin(item.css(a::attr(href)).get()), anti_source: static } # 2. 构造 AJAX 请求注入 anti_content search_key response.url.split(search_key)[-1].split()[0] ajax_url fhttps://api.pinduoduo.com/api/search/items?search_key{search_key}page1 yield scrapy.Request( urlajax_url, headers{anti_content: response.request.headers.get(anti_content, b).decode()}, callbackself.parse_ajax_items, meta{search_key: search_key, page: 1} ) def parse_ajax_items(self, response): data json.loads(response.text) for item in data.get(items, []): yield { title: item.get(goods_name), price: item.get(min_group_price) / 100, # 单位分转元 url: fhttps://yangkeduo.com/goods.html?goods_id{item.get(goods_id)}, anti_source: ajax } # 分页逻辑检查是否还有更多 if data.get(has_more): next_page response.meta[page] 1 next_url fhttps://api.pinduoduo.com/api/search/items?search_key{response.meta[search_key]}page{next_page} yield scrapy.Request( urlnext_url, headers{anti_content: response.request.headers.get(anti_content, b).decode()}, callbackself.parse_ajax_items, meta{search_key: response.meta[search_key], page: next_page} )3.2.1 关键参数说明与容错设计response.css(div.item-wrap)匹配静态 HTML 中的商品容器CSS 选择器比 XPath 更简洁适用于结构稳定的首屏response.urljoin()确保相对链接转为绝对 URL避免https://yangkeduo.com//goods.html这类双斜杠错误item.get(min_group_price) / 100拼多多 API 返回价格单位为「分」必须除以 100 转为「元」否则数据失真data.get(has_more)分页判断依据而非len(data[items]) 20因最后一页可能不足 20 条headers{anti_content: ...}显式传递 anti_contentScrapy 默认不继承上一请求的 headers。3.3 商品详情页抓取处理 iframe 嵌套与动态 SKU 数据pdd.py中的parse_item()针对goods.html页面但拼多多详情页大量使用iframe srchttps://mms.pinduoduo.com/...加载 SKU、评论等模块。response.body仅包含 iframe 标签真实数据在 iframe 内。必须提取 iframe URL 并发起二次请求def parse_item(self, response): # 主商品信息DOM 中直接可得 yield { goods_id: response.url.split(goods_id)[-1], title: response.css(h1.goods-title::text).get(), desc: response.css(div.goods-desc::text).get(), sales: response.css(span.sales-count::text).re_first(r(\d)人付款) } # 提取 iframe URLSKU 数据所在 iframe_src response.css(iframe#sku-frame::attr(src)).get() if iframe_src: # 构造带 anti_content 的 iframe 请求 yield scrapy.Request( urliframe_src, headers{anti_content: response.request.headers.get(anti_content, b).decode()}, callbackself.parse_sku, meta{goods_id: response.meta[goods_id]} ) def parse_sku(self, response): # iframe 内部 HTML 结构不同需重新选择 sku_list [] for sku in response.css(div.sku-item): sku_list.append({ name: sku.css(div.sku-name::text).get(), price: sku.css(span.sku-price::text).re_first(r¥(\d\.\d)), stock: sku.css(div.stock::text).get() }) yield { goods_id: response.meta[goods_id], skus: sku_list }3.3.1 iframe 抓取的三大陷阱Referer 校验iframe 请求必须携带Referer: https://yangkeduo.com/goods.html?goods_idxxx否则返回 403需在scrapy.Request中显式设置headers{Referer: response.url}anti_content 复用失效iframe 页面域名mms.pinduoduo.com与主站yangkeduo.com不同AntiPool会为其创建独立页面实例避免混淆动态加载延迟iframe 内容可能由 JS 动态注入response.body初始为空需在parse_sku()中检查response.css(div.sku-item)是否为空为空则触发重试通过RetryMiddleware。4. 并发控制与反爬对抗基于请求队列的动态限速与异常响应分流4.1 请求队列化用 Redis 实现跨进程 anti_content 复用与请求节流run.py启动多个爬虫实例时若每个实例都维护独立 Puppeteer 页面内存开销巨大且 anti_content 生成重复。应将AntiPool升级为分布式服务使用 Redis 存储各域名的anti_content缓存及最后更新时间# pdd_sanic/redis_anti_pool.py import redis import json import time class RedisAntiPool: def __init__(self, hostlocalhost, port6379, db0): self.r redis.Redis(hosthost, portport, dbdb, decode_responsesTrue) def get(self, domain): key fanti:{domain} data self.r.hgetall(key) if data and float(data.get(ts, 0)) time.time() - 60: # 60秒内有效 return data[value] return None def set(self, domain, value): key fanti:{domain} self.r.hset(key, mapping{value: value, ts: str(time.time())}) self.r.expire(key, 300) # 5分钟过期强制刷新 def need_refresh(self, domain): # 检查是否需刷新如连续3次请求失败 fail_key ffail:{domain} count int(self.r.get(fail_key) or 0) if count 3: self.r.delete(fail_key) return True return False4.1.1 Scrapy 中集成 RedisAntiPool在settings.py中启用中间件并注入 Redis 连接# settings.py REDIS_ANTI_POOL { host: localhost, port: 6379, db: 0 } DOWNLOADER_MIDDLEWARES { pdd_sanic.redis_anti_middleware.RedisAntiMiddleware: 543, }# redis_anti_middleware.py from scrapy import signals from pdd_sanic.redis_anti_pool import RedisAntiPool class RedisAntiMiddleware: def __init__(self): self.pool RedisAntiPool(**self.settings.get(REDIS_ANTI_POOL, {})) def process_request(self, request, spider): domain request.url.split(/)[2] anti self.pool.get(domain) if not anti: # 触发异步刷新此处简化为同步生产环境应走 Celery anti self._refresh_anti(domain) self.pool.set(domain, anti) request.headers[anti_content] anti def _refresh_anti(self, domain): # 调用 Puppeteer 服务如 FastAPI 接口生成新 anti import requests resp requests.post(fhttp://localhost:8000/anti/{domain}) return resp.json().get(anti_content)4.2 异常响应分流针对不同 HTTP 状态码的差异化重试策略拼多多反爬响应不止 403还包括412 Precondition Failedanti_content 过期或格式错误需立即刷新该域名的 anti429 Too Many RequestsIP 限频需切换代理或延长间隔503 Service Unavailable服务端过载可快速重试1s 内200 {error:invalid anti}JSON 响应体报错说明 anti_content 生成逻辑变更需人工介入分析。RetryMiddleware需覆盖这些场景# middlewares.py class CustomRetryMiddleware(RetryMiddleware): def process_response(self, request, response, spider): if response.status in [412, 429, 503]: # 412刷新 anti_content 并重试 if response.status 412: domain request.url.split(/)[2] self._refresh_anti(domain) # 429增加 delay if response.status 429: request.meta[retry_times] request.meta.get(retry_times, 0) 1 delay 2 ** request.meta[retry_times] * 0.5 spider.logger.info(f429 detected, retrying in {delay}s) time.sleep(delay) return self._retry(request, spider) or response # 检查 JSON 响应体 if response.headers.get(Content-Type, ).startswith(application/json): try: data json.loads(response.text) if data.get(error) invalid anti: domain request.url.split(/)[2] self._refresh_anti(domain) return self._retry(request, spider) or response except: pass return response4.2.1 重试参数表不同错误类型的应对动作HTTP 状态码响应特征重试动作最大重试次数后续动作412{error:invalid anti}刷新当前域名 anti_content2若仍失败标记域名需人工分析429{error:rate limit}指数退避延迟0.5s, 1s, 2s3触发代理 IP 切换503HTML 或 JSON立即重试无延迟5无需刷新 anti仅等待服务恢复200 error{error:invalid anti}刷新 anti_content1成功则继续失败则记录日志告警5. 生产环境部署技巧Docker 化 Puppeteer 服务与 anti_content 日志审计5.1 Dockerfile 构建轻量 Puppeteer 服务将 Puppeteer 封装为独立 HTTP 服务避免 Scrapy 进程直接管理浏览器提升稳定性# Dockerfile.puppeteer FROM zenika/alpine-chrome:with-puppeteer WORKDIR /app COPY requirements.txt . RUN pip install -r requirements.txt COPY server.py . EXPOSE 8000 CMD [python, server.py]# server.py from flask import Flask, request, jsonify from pyppeteer import launch import asyncio app Flask(__name__) browser None app.before_first_request async def init_browser(): global browser browser await launch(headlessTrue, args[--no-sandbox, --disable-setuid-sandbox]) app.route(/anti/domain) async def get_anti(domain): page await browser.newPage() try: await page.goto(fhttps://{domain}, waitUntilnetworkidle0) await page.waitForFunction(window._pdd_ window.__anti__, timeout10000) anti await page.evaluate(() window.__anti__(window._pdd_.data, window._pdd_.config)) return jsonify({anti_content: anti}) finally: await page.close() if __name__ __main__: app.run(host0.0.0.0:8000)构建命令docker build -f Dockerfile.puppeteer -t pdd-puppeteer .运行命令docker run -p 8000:8000 --shm-size2g pdd-puppeteer提示--shm-size2g是关键Puppeteer 在 headless 模式下需要足够共享内存否则页面加载失败zenika/alpine-chrome镜像比官方镜像小 50%且预装 Chrome 124。5.2 anti_content 日志审计记录生成上下文与请求结果关联在RedisAntiPool.set()中增加审计日志便于追踪问题def set(self, domain, value): key fanti:{domain} # 记录生成时间、IP、User-Agent来自请求头 log_entry { ts: time.time(), value: value, ip: request.remote_addr if request in locals() else unknown, ua: request.headers.get(User-Agent, unknown)[:100], url: request.url if request in locals() else unknown } self.r.lpush(fanti:log:{domain}, json.dumps(log_entry)) self.r.ltrim(fanti:log:{domain}, 0, 99) # 仅保留最近100条 self.r.hset(key, mapping{value: value, ts: str(time.time())}) self.r.expire(key, 300)审计日志可用于当某域名anti_content突然批量失效时检查日志中 UA 是否被统一更换如全部变成curl/7.68.0判断是否遭平台识别对比成功请求与失败请求的ip字段确认是否因 IP 被封禁导致分析url字段发现是否因特定页面路径如/search_result.html触发了更严反爬需单独优化。5.3 关键验证技巧用 curl 模拟请求验证 anti_content 有效性不依赖 Python 环境直接用命令行验证 anti_content 是否可用# 1. 获取 anti_content假设已通过服务生成 ANTI$(curl -s http://localhost:8000/anti/yangkeduo.com | jq -r .anti_content) # 2. 构造搜索请求替换 YOUR_SEARCH_KEY curl -H anti_content: $ANTI \ -H User-Agent: Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 \ https://api.pinduoduo.com/api/search/items?search_key手机page1 | jq .items[0].goods_name # 3. 检查返回若输出商品名如 iPhone 15说明 anti_content 有效若返回 error则需检查 UA、Referer、anti_content 格式。5.3.1 curl 参数说明-H anti_content: $ANTI注入 anti_content注意$ANTI不能含空格或换行-H User-Agent: ...UA 必须与 Puppeteer 启动时一致拼多多严格校验jq .items[0].goods_name解析 JSON提取首个商品名称快速验证数据通路若返回null或error立即检查ANTI变量值是否为null或空字符串这是环境未就绪的明确信号。本文还有配套的精品资源点击获取