尧图网站设计 尧图网站设计YAOTU DESIGN
ARTICLE DETAIL

资讯详情

深耕网站设计与一线实操的经验洞察。

微信公众号历史文章爬虫实战:接口原理与代理池并发设计

微信公众号历史文章爬虫实战:接口原理与代理池并发设计 简介面向需要批量获取微信公众号历史文章的开发者这份资源提供了一套基于Go语言实现的爬虫方案。核心亮点在于仅需配置代理即可一键抓取全部历史文章省去繁琐的登录与防反爬处理适合有一定编程基础、希望快速搭建公众号数据采集工具的读者参考。压缩包共6个文件包含4个Go源文件、1个说明文档和1个gitignore配置整体仅4KB轻量易读。Go源文件分别承担处理器、输出服务、简单服务等模块README可帮助快速理解运行流程适合直接改造成自有采集任务。目前已有116人学习下载对于想了解Go爬虫工程结构或复用现成采集逻辑的开发者而言是一份简洁实用的参考资料。1. 微信公众号爬虫为什么需要代理先弄清文章数据从哪里来一个面向微信公众号历史文章的爬虫为什么非要强调“只需设置代理”最常见的误解是代理能解决“能不能访问”但实际上它解决的是“被限制之后能不能换条路继续”。微信公众号文章不像普通网页那样有清晰的目录页它依赖一组带 offset 和 count 参数的 JSON 接口在微信客户端里才能稳定请求。无论你用 requests 爬虫、Scrapy 还是封装好的 zip 工具包第一件事不是填代理而是定位到/mp/profile_ext这个历史消息接口拿到__biz、offset、count这几个核心参数。等你把请求头、会话、超时和翻页逻辑都理顺之后代理只是一个可插拔的选项用来降低 IP 请求频次过高带来的封禁风险。这篇文章会从接口原理讲起逐步落到可运行的 Python 代码并给出并发和代理池的具体设计。2. 从一个公众号主页到第一份文章列表抓包与 URL 还原2.1 用开发者工具定位历史文章的接口在 PC 版微信里打开任意公众号的“查看历史消息”按 F12 打开开发者工具切到 Network 面板再往下滚动消息列表就能看到名为/mp/profile_ext的请求。它的 URL 类型如下https://mp.weixin.qq.com/mp/profile_ext?actiongetmsg__bizMzA3ODUy...fjsonoffset10count10is_ok1scene124这里的__biz是公众号身份标识offset用于翻页count控制单次返回条数actiongetmsg表示拉取文章列表。把这次请求复制成 curl 命令在终端执行一遍能拿到 JSON 再转成 Python 代码是最常见的稳定起步方式。我习惯先验证 curl 能返回数据再写 requests 脚本这样能快速排查是代理问题还是请求头问题。提示以下内容仅用于学习 HTTP 抓包和 Python 爬虫技术请勿用于违反微信公众平台规则的大规模下载。2.2 用 Python 复现第一次请求把 curl 转换成 Python 时主要关注两个地方请求头headers和查询参数params。代码里要显式设置User-Agent和Referer否则微信服务端会拒绝请求。import requests biz MzA3ODUy... # 从历史页 URL 里复制的 __biz headers { User-Agent: Mozilla/5.0 (Windows NT 10.0; Win64; x64) Chrome/120.0.0.0 Safari/537.36, Referer: fhttps://mp.weixin.qq.com/mp/profile_ext?actionhome__biz{biz}scene124, } params { __biz: biz, action: getmsg, f: json, offset: 0, count: 10, is_ok: 1, scene: 124, uin: 777, key: , pass_ticket: , wxtoken: , appmsg_token: , x5: 0, } url https://mp.weixin.qq.com/mp/profile_ext resp requests.get(url, paramsparams, headersheaders, timeout10) print(resp.status_code)用params传查询参数的好处是 requests 会自动进行 URL 编码避免biz里的、/、被误解。uin、key、pass_ticket通常直接填空字符串不需要登录态。如果返回的 JSON 里ret为 200003说明被频率限制了可以降低count或更换代理 IP 后重试。2.3 参数表哪些字段决定成败参数名含义抓取历史文章时的建议值__biz公众号唯一标识必填从历史页 URL 提取action操作类型getmsgoffset文章偏移量从 0 开始每页加 10count本次返回条数10 到 20建议取 10f返回数据格式jsonscene场景编号124 或 126appmsg_token某些场景下的临时 token多数情况可留空uin/key用户标识和票据非登录态填空字符串offset不是文章序号而是游标。微信接口最多返回约一个微信版本窗口内的数据翻到尾部时can_msg_continue会变成 0所以循环终止条件要按返回字段判断不能盲目设定固定次数。3. 用代理和并发设计把抓取速度拉高同时不被限流3.1 代理在这个爬虫里到底解决什么当脚本连续抓取几百页时微信服务端会按照 IP 维度统计请求频率达到阈值后开始要求验证码或返回空数据。代理的本质是让每次请求走一个不同的出口 IP把压力分散到多个地址上。这里说的代理指通用 HTTP/HTTPS 代理在 requests 中通过proxies字典传入即可不涉及任何特殊链路。代理池的实现不必复杂一个列表就能跑起来proxies_list [ {http: http://user:pass123.45.67.89:8080, https: http://user:pass123.45.67.89:8080}, {http: http://user:pass98.76.54.32:3128, https: http://user:pass98.76.54.32:3128}, ]使用前先验证连通性用一个代理连续请求三次公众号接口检查返回的数据里是否包含目标__biz以及是否频繁出现“访问过于频繁”的提示。有些代理本身响应很快但它出口的 IP 已经被微信列入风控名单会导致请求返回验证页这类代理要立刻从池子里移掉。3.2 Session 与连接池为什么不直接 requests.get把requests.get直接放在循环里每次都会建立新的 TCP 连接抓 1000 篇文章意味着要做上千次 TLS 握手耗时和资源消耗都很高。更推荐用requests.Session()保持连接让底层连接池复用同时在 Session 层面统一设置请求头代码也会更干净。import requests from random import choice session requests.Session() session.headers.update(headers) # 复用前面定义的请求头字典 def fetch_one(offset, count10): params[offset] offset params[count] count proxies choice(proxies_list) if proxies_list else None resp session.get(url, paramsparams, headersheaders, proxiesproxies, timeout10) return resp.json()这里需要注意session.headers.update()设置了全局请求头但如果每次get()又传一次headers临时 headers 会覆盖全局值二者不要混成两套。代理参数可以放在session.proxies.update()里全局生效也可以临时传参数。只有个别代理失效时临时传参会更方便控制。连接池大小由 requests 底层的 urllib3 自动管理线程数较多时可以通过HTTPAdapter的pool_maxsize参数调整避免出现连接池满的告警。3.3 ThreadPoolExecutor 并发到底设多少个线程合适微信公众号历史接口的瓶颈是网络延迟所以用多线程能明显提速。但线程数不是越大越好微信对同一 IP 的接口频率限制约在每秒 5 到 10 次超过后便开始要求验证码。实际操作中我会把线程数控制在 3 到 5并在每个请求之间加入随机 sleep让请求节奏更接近真人操作。from concurrent.futures import ThreadPoolExecutor, as_completed import time, random def worker(offset): time.sleep(random.uniform(0.5, 1.0)) data fetch_one(offset) return offset, data offset_list list(range(0, 100, 10)) # 前 100 条文章 with ThreadPoolExecutor(max_workers4) as executor: futures [executor.submit(worker, off) for off in offset_list] for future in as_completed(futures): off, data future.result() # 这里把 data 交给解析函数 passmax_workers4是单代理 IP 下的安全值。如果你有大量高匿代理可以提高到 8但建议不要超过 10。线程越多遇到代理失效时越难排查是哪一个请求出的问题而且多个线程共用同一个 Session 时输出顺序会乱不利于日志归因。并发设计的目标是“可控”而不是“极快”先稳定跑通再逐渐调高线程数量。4. 历史文章翻页与去重从“一页”到“全部”4.1 解析 general_msg_list 里的文章数组接口返回的 JSON 里有个特殊结构general_msg_list本身是一个被编码成字符串的 JSON。先json.loads()解码再遍历list字段里面的每个元素就是一组文章消息。每组消息的首条图文在app_msg_ext_info里发布时间在comm_msg_info里。import json def parse_msg_list(raw_json): msg_list json.loads(raw_json[general_msg_list]) for msg in msg_list[list]: article msg[app_msg_ext_info] item { title: article[title], digest: article[digest], link: article[content_url].replace(amp;, ), publish_time: msg[comm_msg_info][datetime], } # 多图文时还需要展开 article[multi_item] yield item这里最容易踩的坑是content_url里的amp;没有被自动转义。直接拿这个链接请求文章详情页参数会丢失导致页面返回错误。所以替换amp;为是必须步骤。另外公众号经常发多图文multi_item数组里的子文章也包含独立标题和链接只取首条会漏掉大量内容。4.2 用 can_msg_continue 控制终止条件历史接口的翻页由offset和count控制但到底有没有下一页要检查返回 JSON 中的can_msg_continue字段。当它变成 0 时表示已经没有更多文章。offset 0 results [] while True: data fetch_one(offset) can_continue data.get(can_msg_continue, 0) for item in parse_msg_list(data): results.append(item) if not can_continue: break offset 10这个循环里每次offset加 10与count保持一致。假如某次请求因为代理失效返回的是错误结构can_msg_continue取不到值会陷入死循环。改进方法是加一个失败计数器连续失败 3 次就退出并把已抓取数据保存下来避免白跑。4.3 增量抓取怎么做到只抓新文章已经抓过的链接不需要重复下载否则每天跑一次会浪费大量时间和代理流量。常见做法是把文章链接存到本地 CSV 或 SQLite 中启动爬虫时先加载已见集合遇到重复链接直接跳过。import os, csv seen set() if os.path.exists(articles.csv): with open(articles.csv, r, encodingutf-8) as f: for row in csv.DictReader(f): seen.add(row[link]) def is_duplicate(item): if item[link] in seen: return True seen.add(item[link]) return False去重判断要放在解析之后、请求详情页之前。文章链接在整个微信生态里是全局唯一的比标题更可靠。即使公众号重新推送同一篇内容URL 中的参数也会变化所以用链接做键可以保证不误判。4.4 反爬检查看不到错误码时看返回内容微信接口的错误码比较明确ret 0成功ret 200003触发频率限制ret 200013参数错误。但最迷惑人的是请求成功却返回空数组或验证页。这时要在fetch_one里打印返回文本前 200 个字符检查是否有“访问过于频繁”“请长按识别二维码”等提示。如果发现这种静默拦截优先停下来等待 30 秒再更换代理重跑而不是盲目调大并发。5. 把脚本封装成“一键”运行输入公众号 ID输出全部文章5.1 命令行参数与配置文件一个 zip 包交付给同事或客户时对方不会愿意翻源码改参数。合理的做法是把__biz、代理地址、并发数、保存路径全放进config.ini脚本入口只接收必要的命令行参数。python wechat_crawler.py --config config.ini --biz MzA3... --offset 0import configparser, argparse def load_args(): parser argparse.ArgumentParser() parser.add_argument(--config, defaultconfig.ini) parser.add_argument(--biz, defaultNone) parser.add_argument(--offset, typeint, default0) return parser.parse_args()--biz不传时从配置文件读取--offset让你可以接着上次的断点继续抓。这样设计更符合“一键”的使用习惯也让 zip 包里的脚本可以重复运行不需要每次改代码和重启。5.2 数据落地CSV 和 JSON 双写抓取结果不能只存在内存里否则程序中断一次就全丢。我会每解析成功一页就追加写入 CSV文件编码用utf-8-sig解决 Excel 打开乱码的问题。同时把所有数据保存为 JSON方便后续用 pandas 分析。def append_to_csv(item, patharticles.csv): file_exists os.path.exists(path) with open(path, a, newline, encodingutf-8-sig) as f: writer csv.DictWriter(f, fieldnames[title, digest, link, publish_time]) if not file_exists: writer.writeheader() writer.writerow(item)每次追加一条就立刻 flush虽然磁盘写入次数变多但能保证断电或者断网时不丢失进度。全部抓完之后再把整个results列表 dump 成 JSONCSV 用于人工查看JSON 用于程序和后续处理双保险更省事。5.3 用日志与进度计数验证是否抓全最后分享一个检查抓取是否完整的技巧在公众号主页底部通常会显示历史文章总数。抓取结束后把日志里的total与页面显示的数量对比。如果差距大优先检查multi_item子图文是否解析完整以及翻页终止条件是否过早触发。把每一页的 offset 和返回条数打印到日志里能很快定位到断掉的那一页。total len(results) unique_total len({item[link] for item in results}) print(fdone, total{total}, unique{unique_total})建议拿到任何 zip 包之后先用一个小型测试公众号跑通流程再替换成真正的目标。运行过程中盯住代理池的失败率如果单次请求失败率超过 10%就先降低线程数到 2等代理池刷新后再调回 4。这样既能保证“一键”顺利跑完也能让最终拿到的历史文章列表完整可复用。本文还有配套的精品资源点击获取
返回列表