# Author renyupeng # coding=utf-8 # @Time : 2023/3/23 2:49 下午 # @Site : # @File : SpiderPromoterHeader.py # @Software: PyCharm # @contact: renyupeng@c-top.com.cn # @Tel 1501435553 import json import requests from utils.mysql_utils_pro import MysqlProUtils from utils.selenium import selenium class SpiderPromoterHeader: def __init__(self): self.conn = MysqlProUtils() @staticmethod def GetUserID(promoter_id): url = "https://www.kuaishou.com/graphql" payload = json.dumps({ "operationName": "graphqlSearchUser", "variables": { "keyword": "{promoter_id}".format(promoter_id=promoter_id) }, "query": "query graphqlSearchUser($keyword: String, $pcursor: String, $searchSessionId: String) {\n visionSearchUser(keyword: $keyword, pcursor: $pcursor, searchSessionId: $searchSessionId) {\n result\n users {\n fansCount\n photoCount\n isFollowing\n user_id\n headurl\n user_text\n user_name\n verified\n verifiedDetail {\n description\n iconType\n newVerified\n musicCompany\n type\n __typename\n }\n __typename\n }\n searchSessionId\n pcursor\n __typename\n }\n}\n" }) headers = { 'Cookie': 'clientid=3; client_key=65890b29; soft_did=1619580708547; didv=1674986117437; kpf=PC_WEB; clientid=3; did=web_6cdc432ef91373345c3b4f1b16559976; apdid=87629ef8-112a-405e-aa68-6532dad20945f1aacbdb5b9379d0b02fa529ecbc80c2:1681378698:1; userId=1433645207; kuaishou.server.web_st=ChZrdWFpc2hvdS5zZXJ2ZXIud2ViLnN0EqABO7d6hu9qxJCzWjVA62hSp9zWXg9SCNBU45lpRe7kz0KnW7FZL4QvULoBerxFtRIZceE96YcmTFvXPVu02rKu_IhCjfE5toca2N1Hxq6NNsk42T2s1U48JjbYe-48lzAOp0SU3G9wM6zSeTNcBwVJVZ-FL5u5Je0j9gzWo4gxhKIuRgFY37kWOpa7uG9JINrjynHlOhHm2Qrs_6HZHk-4ahoSWXnZQFypWC8Fi7687FtZGgfDIiCPf5NE2AVLBwwduWgffV0fhRYpMDq9izf6QN8O_7b0CCgFMAE; kuaishou.server.web_ph=cdd0202e2c6cbffc6c4425d8534a070a67c1; kpn=KUAISHOU_VISION', 'User-Agent': 'Apifox/1.0.0 (https://www.apifox.cn)', 'Content-Type': 'application/json', 'Accept': '*/*', 'Host': 'www.kuaishou.com', 'Connection': 'keep-alive' } response = requests.request("POST", url, headers=headers, data=payload) print(response.text, '--------text-------------') try: if json.loads(response.text) is not None: data = json.loads(response.text)["data"] if data is not None: visionSearchUser = data["visionSearchUser"] if visionSearchUser is not None: userList = visionSearchUser["users"] if userList is not None: List = [] for user in userList: user_id = user["user_id"] List.append(user_id) return List else: pass except Exception as e: print('获取用户ID错误', response.text) @staticmethod def GetPromoterInfo(promoter_id): List = SpiderPromoterHeader.GetUserID(promoter_id) listItem = [] for user_id in List: url = "https://www.kuaishou.com/graphql" payload = json.dumps({ "operationName": "visionProfile", "variables": { "userId": user_id }, "query": "query visionProfile($userId: String) {\n visionProfile(userId: $userId) {\n result\n hostName\n userProfile {\n ownerCount {\n fan\n photo\n follow\n photo_public\n __typename\n }\n profile {\n gender\n user_name\n user_id\n headurl\n user_text\n user_profile_bg_url\n __typename\n }\n isFollowing\n __typename\n }\n __typename\n }\n}\n" }) headers = { 'Cookie': 'clientid=3; client_key=65890b29; soft_did=1619580708547; didv=1674986117437; kpf=PC_WEB; clientid=3; did=web_6cdc432ef91373345c3b4f1b16559976; apdid=87629ef8-112a-405e-aa68-6532dad20945f1aacbdb5b9379d0b02fa529ecbc80c2:1681378698:1; userId=1433645207; kpn=KUAISHOU_VISION; kuaishou.server.web_st=ChZrdWFpc2hvdS5zZXJ2ZXIud2ViLnN0EqABJmId7nrMbNwNCAHx1plSfa7wKs8OKhC3Sh7yNwmIUAgRDb0NpoLqRQbvI4wLa-2oGiq5jAT0X4ZVgIcC1VkicZ4nfCqMoOaP_ZdJxTXtr1okc0NLqVbArL_daeHhvFs9foIyhppgogJzQIWmEelbFhy66FtfTrpPIDDU4U3iE68XzKJ3qQBPiTiRSqdybKLbiG7g4dIOJEcaR-uRWz2I7RoSTdCMiCqspRXB3AhuFugv61B-IiD_sF73q-l2oDD0TV2N22UFgv4yH9grr9eTElLXKz2tjSgFMAE; kuaishou.server.web_ph=f30c9f58ebbab9e7949f3bcf778585b81a3e', 'User-Agent': 'Apifox/1.0.0 (https://www.apifox.cn)', 'Content-Type': 'application/json', 'Accept': '*/*', 'Host': 'www.kuaishou.com', 'Connection': 'keep-alive' } try: response = requests.request("POST", url, headers=headers, data=payload) data = json.loads(response.text)["data"] if data is not None: visionProfile = data["visionProfile"] if visionProfile is not None: userProfile = visionProfile["userProfile"] if userProfile is not None: fans = userProfile["ownerCount"]["fan"] headurl = userProfile["profile"]["headurl"] user_name = userProfile["profile"]["user_name"] item = {"fans": fans, "handurl": headurl, "promoter_name": user_name, "promoter_id": promoter_id} listItem.append(item) print(listItem) return listItem else: continue except Exception as e: print('请求达人详情异常', e) def GetPromoterID(self): try: sql = """select a.promoter_id from ruixuan.promoter_oerder_amount a left join ruixuan.promoter_info b on a.promoter_id=b.promoter_id where b.promoter_id is null and a.`30day_gmv` is not null group by a.promoter_id""" promoterIdList = self.conn.QueryAll(sql) return promoterIdList except Exception as e: print(e) def PromoterInsert(self): promoterIdList = self.GetPromoterID() try: for promoterId in promoterIdList: print(promoterId) try: listItem = SpiderPromoterHeader.GetPromoterInfo(promoterId[0]) if listItem is not None: if listItem.__len__() > 0: for item in listItem: try: self.conn.insert(item) except Exception as e: print('插入错误', item) except Exception as e: print(e) except Exception as e: print('获取达人详情错误', e) if __name__ == '__main__': SpiderPromoterHeader().PromoterInsert()