Python 之微信指数小程序数据抓取

news2024/10/6 4:01:06

Fiddler安装和设置

安装

Fiddler 安装包可以从这里获取，如果失效了可以自己网上找一个安装。

链接：https://pan.baidu.com/s/1N30BoDWm2_dBL8i8GRzK5g?pwd=1znv
提取码：1znv

然后就是点击安装就好了，没什么好多说的。

启用HTTPS捕获

进入软件界面，点击 Tools -> Options -> HTTPS 启用捕获 https 请求并解密。

设置自动转发

设置指定 url 自动转发到本地，我这里是自动把请求转发到了我本地一个 Flask 搭建的服务，设置好以后进行保存（转发地址记得和你服务的地址保持一致）。

设置自动转发 https://search.weixin.qq.com/cgi-bin/wxaweb/wxindexfluctuations 的目的主要是为了获取数据请求参数中的 openid 和 search_key，因为我需要这两个请求参数去构造新的 body。

开启捕获

可以从 File -> Capture Traffic 开启捕获，也可以用 F12 快捷键开启捕获，当左下角有 Capturing 字样时，表示捕获已开启。

数据抓取处理

搭建并启动本地服务

可以自己在本地简单写一个服务接收和转发的请求并处理。我这里构造了两个 body 去分别获取 指数趋势 和 数据来源。

# coding:utf-8
import datetime
import json
import pygal
from pygal.style import Style

import requests
import urllib3
from flask import Flask, request

app = Flask(__name__)

headers = {'Host': 'search.weixin.qq.com',
           'Connection': 'keep-alive',
           # 'Content-Length': '182',
           'xweb_xhr': '1',
           'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/116.0.0.0 Safari/537.36 MicroMessenger/7.0.20.1781(0x6700143B) NetType/WIFI MiniProgramEnv/Windows WindowsWechat/WMPF WindowsWechat(0x6309092b) XWEB/9129',
           'Content-Type': 'application/json',
           'Accept': '*/*',
           'Sec-Fetch-Site': 'cross-site',
           'Sec-Fetch-Mode': 'cors',
           'Sec-Fetch-Dest': 'empty',
           'Referer': 'https://servicewechat.com/wxc026e7662ec26a3a/53/page-frame.html',
           'Accept-Encoding': 'gzip, deflate, br',
           'Accept-Language': 'zh-CN,zh;q=0.9',
           }


def draw_line(title, time_indexes, last_day=7):
    time_indexes = time_indexes[-last_day:]
    date_chart = pygal.StackedLine(fill=True, interpolate='hermite', x_label_rotation=-20, style=pygal.style.LightGreenStyle)
    date_chart.x_labels = [str(x["time"])[4:] for x in time_indexes]
    date_chart.add(title, [x["score"] for x in time_indexes])
    date_chart.render_to_file(f"line.svg")


def draw_pie(title, channel_scores, last_day=7):
    # 颜色对应关系可以使用 pyautogui 的 getpixel 取色器获取
    # colors_map = {
    #     "ad_score": "#eda150",
    #     "extlink_score": "#a9e87a",
    #     "finder_score": "#f6c443",
    #     "live_score": "#ff6146",
    #     "mpdoc_score": "#7c160",
    #     "query_score": "#4fadf8"
    # }
    # style = Style(colors=("#eda150", "#a9e87a", "#f6c443", "#ff6146", "#7c160", "#4fadf8"))
    legend_map = {
        "ad_score": "其他",
        "extlink_score": "网页",
        "finder_score": "视频号",
        "live_score": "直播",
        "mpdoc_score": "公众号",
        "query_score": "搜一搜",
    }
    channel_scores = channel_scores[-last_day:]
    channel_scores = [c["channel_score"] for c in channel_scores]
    channel_score = channel_scores.pop()
    for cs in channel_scores:
        for key, score in cs.items():
            channel_score[key] += score
    # pie_chart = pygal.Pie(inner_radius=0.5, style=pygal.style.LightSolarizedStyle)
    pie_chart = pygal.Pie(inner_radius=0.5)
    pie_chart.title = title
    # print(channel_score)
    total_score = channel_score["total_score"]
    for key, score in channel_score.items():
        if key in ["score_exp", "total_score"]:
            continue
        percent = float("{:.2f}".format(100 * score / total_score))
        pie_chart.add(legend_map[key], percent)
    pie_chart.render_to_file(f"pie.svg")


@app.route('/post_data', methods=['POST'])
def post():
    if request.method == 'POST':
        urllib3.disable_warnings()
        data = request.get_json()
        # print(data)
        openid = data.get("openid")
        search_key = data.get("search_key")
        query = [data.get("query")]
        end_ymd = datetime.datetime.now().strftime("%Y%m%d")
        start_ymd = (datetime.datetime.now() - datetime.timedelta(365)).strftime("%Y%m%d")
        forward_url = 'https://search.weixin.qq.com/cgi-bin/wxaweb/wxindex'
        # 指数趋势
        json_data = {'openid': openid, 'search_key': search_key, 'cgi_name': 'GetDefaultIndex',
                     'query': query, 'compound_word': [], 'start_ymd': start_ymd, 'end_ymd': end_ymd}
        response = requests.post(forward_url, json=json_data, headers=headers, verify=False)
        response_data = response.json()
        # json.dump(response_data, open("test1.json", "w"), indent=2)
        # print(response_data)
        # return
        title = response_data["content"]["resp_list"][0]["query"]
        time_indexes = response_data["content"]["resp_list"][0]["indexes"][0]["time_indexes"]
        # print(time_indexes)
        draw_line(title, time_indexes, 30)
        # # 数据来源
        json_data2 = {'openid': openid, 'search_key': search_key, 'cgi_name': 'GetMultiChannel',
                      'query': query, 'start_ymd': start_ymd, 'end_ymd': end_ymd}
        response = requests.post(forward_url, json=json_data2, headers=headers, verify=False)
        response_data = response.json()
        # json.dump(response_data, open("test2.json", "w"), indent=2)
        result_list = response_data["content"]["result_list"]
        draw_pie(title, result_list, 30)

    return {}


if __name__ == '__main__':
    app.run(host="127.0.0.1", debug=True)