ks_automation/channel_ks/ks_feed_list.py

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