| 123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566 |
- import random
- from applications.utils 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/token?grant_type=client_credential&appid={app_id}&secret={app_secret}"
- async with AsyncHttpClient(timeout=100) as http_client:
- response = await http_client.get(url)
- 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
|
