新增抓公众号粉丝

applications/crawler/wechat/gzh_fans.py

@@ -0,0 +1,66 @@
+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