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 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", '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", '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 >= 50: return True, value else: return False, 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 """ 生成目录 """ @classmethod def create_folders(cls): video_path_url = "/Users/tzld/Desktop/ks_automation/path_video/" if not os.path.exists(video_path_url): os.makedirs(video_path_url) return video_path_url """ 删除文件 """ @classmethod def remove_files(cls, video_path_url): if os.path.exists(video_path_url) and os.path.isdir(video_path_url): for root, dirs, files in os.walk(video_path_url): for file in files: file_path = os.path.join(root, file) os.remove(file_path) for dir in dirs: dir_path = os.path.join(root, dir) os.rmdir(dir_path) """ 视频下载 """ @classmethod def download_video(cls, video_url, path_url, video_id): for i in range(3): payload = {} headers = {} response = requests.request("GET", video_url, headers=headers, data=payload) if response.status_code == 200: # 以二进制写入模式打开文件 video = path_url + str(video_id) + '.mp4' with open(f"{video}", "wb") as file: # 将响应内容写入文件 file.write(response.content) time.sleep(5) return video return '' """ 获取票圈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): 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 = { '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): for category_id, category_name in cls.CATEGORY_IDS.items(): 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.is_used(photo_id) if status: 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 time_data = cls.get_video_data(time_data) if time_data: continue video_percent = '%.2f' % (share_count / like_count) special = float(0.2) # if float(video_percent) < special or share_count < 2000 or duration < 30 or duration > 6000: # continue value, age = cls.analyze_photo(photo_id) # path_url = cls.create_folders() # 创建目录 # video_path = cls.download_video(main_mv_url, path_url, photo_id) # if not os.path.isfile(video_path): # cls.remove_files(path_url) # continue oss_object_key = Oss.channel_upload_oss(main_mv_url, photo_id) time.sleep(2) oss_object = oss_object_key.get("oss_object_key") pq_id = cls.get_id_by_category(category_name) if pq_id: video_uid = cls.insert_piaoquantv(oss_object_key, caption, pq_id) if video_uid: print(video_uid) return if oss_object: pass if value: pass else: pass # Example usage: if __name__ == "__main__": KsFeedVideo.get_data()