# 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)