前一阵子在教 Cathy 编程，作为小白的话用 Python 入门自然是极好的。

最近她有了个需求，说要爬取“材料人”这个公众号的所有文章。我吹牛说1分钟可以搞定，其实是因为之前早就把代码写好了（笑~）。代码也不难，不到50行，掌握思路后其实每个人都可以学编程。

整体思路

首先，爬取微信公众号的文章需要几个参数，

  • biz：这个是每个公众号base64的编码，基本不变
  • key：15分钟变一次，每次必须重新生成
  • uin/pass_ticket：用户信息相关，基本不变

这四个参数都可以通过 Charles 等抓包软件看到，就不再具体展开，基本上也只需要修改 key 这一个参数即可。

有了请求数据，咱们先请求下 mp.weixin.qq.com/mp/getmasssendmsg ，它可以帮忙重定向并且设置一些用户信息(cookie)，接着再用递归的方式来不断请求 mp.weixin.qq.com/mp/profile_ext 这个接口，它就是模拟下拉刷新来不断翻页获取历史文章。当然，为了避免被微信发现你其实是只爬虫🕷，适当的 sleep 一秒钟才是上策。代码中大部分时候是在解析爬取到的 JSON 数据，解析完毕之后用表格（csv）一装，欧耶，可视化历史文章数据搞定！是不是相当简单？

效果预览：

整体代码如下，记得先用 pip install requests 装个网络请求库：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
import time, json, csv, requests
class WeChat:
    def __init__(self, biz, key):
        self.biz = biz # 想爬的公众号
        self.key = key # key是不断变化的
        self.uin = YOUR_UIN
        self.pass_ticket = YOUR_PASS_TICKET
        self.headers = {'user-agent': 'Mozilla/5.0 (Macintosh; Intel Mac OS X 10_12_6)'}
        self.session = requests.Session()
        self.articles = []
    def prepare_for_load(self):
        url = 'https://mp.weixin.qq.com/mp/getmasssendmsg'
        params = { '__biz' : self.biz, 'uin' : self.uin, 'key' : self.key, 'pass_ticket' : self.pass_ticket }
        r = self.session.get(url, params=params, headers=self.headers, allow_redirects=True, verify=False)
    def load_articles(self, offset=0):
        url = 'https://mp.weixin.qq.com/mp/profile_ext'
        params = { 'action' : 'getmsg', '__biz' : self.biz, 'f' : 'json', 'offset' : offset, 'count' : 10 }
        r = self.session.get(url, params=params, headers=self.headers, allow_redirects=True, verify=False)
        time.sleep(1) # 让微信认为你是真实用户
        d = r.json()
        general_msg_list = d.get('general_msg_list')
        can_msg_continue = d.get('can_msg_continue')
        next_offset = d.get('next_offset')
        cards = []
        try:
            cards = json.loads(general_msg_list).get('list')
        except Exception as e:
            pass
        for card in cards:
            if card.get('comm_msg_info').get('type') == 49: # 图文
                self.articles.append(card.get('app_msg_ext_info'))
                if app_msg_ext_info.get('is_multi'):
                    multi_app_msg_item_list = app_msg_ext_info.get('multi_app_msg_item_list') || []
                    for multi_app_msg_item in multi_app_msg_item_list:
                        self.articles.append(multi_app_msg_item)
        if can_msg_continue:
            self.load_articles(offset=next_offset) #模拟下拉刷新递归获取更多文章
wechat = WeChat(biz=YOUR_BIZ, key=YOUR_KEY)
wechat.prepare_for_load()
wechat.load_articles()
with open('wechat.csv', 'wt', encoding='utf-8') as csvFile:
    for a in wechat.articles:
        csv.writer(csvFile).writerow([a.title, a.content_url])

进阶

以上代码只适用于个人方便收集某个公众号的全部历史文章，如果要像即刻或者一订那样进行大规模微信公众号文章的采集，就不能依赖微信的接口直接访问，而应该通过搜狗微信搜索暴露的接口进行爬取，这将会涉及到代理服务器的编写，分布式爬虫系统的搭建，动态IP池的管理，由于目前我个人精力有限，可能暂时不会投入时间和精力去实现，有想法的朋友可以直接找我交流方案。


阅读