import json import os from hashlib import md5 import requests import time from urllib.parse import urlencode from datetime import datetime, timedelta from common import Oss, Feishu, Common, AliyunLogger from common.sql_help import sqlCollect headers = { 'Accept-Language': 'zh-cn', 'Connection': 'keep-alive', 'Content-Type': 'application/x-www-form-urlencoded', 'Host': 'creator-app.kuaishou.com', 'User-Agent': 'kwai-android aegon/3.12.1', } class KsFeedVideo: CATEGORY_IDS = { 1: "生活", 2: "才艺", # 3: "时尚", # 4: "宠物", 5: "读书", # 6: "二次元", 7: "家居", # 8: "数码", 9: "搞笑", 10: "健康", 11: "旅游", 12: "美食", # 13: "美妆", # 14: "汽车", 15: "亲子", 16: "情感", # 17: "三农", # 18: "摄影", # 19: "舞蹈", # 20: "颜值", # 21: "音乐", # 22: "影视", # 23: "短剧", # 24: "游戏", 25: "运动", 26: "资讯", 27: "人文" } current_category_index = 0 @staticmethod def calculate_sig(data): src = ''.join([f'{key}={data[key]}' for key in sorted(data.keys())]) salt = '08d8eece8e83' return md5(f'{src}{salt}'.encode()).hexdigest() """ 切换品类 """ @classmethod def switch_category(cls): if cls.current_category_index >= len(cls.CATEGORY_IDS): cls.current_category_index = 0 category_id = list(cls.CATEGORY_IDS.keys())[cls.current_category_index] url = 'https://creator-app.kuaishou.com/rest/bamboo/inspiration/n/category/confirm/optimize' data = { 'isRecommendChange': False, 'categoryId': category_id, # 'kuaishou.api_st': "Cg9rdWFpc2hvdS5hcGkuc3QSkAGMQoIK2ZpwlQszYISTxSFxzugi58w2U5gpPqa6an0eU6MFcVsXq2rd_K16UTItZ_OzPV-4jmVN5rNXKXW9jL97JV79Y9PqxaR9xOIr1TEyDzpOq2GM-0W1QRW3M8Li_J6NZ5t1hRFCWHBlOESjiBWs7vq4m1bq_ml0dZ6pgEDfpsWNpBaLRzwZwOO1mD4LqO4aEokh6uHql0RmmtbfoBF25r7QOyIgqNv0TBf6mlwS3bjE0K6sl08M1mMPjW1PB9e0Qr494H8oBTAB", 'kuaishou.api_st': 'Cg9rdWFpc2hvdS5hcGkuc3QSkAGMQoIK2ZpwlQszYISTxSFxzugi58w2U5gpPqa6an0eU6MFcVsXq2rd_K16UTItZ_OzPV-4jmVN5rNXKXW9jL97JV79Y9PqxaR9xOIr1TEyDzpOq2GM-0W1QRW3M8Li_J6NZ5t1hRFCWHBlOESjiBWs7vq4m1bq_ml0dZ6pgEDfpsWNpBaLRzwZwOO1mD4LqO4aEokh6uHql0RmmtbfoBF25r7QOyIgqNv0TBf6mlwS3bjE0K6sl08M1mMPjW1PB9e0Qr494H8oBTAB;region_ticket=RT_FAC86448E713714136C088FFCC4431455D1FA7E05A6D25DAD4E4B8CC011FB6E8294169DD9', 'client_key': '214c9979', } sig = cls.calculate_sig(data) data['sig'] = sig response = requests.post(url=url, headers=headers, data=data) body = response.content.decode() cls.current_category_index += 1 return body """ 获取feed流信息 """ @classmethod def get_feed_list(cls): cls.switch_category() url = 'https://creator-app.kuaishou.com/rest/bamboo/inspiration/n/feed' data = { 'cs': False, 'kuaishou.api_st': 'Cg9rdWFpc2hvdS5hcGkuc3QSkAGMQoIK2ZpwlQszYISTxSFxzugi58w2U5gpPqa6an0eU6MFcVsXq2rd_K16UTItZ_OzPV-4jmVN5rNXKXW9jL97JV79Y9PqxaR9xOIr1TEyDzpOq2GM-0W1QRW3M8Li_J6NZ5t1hRFCWHBlOESjiBWs7vq4m1bq_ml0dZ6pgEDfpsWNpBaLRzwZwOO1mD4LqO4aEokh6uHql0RmmtbfoBF25r7QOyIgqNv0TBf6mlwS3bjE0K6sl08M1mMPjW1PB9e0Qr494H8oBTAB;region_ticket=RT_FAC86448E713714136C088FFCC4431455D1FA7E05A6D25DAD4E4B8CC011FB6E8294169DD9', # 'kuaishou.api_st': "Cg9rdWFpc2hvdS5hcGkuc3QSkAGMQoIK2ZpwlQszYISTxSFxzugi58w2U5gpPqa6an0eU6MFcVsXq2rd_K16UTItZ_OzPV-4jmVN5rNXKXW9jL97JV79Y9PqxaR9xOIr1TEyDzpOq2GM-0W1QRW3M8Li_J6NZ5t1hRFCWHBlOESjiBWs7vq4m1bq_ml0dZ6pgEDfpsWNpBaLRzwZwOO1mD4LqO4aEokh6uHql0RmmtbfoBF25r7QOyIgqNv0TBf6mlwS3bjE0K6sl08M1mMPjW1PB9e0Qr494H8oBTAB", 'client_key': '214c9979', } sig = cls.calculate_sig(data) data['sig'] = sig response = requests.post(url=url, headers=headers, data=data) body = response.content.decode() return body """ 获取观众画像 """ @classmethod def analyze_photo(cls, photo_id): url = 'https://creator-app.kuaishou.com/rest/bamboo/inspiration/n/photo/analysis' headers = { 'Accept-Language': 'zh-cn', 'Connection': 'keep-alive', 'Content-Type': 'application/x-www-form-urlencoded', 'Host': 'creator-app.kuaishou.com', } data = { 'photoId': photo_id, 'client_key': '214c9979', } sig = cls.calculate_sig(data) data['sig'] = sig response = requests.post(url=url, headers=headers, data=data) body = response.content.decode() json_body = json.loads(body) user_range = json_body['data']['play']['userRange'] if len(user_range) == 0: return False, "无画像" age_range = user_range['ageRange'] value = age_range[5]['value'] value = int(value.strip('%')) if value >= 40: return False, value else: return True, value """ 视频时长转换成秒 """ @classmethod def milliseconds_to_seconds(cls, milliseconds): seconds = milliseconds / 1000 return int(seconds) """ 判断当前视频是否在90天内 """ @classmethod def get_video_data(cls, timestamp_str): timestamp = datetime.strptime(timestamp_str, "%Y-%m-%d %H:%M:%S") # 获取当前时间 current_time = datetime.now() difference = current_time - timestamp if difference <= timedelta(days=90): return False else: return True """ 获取票圈ID """ @classmethod def get_id_by_category(cls, category_name): category_list = [ {"id": 71502003, "category": "生活"}, {"id": 71502004, "category": "才艺"}, {"id": 71502005, "category": "时尚"}, {"id": 71502006, "category": "宠物"}, {"id": 71502007, "category": "读书"}, {"id": 71502008, "category": "二次元"}, {"id": 71502009, "category": "家居"}, {"id": 71502010, "category": "数码"}, {"id": 71502011, "category": "搞笑"}, {"id": 71502012, "category": "健康"}, {"id": 71502013, "category": "旅游"}, {"id": 71502014, "category": "美食"}, {"id": 71502015, "category": "美妆"}, {"id": 71502016, "category": "汽车"}, {"id": 71502018, "category": "亲子"}, {"id": 71502019, "category": "情感"}, {"id": 71502020, "category": "三农"}, {"id": 71502021, "category": "摄影"}, {"id": 71502022, "category": "舞蹈"}, {"id": 71502023, "category": "颜值"}, {"id": 71502024, "category": "音乐"}, {"id": 71502025, "category": "影视"}, {"id": 71502026, "category": "短剧"}, {"id": 71502027, "category": "游戏"}, {"id": 71502028, "category": "运动"}, {"id": 71502029, "category": "资讯"}, {"id": 71502030, "category": "人文"} ] for category in category_list: if category['category'] == category_name: return category['id'] return None """ 新生成视频上传到对应账号下 """ @classmethod def insert_piaoquantv(cls, new_video_path, new_title, n_id, cover): url = "https://vlogapi.piaoquantv.com/longvideoapi/crawler/video/send" headers = { 'User-Agent': 'PQSpeed/486 CFNetwork/1410.1 Darwin/22.6.0', 'cookie': 'JSESSIONID=4DEA2B5173BB9A9E82DB772C0ACDBC9F; JSESSIONID=D02C334150025222A0B824A98B539B78', 'referer': 'http://appspeed.piaoquantv.com', 'token': '524a8bc871dbb0f4d4717895083172ab37c02d2f', 'accept-language': 'zh-CN,zh-Hans;q=0.9', 'Content-Type': 'application/x-www-form-urlencoded' } payload = { 'coverImgPath': cover, 'deviceToken': '9ef064f2f7869b3fd67d6141f8a899175dddc91240971172f1f2a662ef891408', 'fileExtensions': 'MP4', 'loginUid': n_id, 'networkType': 'Wi-Fi', 'platform': 'iOS', 'requestId': 'fb972cbd4f390afcfd3da1869cd7d001', 'sessionId': '362290597725ce1fa870d7be4f46dcc2', 'subSessionId': '362290597725ce1fa870d7be4f46dcc2', 'title': new_title, 'token': '524a8bc871dbb0f4d4717895083172ab37c02d2f', 'uid': n_id, 'versionCode': '486', 'versionName': '3.4.12', 'videoFromScene': '1', 'videoPath': new_video_path, 'viewStatus': '1' } encoded_payload = urlencode(payload) response = requests.request("POST", url, headers=headers, data=encoded_payload) data = response.json() code = data["code"] if code == 0: new_video_id = data["data"]["id"] return new_video_id else: return None @classmethod def get_data(cls, channel_id, name): number = 1 list = [] for category_id, category_name in cls.CATEGORY_IDS.items(): try: feed_data = cls.get_feed_list() feed_data = json.loads(feed_data) feeds = feed_data['feeds'] for feed in feeds: photo_id = feed["photo_id"] # 视频ID status = sqlCollect.ks_is_used(photo_id) user_name = feed["user_name"] # 用户名 user_sex = feed["user_sex"] # 性别 F为女，U为男 time_data = feed["time"] # 发布时间 caption = feed["caption"] # 标题 view_count = feed["view_count"] # 浏览数 like_count = feed["like_count"] # 点赞数 share_count = feed["share_count"] # 分享数 duration = feed["duration"] # 时长/秒 duration = cls.milliseconds_to_seconds(duration) main_mv_url = feed["main_mv_url"] # 视频链接 thumbnail_url = feed["thumbnail_url"] # 视频封面 user_id = feed["user_id"] # 用户id非用户主页id log_data = f"user:{user_name},,video_id:{photo_id},,video_url:{main_mv_url},,original_title:{caption},,share_count:{share_count},,view_count:{view_count},,duration:{duration}" AliyunLogger.logging(channel_id, name, user_name, photo_id, "扫描到一条视频", "2001", log_data) value, age = cls.analyze_photo(photo_id) if status: AliyunLogger.logging(channel_id, name, user_name, photo_id, "该视频已改造过", "2001", log_data) continue if value: AliyunLogger.logging(channel_id, name, user_name, photo_id, f"不符合规则:50+年龄占比小于40%，实际占比{age}", "2003", log_data) sqlCollect.insert_ks_data(user_name, user_sex, time_data, caption, view_count, like_count, share_count, duration, main_mv_url, thumbnail_url, user_id, '1', photo_id, category_name, age, oss_object=None, video_uid=None) continue video_percent = '%.4f' % (share_count / view_count) special = float(0.0005) if float(video_percent) < special: AliyunLogger.logging(channel_id, name, user_name, photo_id, "不符合规则:分享/浏览小于0.0005", "2003", log_data) sqlCollect.insert_ks_data(user_name, user_sex, time_data, caption, view_count, like_count, share_count, duration, main_mv_url, thumbnail_url, user_id, '1', photo_id, category_name, age, oss_object=None, video_uid=None) continue if share_count < 100: AliyunLogger.logging(channel_id, name, user_name, photo_id, "不符合规则:分享小于100", "2003", log_data) sqlCollect.insert_ks_data(user_name, user_sex, time_data, caption, view_count, like_count, share_count, duration, main_mv_url, thumbnail_url, user_id, '1', photo_id, category_name, age, oss_object=None, video_uid=None) continue if duration < 30 or duration > 720: AliyunLogger.logging(channel_id, name, user_name, photo_id, "不符合规则:时长不符合规则大于720秒/小于30秒", "2003", log_data) sqlCollect.insert_ks_data(user_name, user_sex, time_data, caption, view_count, like_count, share_count, duration, main_mv_url, thumbnail_url, user_id, '1', photo_id, category_name, age, oss_object=None, video_uid=None) continue sqlCollect.insert_ks_data(user_name, user_sex, time_data, caption, view_count, like_count, share_count, duration, main_mv_url, thumbnail_url, user_id, '0', photo_id, category_name, age, oss_object=None, video_uid=None) all_data = {"video_id": photo_id, "cover": thumbnail_url, "video_url": main_mv_url, "rule": video_percent, "old_title": caption} AliyunLogger.logging(channel_id, name, user_name, photo_id, "符合规则等待改造", "2004", log_data) list.append(all_data) current_time = datetime.now() formatted_time = current_time.strftime("%Y-%m-%d %H:%M:%S") values = [ [category_name, user_name, photo_id, user_sex, caption, view_count, like_count, share_count, duration, main_mv_url, thumbnail_url, user_id, age, '', '', time_data, formatted_time]] Feishu.insert_columns("PlcisKhObhzmBothRutc65sJnph", "8fQxFv", "ROWS", 2, 3) time.sleep(0.5) Feishu.update_values("PlcisKhObhzmBothRutc65sJnph", "8fQxFv", "A3:Z3", values) if len(list) == int(number): Common.logger("ks-czz").info(f"获取快手创作者视频总数:{len(list)}\n") return list time.sleep(5) except Exception as exc: print(f"异常信息: {exc}") return list return list # Example usage: if __name__ == "__main__": KsFeedVideo.get_data(1)