import random import time import requests import json import urllib3 from requests.adapters import HTTPAdapter from common import Feishu, Material, AliyunLogger from common.sql_help import sqlCollect from data_channel.data_help import dataHelp class KS: @classmethod def get_share_count(cls, v_id): url = "http://8.217.192.46:8889/crawler/kuai_shou/detail" payload = json.dumps({ "content_id": v_id }) headers = { 'Content-Type': 'application/json' } try: time.sleep(random.uniform(1, 10)) response = requests.request("POST", url, headers=headers, data=payload, timeout=30) response = response.json() if response["code"] == 0: data = response["data"]["data"] share_count = data.get("share_count") return int(share_count) except KeyError as e: return 0 @classmethod def get_ks_url(cls, task_mark, url_id, number, mark, feishu_id, cookie_sheet, channel_id, name): list = [] pcursor = "" url = "https://www.kuaishou.com/graphql" for i in range(3): cookie = Material.get_cookie_data(feishu_id, cookie_sheet, channel_id) time.sleep(random.randint(1, 5)) payload = json.dumps({ "operationName": "visionProfilePhotoList", "variables": { "userId": url_id, "pcursor": pcursor, "page": "profile" }, "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 visionProfilePhotoList($pcursor: String, $userId: String, $page: String, $webPageArea: String) {\n visionProfilePhotoList(pcursor: $pcursor, userId: $userId, page: $page, webPageArea: $webPageArea) {\n result\n llsid\n webPageArea\n feeds {\n ...feedContent\n __typename\n }\n hostName\n pcursor\n __typename\n }\n}\n" }) headers = { 'accept': '*/*', 'content-type': 'application/json', 'Origin': 'https://www.kuaishou.com', 'Cookie': cookie, 'Accept-Language': 'zh-CN,zh;q=0.9', 'User-Agent': 'Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/125.0.0.0 Safari/537.36', 'Referer': f'https://www.kuaishou.com/profile/{url_id}', 'Accept-Encoding': 'gzip, deflate, br', 'Connection': 'keep-alive' } urllib3.disable_warnings() s = requests.session() s.mount('http://', HTTPAdapter(max_retries=3)) s.mount('https://', HTTPAdapter(max_retries=3)) # response = requests.request("POST", url, headers=headers, data=payload, timeout=10) try: response = s.post(url=url, headers=headers, data=payload, verify=False, timeout=10) response.close() if response.status_code != 200: return list elif response.status_code == 200 and "error_msg" in response.text: Feishu.bot("wangxueke", '机器自动改造消息通知', f'快手-{name}cookie过期，请及时更换', 'wangxueke') Feishu.bot("liuzhaoheng", '机器自动改造消息通知', f'快手-{name}cookie过期，请及时更换', '刘兆恒') return list elif "visionProfilePhotoList" not in response.json()["data"]: if name == '快手品类账号': Feishu.bot("wangxueke", '机器自动改造消息通知', f'快手-{name}cookie过期，请及时更换', 'wangxueke') Feishu.bot("liuzhaoheng", '机器自动改造消息通知', f'快手-{name}cookie过期，请及时更换', '刘兆恒') else: Feishu.bot(mark, '机器自动改造消息通知', f'快手-{name}cookie过期，请及时更换', name) return list elif "feeds" not in response.json()["data"]["visionProfilePhotoList"]: if name == '快手品类账号': Feishu.bot("wangxueke", '机器自动改造消息通知', f'快手-{name}cookie过期，请及时更换', 'wangxueke') Feishu.bot("liuzhaoheng", '机器自动改造消息通知', f'快手-{name}cookie过期，请及时更换', '刘兆恒') else: Feishu.bot(mark, '机器自动改造消息通知', f'快手-{name}cookie过期，请及时更换', name) return list elif len(response.json()["data"]["visionProfilePhotoList"]["feeds"]) == 0: if name == '快手品类账号': Feishu.bot("wangxueke", '机器自动改造消息通知', f'快手-{name}cookie过期，请及时更换', 'wangxueke') Feishu.bot("liuzhaoheng", '机器自动改造消息通知', f'快手-{name}cookie过期，请及时更换', '刘兆恒') else: Feishu.bot(mark, '机器自动改造消息通知', f'快手-{name}cookie使用频繁无法获取到数据，请及时更换', name) return list pcursor = response.json()['data']['visionProfilePhotoList']['pcursor'] feeds = response.json()['data']['visionProfilePhotoList']['feeds'] for i in range(len(feeds)): # try: # video_id = feeds[i].get("photo", {}).get("videoResource").get("h264", {}).get("videoId", "") # except KeyError: # video_id = feeds[i].get("photo", {}).get("videoResource").get("hevc", {}).get("videoId", "") # status = sqlCollect.is_used(task_mark, video_id, mark, channel_id) # if status: # continue video_id = feeds[i].get("photo", {}).get("id", "") status = sqlCollect.is_used(task_mark, video_id, mark, channel_id) old_title = feeds[i].get("photo", {}).get("caption") cover_url = feeds[i].get('photo', {}).get('coverUrl', "") video_url = feeds[i].get('photo', {}).get('photoUrl', "") view_count = int(feeds[i].get('photo', {}).get('viewCount', 0)) realLikeCount = int(feeds[i].get('photo', {}).get('realLikeCount', 0)) video_percent = '%.4f' % (share_count / view_count) duration = dataHelp.video_duration(video_url) log_data = f"user:{url_id},,video_id:{video_id},,video_url:{video_url},,original_title:{old_title},,share_count:{share_count},,view_count:{view_count},,duration:{duration}" # log_data = f"user:{url_id},,video_id:{video_id},,video_url:{video_url},,original_title:{old_title},,view_count:{view_count},,duration:{duration}" AliyunLogger.logging(channel_id, name, url_id, video_id, "扫描到一条视频", "2001", log_data) # if status: # AliyunLogger.logging(channel_id, name, url_id, video_id, "该视频已改造过", "2002", log_data) # continue share_count = cls.get_share_count(video_id) special = float(0.001) if float(video_percent) < special: AliyunLogger.logging(channel_id, name, url_id, video_id, "不符合规则:分享/浏览小于0.001", "2003", log_data) continue if share_count < 500: AliyunLogger.logging(channel_id, name, url_id, video_id, "不符合规则:分享小于500", "2003", log_data) continue if duration < 30 or duration > 720: AliyunLogger.logging(channel_id, name, url_id, video_id, "不符合规则:时长不符合规则大于720秒/小于30秒", "2003", log_data) continue all_data = {"video_id": video_id, "cover": cover_url, "video_url": video_url, "rule": video_percent, "old_title": old_title} list.append(all_data) AliyunLogger.logging(channel_id, name, url_id, video_id, "符合规则等待改造", "2004", log_data) if len(list) == int(number): return list except Exception as exc: # Feishu.bot("wangxueke", '机器自动改造消息通知', f'快手-{name}cookie过期，请及时更换', 'wangxueke') # Feishu.bot("liuzhaoheng", '机器自动改造消息通知', f'快手-{name}cookie过期，请及时更换', '刘兆恒') return list return list if __name__ == '__main__': KS.get_ks_url("1","3xzicxg2nandemc",1,"1",'WuoQsVFXChVMK4tDHqLcwLWgnjh','Sjk8p8','','')