Python获取微信公众号文章数据

news2024/11/18 7:36:41

这是一个通过 Python mitmproxy 库实现获取某个微信公众号下全部文章数据的解决方案。首先需要创建一个 Python 虚拟环境，并进入虚拟环境下：

$ python -m venv venv
$ venv/Scripts/activate

我们需要使用 mitmproxy 库来建立一个网络代理，以实现监控微信公众号请求的需求。通过下面的命令安装 mitmproxy 库到虚拟环境：

$ pip install mitmproxy

然后在项目的根目录下创建一个 mitmproxy 插件文件，这个插件的核心逻辑在 HTTP 事件钩子 - 收到完整的响应 时，判断是否微信公众号的文章数据相关请求和响应，如果是就取出我们需要的数据，并写入文件中。插件的完整代码如下：

import json
from mitmproxy import ctx, http

class Agency:
    def __init__(self):
        self.filter_host = 'mp.weixin.qq.com'
        self.filter_path_list = ['/mp/getappmsgext', '/mp/appmsg_comment', '/cgi-bin/appmsg']
        self.article_data = {}

    def request(self, flow: http.HTTPFlow):
        """ HTTP 事件钩子 - 发出完整的请求 """
        pass
    
    def response(self, flow: http.HTTPFlow):
        """ HTTP 事件钩子 - 收到完整的响应 """
        if flow.request.host == self.filter_host:
            path = flow.request.path.split('?')[0]
            if path in self.filter_path_list:
                if path == '/mp/getappmsgext':
                    response_json = json.loads(flow.response.text)
                    if 'appmsgstat' in response_json:
                        appmsgstat = response_json['appmsgstat']
                        self.article_data['read_num'] = int(appmsgstat['read_num'])  # 阅读数
                        self.article_data['like_num'] = int(appmsgstat['like_num'])  # 在看数
                        self.article_data['old_like_num'] = int(appmsgstat['old_like_num'])  # 点赞数
                elif path == '/mp/appmsg_comment':
                    response_json = json.loads(flow.response.text)
                    if 'elected_comment_total_cnt' in response_json:
                        comment_total = response_json['elected_comment_total_cnt']
                        self.article_data['elected_comment_total_cnt'] = int(comment_total)  # 评论数
                elif path == '/cgi-bin/appmsg':
                    response_json = json.loads(flow.response.text)
                    if 'app_msg_list' in response_json:
                        app_msg_list = response_json['app_msg_list']
                        for app_msg in app_msg_list:
                            title = app_msg['title']  # 标题
                            digest = app_msg['digest']  # 概要
                            link = app_msg['link']  # 链接
                            msg = {'title': title, 'digest': digest, 'link': link}
                            with open('article_list.log', 'a+', encoding='utf-8') as file:
                                file.write(f'{json.dumps(msg, ensure_ascii=False)}\n')
        if len(self.article_data) == 4:
            with open('article_detail.log', 'w+', encoding='utf-8') as file:
                file.write(f'{json.dumps(self.article_data, ensure_ascii=False)}')
            self.article_data = {}

addons = [Agency()]