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, Feishu
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 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):
        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.is_used(photo_id)
                    if status:
                        continue
                    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_status = cls.get_video_data(time_data)
                    if time_status:
                        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=None, oss_object=None, video_uid=None)
                        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:
                        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=None, oss_object=None, video_uid=None)
                        continue
                    value, age = cls.analyze_photo(photo_id)
                    if value:
                        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=None, oss_object=None, video_uid=None)

                        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, caption, pq_id, thumbnail_url)
                        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, video_uid)
                        current_time = datetime.now()
                        formatted_time = current_time.strftime("%Y-%m-%d %H:%M:%S")
                        values = [
                            [category_name, user_name, user_sex, caption, view_count, like_count, share_count, duration,
                             main_mv_url, thumbnail_url, user_id, age, pq_id, video_uid, time_data, formatted_time]]
                        Feishu.insert_columns("PlcisKhObhzmBothRutc65sJnph", "823f74", "ROWS", 1, 2)
                        time.sleep(0.5)
                        Feishu.update_values("PlcisKhObhzmBothRutc65sJnph", "823f74", "A2:Z2", values)
                time.sleep(80)
            except Exception as exc:
                print(f"异常信息: {exc}")
                continue



# Example usage:
if __name__ == "__main__":
    KsFeedVideo.get_data()