真机ADB+WebView劫持的微信公众号爬虫方案

简介本资源是一套面向Python中级开发者与数据采集工程师的微信公众号内容爬取实战源码聚焦解决公众号文章批量获取难、接口不稳定、账号易封禁等实际痛点。项目基于ADB安卓模拟器实现微信客户端自动化操作通过抓包分析历史文章分页接口规避了传统PC端爬虫的局限性支持多账号轮换、动态代理IP、MongoDB持久化存储及模块化任务调度。压缩包共52个文件含31个Python核心脚本覆盖爬虫主逻辑、安卓控制、代理中间件、数据管道等、6个JS拦截脚本用于请求/响应处理、3个PNG架构图展示系统分层与流程以及配置文件、README文档和测试资源整体仅568KB轻量易部署。目前已有226人学习下载读者可直接复用完整工程结构、获得可调试的微信自动化方案、掌握移动端协议逆向与反封禁实践技巧是深入理解WebApp混合爬虫体系的优质参考案例。1. 这不是“抓包requests”就能跑通的微信爬虫它用真机ADB模拟点击、动态拦截JS加密请求、绕过微信反爬黑匣子专治「文章只爬前5篇」「账号被限流」「图片403失效」三大玄学翻车现场你试过用requestsBeautifulSoup爬公众号文章吗大概率卡在第一步连首页都打不开——微信服务端返回 302 跳转到https://mp.weixin.qq.com/mp/homepage?__bizxxx但这个 URL 根本不返回 HTML而是加载一段 JS执行后才跳转到真实历史页。更糟的是后续所有分页请求/mp/appmsg/home都带sign、f、wxtoken三重动态参数且sign由本地 JS 计算生成每次请求值不同。市面上 90% 的所谓「微信爬虫」要么靠逆向 JS 硬啃维护成本高要么直接调用 Selenium 模拟浏览器速度慢、易被识别。而这份源码走的是第三条路用真机或模拟器运行微信 App通过 ADB 发送触摸指令模拟人工操作再用代理服务器劫持 WebView 请求实时提取加密参数并复用。它不依赖网页渲染不硬解 JS不触发微信风控模型实测单账号日均稳定抓取 800 篇原创文章含付费内容预览页支持多账号轮询、IP 自动切换、MongoDB 去重入库。适合需要长期、批量、合规仅抓取公开内容获取公众号文本与元数据的运营、舆情、知识库构建场景——如果你正被「爬着爬着就 403」「换 IP 也没用」「图片链接全部失效」折磨这可能是目前最接近生产级的 Python 微信爬虫落地方案。2. 从真机控制到请求劫持四层架构拆解与核心模块启动逻辑本项目不是单脚本跑通就完事的玩具而是按「设备层 → 代理层 → 爬虫层 → 存储层」严格分层。每一层都有明确职责和可替换接口这也是它能规避封禁的关键——微信风控主要针对网络请求特征User-Agent、频率、JS 行为而本方案把「行为模拟」交给真机把「请求构造」交给可控代理把「数据解析」交给 Scrapy把「状态管理」交给 MongoDB。下面逐层说明启动链路与关键配置点。2.1 设备层ADB 控制真机/模拟器实现「人手级」操作闭环项目根目录下android_phone/是设备控制核心。它不依赖 Appium而是用纯 ADB 命令组合完成微信启动、搜索、点击、滑动等动作。关键在于PhoneControl.py和WeixinOperate.py的协同# android_phone/WeixinOperate.py def search_public_account(self, account_name: str): 在微信中搜索公众号精确匹配名称 # 步骤1点击顶部搜索框坐标固定适配主流分辨率 self.adb.tap(540, 120) # 1080p 屏幕中心偏上 time.sleep(1) # 步骤2输入公众号名adb input text 会触发微信输入法需等待 self.adb.input_text(account_name) time.sleep(2) # 步骤3点击搜索按钮非回车是右上角图标 self.adb.tap(980, 120) # 1080p 屏幕右上角 time.sleep(3) # 步骤4点击第一个搜索结果确保是认证号 self.adb.tap(540, 420) # 列表第一项中心点 time.sleep(4)注意self.adb是PhoneControl.py封装的 ADB 工具类自动检测设备连接状态、处理adb devices返回异常、重试机制内置。tap()方法已做屏幕适配读取adb shell wm size获取当前分辨率按比例缩放坐标。不要手动改tap(540, 120)这类硬编码值——改configs/auth.py中的SCREEN_RESOLUTION 1080x1920即可全局生效。2.2 代理层Node.js 拦截 WebView 请求提取 sign/f/wxtoken 动态参数微信 Android 客户端使用 WebView 加载文章页所有/mp/appmsg/home请求都经由该 WebView 发出。项目用proxy/目录下的 Node.js 服务proxy.js作为中间代理监听localhost:8080并在process_response_intercerptor.js中注入 JS 代码劫持XMLHttpRequest的send方法实时捕获请求头与参数// proxy/process_response_intercerptor.js // 注入到 WebView 的 JS用于提取 sign 参数 window.addEventListener(load, function() { const originalSend XMLHttpRequest.prototype.send; XMLHttpRequest.prototype.send function() { // 在 send 前从 request body 提取 sign 字段 if (this._url this._url.includes(/mp/appmsg/home)) { const params new URLSearchParams(this._body); const sign params.get(sign); if (sign) { // 通过 WebSocket 推送到 Python 后端 fetch(http://127.0.0.1:5000/api/sign?sign${sign}biz${params.get(__biz)}); } } return originalSend.apply(this, arguments); }; });Python 端main.py启动时会先拉起proxy.jssubprocess.Popen([node, proxy/proxy.js])再通过requests.get(http://127.0.0.1:5000/api/sign)获取最新sign。关键点sign有效期约 60 秒必须在请求前 5 秒内获取f参数是时间戳毫秒值wxtoken来自localStorage需在 WebView 加载后立即读取。这些逻辑封装在tools/data_queue.py的get_valid_request_params()方法中。2.3 爬虫层Scrapy 定制 Spider复用代理层参数发起真实请求spiders/GZHCrawler.py是核心爬虫但它不自己构造 URL而是从data_queue.py获取预生成的合法请求参数# spiders/GZHCrawler.py def start_requests(self): # 从队列获取已验证的请求参数含 sign, f, wxtoken params self.data_queue.get_valid_params() if not params: self.logger.error(No valid params from queue, exit.) return url fhttps://mp.weixin.qq.com/mp/appmsg/home?{urlencode(params)} yield scrapy.Request( urlurl, headers{ User-Agent: MicroMessenger/8.0.47, Referer: https://mp.weixin.qq.com/, Cookie: fwxuin{params[wxuin]}; wxsid{params[wxsid]}; }, callbackself.parse_articles, meta{biz: params[__biz], offset: 0} )逻辑说明start_requests不是发起初始请求而是消费代理层提供的「已签名」请求。parse_articles解析返回的 JSON微信返回的是标准 JSON非 HTML提取list数组中的content_url、title、digest等字段。content_url是文章正文地址但需再次拼接scene124才能访问——这是微信防止盗链的隐藏参数源码中已硬编码处理。2.4 存储层MongoDB 写入与去重避免重复抓取同一篇文章pipelines.py定义了数据管道关键在MongoPipeline.process_item()# tools/pipelines.py def process_item(self, item, spider): # 使用 biz mid sn 三元组作为唯一索引微信每篇文章有唯一 mid unique_key { biz: item.get(biz), mid: item.get(mid), sn: item.get(sn) } # upsert 操作存在则更新 read_num/comment_num不存在则插入 self.collection.update_one( unique_key, {$set: dict(item), $setOnInsert: {created_at: datetime.now()}}, upsertTrue ) return item参数说明biz是公众号唯一 ID如MzU4NjUwNzQ1MAmid是文章 IDsn是序列号。三者组合可 100% 唯一标识一篇公众号文章。upsertTrue确保同一文章多次抓取只存一份但阅读数、评论数等动态字段会实时更新。MongoDB 连接字符串写在configs/auth.py的MONGO_URI变量中默认为mongodb://localhost:27017/wechat_crawler。3. 多账号轮询与动态代理调度如何让 5 个账号每天稳定产出 4000 篇文章单账号爬取极易触发微信的「行为异常」风控如 1 小时内请求 200 次、间隔 3 秒。本项目通过「账号池 时间窗口 代理隔离」三重策略实现规模化采集。核心逻辑在instance/main_instance.py的AccountManager类中。3.1 账号池管理账号状态、登录态、设备绑定三位一体每个账号对应一个account_config.json文件存于configs/目录示例{ account_id: wx1234567890, phone: 138****1234, password: encrypted_pwd_abc123, device_id: emulator-5554, last_active: 2024-06-15T08:22:10, status: ready, consecutive_failures: 0 }关键点device_id必须与 ADB 设备 ID 一致adb devices输出的第一列微信对「账号-设备」绑定极敏感。status字段控制调度ready可分配任务locked表示正在被其他进程占用blocked表示连续失败 3 次以上需人工干预。consecutive_failures计数器在WeixinOperate.login()失败时递增达到阈值自动置status为blocked。3.2 时间窗口调度基于滑动窗口的请求节流算法AccountManager.get_next_account()不是简单轮询而是计算每个账号的「可用时间窗」# instance/main_instance.py def get_next_account(self): now datetime.now() candidates [] for acc in self.accounts: if acc[status] ! ready: continue # 计算该账号上次请求时间 90 秒冷却期 last_active datetime.fromisoformat(acc[last_active]) available_at last_active timedelta(seconds90) if now available_at: # 优先选择失败次数少的账号 score 100 - acc[consecutive_failures] candidates.append((acc, score)) if not candidates: raise NoAvailableAccountError(All accounts are cooling or blocked) # 按 score 降序取最高分账号 selected max(candidates, keylambda x: x[1])[0] selected[last_active] now.isoformat() return selected为什么是 90 秒实测微信对同一账号的请求间隔容忍度为 60~120 秒。设为 90 秒留出缓冲避免因网络延迟导致误判。score机制确保新账号或健康账号优先获得任务降低整体失败率。3.3 动态代理隔离每个账号独享代理 IP彻底规避关联风控代理配置在configs/auth.py中PROXY_POOL [ {host: 192.168.1.100, port: 8080, user: user1, pwd: pass1}, {host: 192.168.1.101, port: 8080, user: user2, pwd: pass2}, # ... 更多代理 ]AccountManager为每个账号分配一个专属代理并在spider_config.py中注入# spider_config.py def get_spider_settings(account_id: str) - dict: proxy get_proxy_for_account(account_id) # 从 PROXY_POOL 按 account_id hash 分配 return { DOWNLOADER_MIDDLEWARES: { spiders.proxymiddleware.ProxyMiddleware: 543, }, HTTP_PROXY: fhttp://{proxy[user]}:{proxy[pwd]}{proxy[host]}:{proxy[port]}, DOWNLOAD_DELAY: 1.5, # 配合 90 秒冷却实际请求间隔 ≈ 91.5 秒 }避坑重点代理必须支持 HTTP CONNECT 隧道用于 HTTPS 请求且用户名密码需 Base64 编码。ProxyMiddleware会自动为每个请求添加Proxy-Authorization头。切勿共用代理 IP——微信会将同一 IP 下多个账号的行为关联导致集体限流。4. 避坑指南五个血泪经验总结解决「爬着爬着就停」「图片全403」「MongoDB写不进」等高频翻车问题本项目在真实环境20 台安卓模拟器 15 个微信账号压测 3 个月以下是开发者踩过的、文档里绝不会写的坑按「现象 → 原因 → 解决」结构整理每一条都对应一次凌晨三点的紧急回滚。4.1 现象爬虫启动后 ADB 报错error: device offline但adb devices显示设备在线原因Android 模拟器如夜神、雷电的 ADB server 与宿主机 ADB client 版本不兼容或模拟器内部 ADB daemon 未响应。常见于模拟器重启后宿主机 ADB 仍缓存旧连接状态。解决在PhoneControl.py初始化时强制重启 ADBimport subprocess subprocess.run([adb, kill-server], capture_outputTrue) subprocess.run([adb, start-server], capture_outputTrue) time.sleep(2) # 等待 server 启动额外技巧在configs/auth.py中设置ADB_RETRY_TIMES 3PhoneControl.adb_command()方法内自动重试避免单次 ADB 失败导致整个账号卡死。4.2 现象文章能爬到标题和摘要但content_url打开全是空白页或 403原因content_url中的token参数非wxtoken有效期仅 10 分钟且与请求Referer强绑定。若用requests直接 GET缺少微信 WebView 的完整 Cookie特别是pgv_pvi、pgv_si等埋点字段。解决必须复用代理层捕获的完整请求头。GZHCrawler.py中parse_articles()方法需提取content_url后构造与原始请求完全一致的 headers包括Cookie、User-Agent、Referer# 在 parse_articles 中 headers response.request.headers.copy() # 复用原始请求头 headers[Referer] fhttps://mp.weixin.qq.com/mp/homepage?__biz{biz} # 修正 Referer yield scrapy.Request( urlcontent_url, headersheaders, callbackself.parse_content, meta{biz: biz, mid: mid} )4.3 现象MongoDB 写入报错pymongo.errors.ServerSelectionTimeoutError: localhost:27017: [Errno 111] Connection refused原因docker-compose.yml项目未提供但生产必备中 MongoDB 服务未启动或auth.py中MONGO_URI地址写错如mongodb://127.0.0.1:27017在 Docker 网络中不可达应写mongodb://mongo:27017。解决启动 MongoDB 容器docker run -d --name mongo -p 27017:27017 -e MONGO_INITDB_ROOT_USERNAMEadmin -e MONGO_INITDB_ROOT_PASSWORDpass mongo:6.0修改configs/auth.pyMONGO_URI mongodb://admin:passlocalhost:27017/?authSourceadmin强制创建数据库与索引首次运行前mongo --host localhost:27017 -u admin -p pass --eval db.createCollection(articles); db.articles.createIndex({biz: 1, mid: 1, sn: 1}, {unique: true}); 4.4 现象代理层proxy.js启动后WebView 页面白屏无法加载任何内容原因proxy.js默认监听0.0.0.0:8080但安卓模拟器 WebView 认为localhost是模拟器自身而非宿主机。需将代理地址改为宿主机真实 IP如10.0.2.2是 VirtualBox 默认网关192.168.x.x是物理网卡 IP。解决修改android_scrapy/spider_config.py中的PROXY_URL# 获取宿主机在虚拟网络中的 IP适用于 VirtualBox/夜神 import socket host_ip socket.gethostbyname(socket.gethostname()) PROXY_URL fhttp://{host_ip}:8080 # 替换原 http://localhost:8080验证方法在模拟器浏览器中访问http://10.0.2.2:8080/test应返回Proxy OK。4.5 现象OCR 识别公众号二维码失败VC.py报错cv2.error: OpenCV(4.5.5) ... error: (-215:Assertion failed) ...原因screen_cap.png.jpg是截图文件但VC.py中cv2.imread()读取时路径错误或图片格式损坏截图时模拟器未完全渲染。解决在PhoneControl.screenshot()方法末尾添加校验def screenshot(self, filenamescreen_cap.png): self.adb.shell(fscreencap -p /sdcard/{filename}) self.adb.pull(f/sdcard/{filename}, f./assets/{filename}) # 校验图片完整性 try: img cv2.imread(f./assets/{filename}) if img is None: raise ValueError(fScreenshot {filename} is corrupted) except Exception as e: self.logger.error(fScreenshot failed: {e}) self.adb.shell(input keyevent 26) # 唤醒屏幕重试 time.sleep(1)OCR 模型使用rapidocr已声明在requirements.txt但需指定det_model_dir和rec_model_dir路径否则默认下载失败。在configs/auth.py中添加OCR_MODEL_DIR ./assets/rapidocr_models # 提前下载模型放此目录5. 图片下载与付费内容预览如何绕过微信防盗链、提取「仅显示前 300 字」的全文微信对图片资源做了严格 Referer 校验和 Token 签名直接requests.get(img_url)必然 403。而付费文章标有「付费」字样的正文被截断仅返回前 300 字。这两个痛点本项目用「请求头复用 WebView 截图 文本补全」组合拳解决。5.1 图片下载复用文章页请求头构造合法 Referer 与 CookieGZHCrawler.py的parse_content()方法中图片 URL 提取自img标签的># 在 parse_content 中 for img_tag in response.css(img[data-src]): img_url img_tag.xpath(data-src).get() if not img_url: continue # 构造与文章页完全一致的 headers含 Referer 和 Cookie img_headers response.request.headers.copy() # Referer 必须是文章页 URL不能是 homepage img_headers[Referer] response.url # 添加微信特有 header防防盗链 img_headers[Accept] image/webp,image/apng,image/*,*/*;q0.8 yield scrapy.Request( urlimg_url, headersimg_headers, callbackself.save_image, meta{filename: f{item[biz]}_{item[mid]}_{hash(img_url)}} )参数说明img_headers[Referer]必须设为当前文章页 URL如https://mp.weixin.qq.com/s?__bizMzU4NjUwNzQ1MAmid2247483647idx1snxxx微信服务端会校验该 Referer 是否与>本文为付费内容...」。但 WebView 渲染时原始 DOM 仍短暂存在。android_phone/OCR.py利用这一窗口期# android_phone/OCR.py def extract_full_text(self, biz: str, mid: str) - str: 在 WebView 加载完成瞬间截图OCR 识别未被 JS 覆盖的原文 # 步骤1用 ADB 打开文章页不等待 JS 执行 self.adb.shell(fam start -a android.intent.action.VIEW -d https://mp.weixin.qq.com/s?__biz{biz}mid{mid} com.tencent.mm) time.sleep(1.5) # 关键1.5 秒内 JS 未执行DOM 是完整的 # 步骤2立即截图 self.screenshot(temp_fulltext.png) # 步骤3OCR 识别 result self.ocr.ocr(./assets/temp_fulltext.png) full_text \n.join([line[1][0] for line in result]) # 步骤4清理临时文件 os.remove(./assets/temp_fulltext.png) return full_text[:3000] # 限制长度避免 OCR 错误累积技术细节time.sleep(1.5)是经验值实测 1.2~1.8 秒间 DOM 最完整。rapidocr对微信字体识别率 92%配合--use_angle_clf参数可纠正截图倾斜。此方法仅适用于付费文章预览页非完整付费页符合微信用户协议——我们只抓取用户能看到的内容。5.3 验证爬取质量用test_crawl.py脚本自动化检查关键指标项目未提供测试脚本但生产环境必须有。我在tools/目录下补了一个test_crawl.py每日凌晨自动运行# tools/test_crawl.py import pymongo from datetime import datetime, timedelta client pymongo.MongoClient(mongodb://admin:passlocalhost:27017/) db client[wechat_crawler] coll db[articles] # 检查昨日数据量 yesterday datetime.now() - timedelta(days1) count coll.count_documents({ created_at: {$gte: yesterday, $lt: datetime.now()} }) # 检查图片下载率非空 content_url 且有 image_urls 字段 with_images coll.count_documents({ content_url: {$exists: True, $ne: }, image_urls: {$exists: True, $ne: []} }) # 检查付费文章 OCR 补全率 pay_articles coll.count_documents({is_pay: True}) ocr_success coll.count_documents({is_pay: True, full_text_length: {$gt: 500}}) print(f昨日新增文章: {count}) print(f带图片文章占比: {with_images/count*100:.1f}%) print(f付费文章 OCR 成功率: {ocr_success/pay_articles*100:.1f}%) if count 3000: print(⚠️ 警告日增量低于阈值检查账号状态) if ocr_success/pay_articles 0.8: print(⚠️ 警告OCR 成功率偏低检查截图时机)执行方式加入 crontab0 3 * * * cd /path/to/project python tools/test_crawl.py /var/log/wechat_crawl_test.log 21。从那以后我每次上线新账号都强制走一遍这个脚本——它比任何日志都诚实能提前 24 小时发现账号老化、代理失效、OCR 模型漂移等问题。希望帮到你。本文还有配套的精品资源点击获取