LongArticleTaskServer/app/infra/crawler/wechat/gzh_fans.py

import random

from app.infra.shared import AsyncHttpClient


# 抓取公众号粉丝
async def get_gzh_fans(token, cookie, cursor_id, cursor_timestamp):
    url = "https://mp.weixin.qq.com/cgi-bin/user_tag"
    params = {
        "action": "get_user_list",
        "groupid": "-2",
        "begin_openid": cursor_id,
        "begin_create_time": cursor_timestamp,
        "limit": "20",
        "offset": "0",
        "backfoward": "1",
        "token": token,
        "lang": "zh_CN",
        "f": "json",
        "ajax": "1",
        "fingerprint": "42694935a543ed714f89d4b05ef3a4e6",
        "random": random.random(),
    }
    headers = {
        "accept": "application/json, text/javascript, */*; q=0.01",
        "accept-language": "zh-CN,zh;q=0.9",
        "cache-control": "no-cache",
        "pragma": "no-cache",
        "priority": "u=1, i",
        "referer": f"https://mp.weixin.qq.com/cgi-bin/user_tag?action=get_all_data&lang=zh_CN&token={token}",
        "sec-ch-ua": '"Google Chrome";v="143", "Chromium";v="143", "Not A(Brand";v="24"',
        "sec-ch-ua-mobile": "?0",
        "sec-ch-ua-platform": '"macOS"',
        "sec-fetch-dest": "empty",
        "sec-fetch-mode": "cors",
        "sec-fetch-site": "same-origin",
        "user-agent": "Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/143.0.0.0 Safari/537.36",
        "x-requested-with": "XMLHttpRequest",
        "Cookie": cookie,
    }
    # 发送请求
    async with AsyncHttpClient(timeout=10) as http_client:
        response = await http_client.get(url, headers=headers, params=params)

    return response


# 获取 access_token
async def get_access_token(app_id, app_secret):
    url = f"https://api.weixin.qq.com/cgi-bin/stable_token"
    data = {"grant_type": "client_credential", "appid": app_id, "secret": app_secret}
    async with AsyncHttpClient(timeout=100) as http_client:
        response = await http_client.post(url, json=data)

    return response


# 批量获取 union_id
async def get_union_id_batch(access_token, user_list):
    url = f"https://api.weixin.qq.com/cgi-bin/user/info/batchget?access_token={access_token}"
    payload = {
        "user_list": user_list,
    }
    async with AsyncHttpClient(timeout=100) as http_client:
        response = await http_client.post(url, json=payload)

    return response