# import time # # import requests # import json # # from common import Common, AliyunLogger, Feishu, Material # from common.sql_help import sqlCollect # from common.userAgent import get_random_user_agent # # # class KsPcKeyword: # @classmethod # def get_key_word(cls, keyword, task_mark, mark, channel_id, name, task): # list = [] # url = "https://www.kuaishou.com/graphql" # # payload = json.dumps({ # "operationName": "visionSearchPhoto", # "variables": { # "keyword": keyword, # "pcursor": "", # "page": "search" # }, # "query": "fragment photoContent on PhotoEntity {\n __typename\n id\n duration\n caption\n originCaption\n likeCount\n viewCount\n commentCount\n realLikeCount\n coverUrl\n photoUrl\n photoH265Url\n manifest\n manifestH265\n videoResource\n coverUrls {\n url\n __typename\n }\n timestamp\n expTag\n animatedCoverUrl\n distance\n videoRatio\n liked\n stereoType\n profileUserTopPhoto\n musicBlocked\n riskTagContent\n riskTagUrl\n}\n\nfragment recoPhotoFragment on recoPhotoEntity {\n __typename\n id\n duration\n caption\n originCaption\n likeCount\n viewCount\n commentCount\n realLikeCount\n coverUrl\n photoUrl\n photoH265Url\n manifest\n manifestH265\n videoResource\n coverUrls {\n url\n __typename\n }\n timestamp\n expTag\n animatedCoverUrl\n distance\n videoRatio\n liked\n stereoType\n profileUserTopPhoto\n musicBlocked\n riskTagContent\n riskTagUrl\n}\n\nfragment feedContent on Feed {\n type\n author {\n id\n name\n headerUrl\n following\n headerUrls {\n url\n __typename\n }\n __typename\n }\n photo {\n ...photoContent\n ...recoPhotoFragment\n __typename\n }\n canAddComment\n llsid\n status\n currentPcursor\n tags {\n type\n name\n __typename\n }\n __typename\n}\n\nquery visionSearchPhoto($keyword: String, $pcursor: String, $searchSessionId: String, $page: String, $webPageArea: String) {\n visionSearchPhoto(keyword: $keyword, pcursor: $pcursor, searchSessionId: $searchSessionId, page: $page, webPageArea: $webPageArea) {\n result\n llsid\n webPageArea\n feeds {\n ...feedContent\n __typename\n }\n searchSessionId\n pcursor\n aladdinBanner {\n imgUrl\n link\n __typename\n }\n __typename\n }\n}\n" # }) # cookie = Material.get_cookie_data("KsoMsyP2ghleM9tzBfmcEEXBnXg", "U1gySe", "快手搜索-cookie") # headers = { # 'Accept-Language': 'zh-CN,zh;q=0.9', # 'Cache-Control': 'no-cache', # 'Connection': 'keep-alive', # 'Origin': 'https://www.kuaishou.com', # 'Pragma': 'no-cache', # 'User-Agent': get_random_user_agent("pc"), # 'accept': '*/*', # 'content-type': 'application/json', # 'Cookie': cookie # } # try: # time.sleep(3) # # 代理信息 # proxy = "http://spkbt3wnzw:cx6R=v5mQuBgqsQ4o7@cn.visitxiangtan.com:30000" # proxies = { # "http": proxy, # "https": proxy # } # response = requests.request("POST", url, headers=headers, data=payload, proxies=proxies) # text = response.text # if text: # response_dict = json.loads(text) # result = response_dict.get('result', None) # if result: # log_type = ['liukunyu', 'wangxueke', 'xinxin'] # mark_name = ['刘坤宇', '王雪珂', '信欣'] # Feishu.bot(log_type, '快手关键词搜索', f'快手关键词搜索cookie过期，请及时更换', mark_name) # time.sleep(10) # return list # response = response.json() # data_list = response['data']['visionSearchPhoto']['feeds'] # for data in data_list: # data = data['photo'] # photo_id = data["id"] # status = sqlCollect.is_used(task_mark, photo_id, mark, channel_id) # # view_count = data["viewCount"] if "viewCount" in data and data["viewCount"] else 0 # like_count = data["likeCount"] if "likeCount" in data and data["likeCount"] else 0 # like_count = cls.convert_to_number(like_count) # video_percent = '%.4f' % (int(like_count) / int(view_count)) # special = 0.015 # old_title = data["caption"] # 标题 # duration = data["duration"] # duration = int(duration) / 1000 # video_url = data["photoUrl"] # image_url = data["coverUrl"] # log_data = f"user:{keyword},,video_id:{photo_id},,video_url:{video_url},original_title:{old_title},,like_count:{like_count},,view_count:{view_count},,duration:{duration}" # AliyunLogger.logging(channel_id, name, keyword, photo_id, "扫描到一条视频", "2001", log_data) # if status: # AliyunLogger.logging(channel_id, name, keyword, photo_id, "该视频已改造过", "2001", log_data) # continue # if int(view_count) < 1000: # AliyunLogger.logging(channel_id, name, keyword, photo_id, f"不符合规则:浏览小于1000", "2003", log_data) # Common.logger("ks-key-word").info( # f"不符合规则：{task_mark},用户主页id:{keyword},视频id{photo_id} ,浏览:{view_count},浏览{view_count} ,时长:{int(duration)} ") # continue # if float(video_percent) < special: # AliyunLogger.logging(channel_id, name, keyword, photo_id, f"不符合规则:点赞/浏览{special}", "2003", log_data) # Common.logger("ks-key-word").info( # f"不符合规则：{task_mark},用户主页id:{keyword},视频id{photo_id} ,浏览:{view_count},浏览{view_count} ,时长:{int(duration)} ") # continue # if int(duration) < 30 or int(duration) > 600: # AliyunLogger.logging(channel_id, name, keyword, photo_id, # f"不符合规则:时长不符合规则大于600秒/小于30秒", "2003", # log_data) # # Common.logger("ks-key-word").info( # f"不符合规则：{task_mark},用户主页id:{keyword},视频id{photo_id} ,浏览:{view_count},浏览{view_count} ,时长:{int(duration)} ") # continue # AliyunLogger.logging(channel_id, name, keyword, photo_id, "符合规则等待改造", "2004", log_data) # all_data = {"video_id": photo_id, "cover": image_url, "video_url": video_url, # "rule": '', # "old_title": old_title} # list.append(all_data) # return list # except Exception as exc: # Common.logger("ks-key-word").info(f"快手搜索词{keyword}获取失败{exc}\n") # return list # # @classmethod # def convert_to_number(cls, value): # if value.endswith("万"): # return float(value[:-1]) * 10000 # 去掉“万”并乘以 10000 # return int(value) # 处理其他格式 # # # if __name__ == '__main__': # keyword = '毛主席故居' # task_mark = '1' # mark = 'pl-gjc' # channel_id = '快手搜索' # name = '1' # task = {'combo': ['最新发布', '近1日', '1分钟内']} # KsPcKeyword.get_key_word(keyword, task_mark, mark, channel_id, name, task)