Python 微信公众号数据采集实战：找号、查号、取文、追热点一次跑通

Python 微信公众号数据采集实战找号、查号、取文、追热点一次跑通【免费下载链接】WechatSogou基于搜狗微信搜索的微信公众号爬虫接口项目地址: https://gitcode.com/gh_mirrors/we/WechatSogou曾经接过一个市场调研的小需求把 30 个竞品公众号最近一个月的推送整理成表格。你猜我怎么做的打开每个公众号一篇篇复制标题、摘录正文再手动贴进 Excel——两天下来腰酸背痛还因为漏了两篇被领导打回。后来同事甩过来一句这事不是有现成的微信公众号数据采集工具吗没错今天要聊的 WechatSogou就是这样一个把搜狗微信搜索封装成 Python 接口的公众号数据采集方案。它不解决所有问题但足够让按关键词找公众号、查档案、抓历史文章、跟热门话题这几件高频杂活从两小时压缩到两分钟。文章会按一条完整的业务链路带你走一遍读完你就能写出自己的第一个可用脚本。先判断这个工具到底适不适合你动手之前先把丑话说在前面。任何爬虫接口都有边界先花 30 秒看这张速览表比什么都重要。能力面具体表现能做什么按关键词搜公众号、查单个公众号档案、搜索公众号文章、拉最近 10 条历史推送、按分类看热门文章、取文章正文、拿搜索联想词适合谁需要批量维护公众号名单的产品/运营/分析师做内容监控或竞品研究的开发者不适合谁需要某个号全部历史文章、需要大规模高频采集、需要实时更新的生产级数据管道硬性边界单号仅能拿到最近 10 条群发文章临时链接有时效过期就得重新抓环境要求Python 2.7 / 3.5一条pip install就能装这部分解决你的问题用 30 秒判断它是否值得你继续读下去。如果上面表格里能做什么恰好命中你的需求下面就直接上手。三分钟跑通第一个采集任务先说结论这个库的接入成本低到不需要写爬虫框架核心就是一个WechatSogouAPI对象。pip install wechatsogou --upgrade装完先做一次冒烟测试验证网络和接口都正常import wechatsogou # 创建 API 实例代理、超时等 requests 参数都能直接传进去 ws wechatsogou.WechatSogouAPI() # 搜一下和数据相关的公众号看看能不能连通 for gzh in ws.search_gzh(数据): print(gzh[wechat_name], gzh.get(wechat_id)) break # 先只打一条确认能通就行能打印出公众号名称说明链路已经打通。注意search_gzh返回的是生成器所以示例里用了for循环取数——这也是它和普通函数在写法上唯一的区别。业务链路一怎么找号把相关公众号一网打尽假设你现在要建一份数据分析领域的公众号名单。手动搜搜狗搜索页一页页翻都不是好办法。用search_gzh按关键词检索支持分页每页返回 10 个把前几页拉完基本就是一份初筛名单。# 第一页拿不到就翻页攒够 30 个候选 candidates [] for page in range(1, 4): for gzh in ws.search_gzh(数据分析, pagepage): candidates.append({ 名称: gzh[wechat_name], 微信号: gzh.get(wechat_id), 简介: gzh.get(introduction), 认证: gzh.get(authentication, 未认证), }) print(f共收集到 {len(candidates)} 个候选公众号)这部分解决你的问题批量发现目标公众号把手动搜索 复制粘贴变成一次循环。search_gzh的实现位于源码的wechatsogou/api.py数据解析逻辑在wechatsogou/structuring.py感兴趣可以去看它对搜索结果页做了哪些字段清洗。业务链路二怎么查号建一份可信的公众号档案光有名字不够还要判断它是不是正经号——认证主体是什么、近一个月发了多少条、简介靠不靠谱。这时用get_gzh_info传入微信号或名称即可返回单个公众号的完整档案。info ws.get_gzh_info(南航青年志愿者) print(f名称: {info[wechat_name]}) print(f认证: {info.get(authentication, 无)}) print(f近一月群发: {info.get(post_perm, 未知)}) print(f简介: {info.get(introduction, )})这部分解决你的问题在建号阶段就把是否值得跟进判断掉避免把营销号、僵尸号混进名单。建议你在建表时把authentication、post_perm这两个字段一起存下来后续做数据清洗会省很多事。业务链路三怎么取文内容抓取的三层递进到了最核心的环节——取文章。这里有三层做法从浅到深按需选用。第一层跨号搜主题用search_article不需要知道是哪个公众号发的直接按主题搜。还支持时间范围和内容类型筛选比如最近一周的原创图文from wechatsogou import WechatSogouConst articles ws.search_article( 人工智能, timesnWechatSogouConst.search_article_time.week, article_typeWechatSogouConst.search_article_type.all, ) for a in articles[:5]: print(a[article][title], a[article].get(abstract, )[:40])WechatSogouConst里预置了一整套常量时间维度有anytime / day / week / month / year内容类型有all / rich / video / image定义在wechatsogou/const.py用到时直接查。第二层锁定单个号用get_gzh_article_by_history拉最近推送监控某个具体竞品时用这个接口。它返回公众号基本信息和最近 10 条群发文章标题、摘要、发布时间、封面、正文链接都在里面。data ws.get_gzh_article_by_history(南航青年志愿者) print(f公众号: {data[gzh][wechat_name]}) print(f文章数: {len(data[article])}) for art in data[article][:3]: print(art[title], art.get(datetime))第三层及时存正文用get_article_content拿到临时链接后别只存链接——微信文章链接会过期。要长期保存就趁有效期内把正文抓下来# 从上面的 article 列表里取第一条的链接 url data[article][0][content_url] html_text, img_list ws.get_article_content(url) # html_text 是清洗后的正文 HTMLimg_list 是文中图片地址列表 with open(saved_article.html, w, encodingutf-8) as f: f.write(html_text)这部分解决你的问题覆盖了搜主题 → 盯单号 → 存全文三个层次的内容获取需求。需要提醒的是正文接口如果遇到链接已过期库会直接抛异常所以拿到链接后尽快落库是最佳实践。业务链路四怎么追热点选题和关键词两手抓内容运营最怕没选题。这一节给你两个现成的抓手。抓手一按分类看热门用get_gzh_article_by_hot热点分类在WechatSogouConst.hot_index里一应俱全科技、财经、美食、教育、汽车、体育……每天跑一遍就能拿到当日的分类热门清单。hot_food ws.get_gzh_article_by_hot(WechatSogouConst.hot_index.food) for item in hot_food[:5]: print(item[gzh][wechat_name], item[article][title])抓手二扩展关键词用get_sugg不确定用户会搜什么直接问搜索引擎。get_sugg返回搜狗的关键词联想做选题和 SEO 都顺手for i, sugg in enumerate(ws.get_sugg(高考), 1): print(i, sugg)这部分解决你的问题把今天写什么从拍脑袋变成看数据同时把关键词库滚动扩起来。进阶一套能直接用的避坑封装到这里单点能力都有了但直接拼在一起跑生产你会撞上三堵墙请求太频被限、网络抖动失败、重复抓取浪费流量。下面这份封装把这三种情况一次性处理掉可以直接复制进你的项目。import hashlib import json import os import random import time class SafeFetcher: 带随机延迟 失败重试 文件缓存的采集外壳 def __init__(self, api, cache_dir./ws_cache, min_delay2, max_delay5, retry_times3): self.api api self.cache_dir cache_dir self.min_delay min_delay self.max_delay max_delay self.retry_times retry_times os.makedirs(cache_dir, exist_okTrue) def _delay(self): # 模拟人工浏览节奏降低触发反爬的概率 time.sleep(random.uniform(self.min_delay, self.max_delay)) def _cache_path(self, key): return os.path.join(self.cache_dir, f{key}.json) def get(self, func_name, *args, **kwargs): 带缓存 重试的通用调用入口 cache_key hashlib.md5( f{func_name}:{args}:{kwargs}.encode()).hexdigest() cache_file self._cache_path(cache_key) # 命中缓存直接返回当天数据不再重复抓 if os.path.exists(cache_file): with open(cache_file, encodingutf-8) as f: return json.load(f) self._delay() last_error None for attempt in range(self.retry_times): try: result getattr(self.api, func_name)(*args, **kwargs) # 生成器类型的结果先落成列表再缓存 if hasattr(result, __next__): result list(result) with open(cache_file, w, encodingutf-8) as f: json.dump(result, f, ensure_asciiFalse) return result except Exception as e: last_error e time.sleep(3 * (attempt 1)) # 失败后退避再试 raise last_error用法很简单fetcher SafeFetcher(ws) gzh_list fetcher.get(search_gzh, 数据分析, page1) # 第二次调用直接读缓存这部分解决你的问题把频率控制、重试退避、结果缓存三件事收敛成一个对象你写的业务代码就只剩调get这一条路径干净很多。再给一份故障排查思路按优先级排遇到问题照着查搜狗弹出验证码页面 → 这是正常的反爬触发。库内置了验证码处理回调默认走手动输入也可以把wechatsogou/identify_image.py里的识别函数换成第三方 OCR 服务自动过码。返回结果为空 → 先单独跑一次同样的关键词确认是真的没结果还是被限流。被限流的表现通常是页面结构异常。频繁 403 / 超时 → 把请求间隔拉大到 5 秒以上并考虑给WechatSogouAPI传入proxies参数切换出口 IP。链接失效 → 检查是不是隔了很久才去抓正文临时链接有效期有限务必当天抓取当天落库。最后今天就能做的三件事把前面所有内容浓缩成一张行动清单照着做即可安装并冒烟执行pip install wechatsogou --upgrade跑通第一段search_gzh示例确认环境没问题。写一个最小监控脚本挑一个你关心的关键词或公众号用SafeFetcher包一层每天定时跑一次把结果存成 JSON 或写入 SQLite——别急着上 MySQL、MongoDB先跑通再谈规模。记住那条边界单号只能拿到最近 10 条群发、文章链接会过期所以采集方案要设计成定期增量 及时落库而不是一次全量。技术本身是中性的公众号数据采集也不例外。搜狗微信搜索和微信官方都设有访问频率与内容使用的限制合理控制请求节奏、尊重内容版权、只采集你确实需要的数据才能让这套脚本长期稳定地跑下去。那么问题来了你的第一份公众号监控名单打算从哪个行业开始建微信生态的数据每天都在滚动更新而获取它的工具已经摆在眼前了。剩下的事就是动手。【免费下载链接】WechatSogou基于搜狗微信搜索的微信公众号爬虫接口项目地址: https://gitcode.com/gh_mirrors/we/WechatSogou创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考