video_rewriting/qpl_channel/qpl_cljx.py

import random
import time
from datetime import datetime

import requests
import json

from common import Feishu, Material
from common.sql_help import sqlCollect


class QplKs:
    @classmethod
    def ks_star_info(cls):
        content_type = [{'美食': '673'}, {'旅游': '679'}, {'音乐': '687'}, {'教育培训': '685'}, {'时政资讯': '692'},
                        {'军事': '701'},{'舞蹈': '688'}, {'生活': '677'},
                        {'母婴亲子': '676'}, {'萌宠': '691'}, {'情感': '696'}, {'搞笑': '695'}, {'人文': '697'},
                        {'三农': '694'}, {'科学与法律': '693'}, {'读书': '689'}, {'奇人异象': '700'}, {'民生资讯': '703'},
                        {'纪实类': '705'}, {'财经投资': '690'}, {'摄影': '704'}, {'艺术文化': '682'}, {'房产家居': '683'},
                        {'时尚': '675'}, {'体育运动': '680'}, {'健身达人': '706'},{'健康医疗': '684'},]
        #{'星座命理': '699'},  {'汽车': '678'},  {'影视娱乐': '702'},{'短剧': '674'},{'高新数码': '686'}
        fans_count = [{10000000: 0}, {5000000: 10000000}, {3000000: 5000000}, {1000000: 3000000}, {100000: 1000000},  {0: 100000}]
        # fans_count = [{5000000: 10000000}, {3000000: 5000000}, {1000000: 3000000}, {100000: 1000000},  {0: 100000}]

        # fans_count = [{0: 100000}]



        # fans_count = [{10000000: 0}, {5000000: 10000000}]
        cookie = Material.get_cookie_data("B3LFsemLuhyJPktxIoMcJM6znSh", "c95862", "磁力聚星")
        url = "https://k.kuaishou.com/rest/web/star/list"
        headers = {
            'Accept': 'application/json',
            'Accept-Language': 'zh-CN,zh;q=0.9',
            'Content-Type': 'application/json',
            'Cookie': cookie,
            'Origin': 'https://k.kuaishou.com',
            'Pragma': 'no-cache',
            'Referer': 'https://k.kuaishou.com/',
            'User-Agent': 'Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/126.0.0.0 Safari/537.36',
        }
        for count in fans_count:
            try:

                (fans_min_num, fans_max_num), = count.items()
                for content in content_type:
                    (content_name, content_id), = content.items()
                    current_page = 1
                    while True:
                        payload_data = {
                            "currentPage": current_page,
                            "pageSize": 20,
                            "starOrderTag":22,
                            "taskType": 1,
                            "marketingGoal": 1,
                            "viewerAgeList": [  # 观众画像 50+
                                {
                                    "from": "0.50",
                                    "to": "1.00",
                                    "queryName": "50+"
                                }
                            ],
                            "userName": "",
                            "vitalityTags": [  # 活跃度 近期短视频活跃
                                "668"
                            ],
                            "contentTagIdList": [  # 内容类型
                                {
                                    "id": str(content_id),
                                    "selectFirstLevelId": True
                                }
                            ],
                            "starTagIds": [  # 高调性
                                # 13
                            ],
                            "fansMinNum": fans_min_num,  # 粉丝数量
                            "fansMaxNum": fans_max_num,  # 粉丝数量
                        }
                        if fans_min_num > 0:
                            payload_data["fansMinNum"] = fans_min_num
                        if fans_max_num > 0:
                            payload_data["fansMaxNum"] = fans_max_num
                        payload = json.dumps(payload_data)

                        time.sleep(random.randint(1, 5))
                        response = requests.request("POST", url, headers=headers, data=payload)
                        if ": 109" in response.text:
                            text = (
                                f"**通知类型**: 磁力聚星cookie过期\n"
                                f"**链接**: https://w42nne6hzg.feishu.cn/sheets/B3LFsemLuhyJPktxIoMcJM6znSh?sheet=c95862\n"
                            )
                            Feishu.finish_bot(text,
                                              "https://open.feishu.cn/open-apis/bot/v2/hook/ba69e1b6-5839-4640-8932-94f53de5a067",
                                              "【 磁力聚星cookie过期 】")
                            time.sleep(360)
                            continue
                        response = response.json()
                        print(f"开始扫描{content},粉丝量{payload_data['fansMinNum']}~{payload_data['fansMaxNum']}")
                        try:
                            result = response["result"]
                            if result == 1:
                                total = response["total"]  # 总条数
                                if total == 0:
                                    print(f"没有扫描到数据{content}")
                                    break
                                star_list = response["starList"]
                                if len(star_list) == 0 or star_list == []:
                                    break
                                for star in star_list:
                                    print(f"扫描到一条数据{content}")
                                    user_id = star["userId"]
                                    star_id = star["starId"]
                                    name = star["name"]  # 用户名
                                    kwai_id = star["kwaiId"]  # 用户名id
                                    gender = star["gender"]  # 性别
                                    fans_number = star["fansNumber"]  # 粉丝数
                                    profile_id = star["profileId"]  # 主页id
                                    id = sqlCollect.select_ks_star_data(profile_id)
                                    if id:
                                        print(
                                            f"重复视频{profile_id},{content},粉丝量{payload_data['fansMinNum']}~{payload_data['fansMaxNum']}")
                                        continue
                                    star_tag_str = star["starTagStr"]  # 内容类型1
                                    industry_tag_str = star["industryTagStr"]  # 内容类型2
                                    photo_expect_play = star["photoExpectPlay"]  # 预期播放量
                                    photo_expect_cpm = star["photoExpectCpm"]  # 预期CPM
                                    photo_interaction_rate = star["photoInteractionRate"]  # 互动率
                                    photo_complete_play_rate = star["photoCompletePlayRate"]  # 完播率
                                    fans_increase_num = star["fansIncreaseNum"]  # 粉丝增长量
                                    fans_increase_rate = star["fansIncreaseRate"]  # 粉丝增长率
                                    sqlCollect.insert_ks_star_data(user_id, star_id, name, kwai_id, gender, fans_number, profile_id, star_tag_str, industry_tag_str, photo_expect_play, photo_expect_cpm, photo_interaction_rate, photo_complete_play_rate, fans_increase_num, fans_increase_rate)
                                    print(f"{star_id}-{profile_id}获取列表入库成功")
                                    cls.ks_star_works_info(star_id, profile_id, cookie)
                                    time.sleep(random.randint(1, 5))
                                    cls.ks_list_portrait_info(star_id, profile_id, cookie)
                                    time.sleep(random.randint(1, 5))
                                current_page += 1
                                if total < 20:
                                    break
                                else:
                                    page = int(total)/20
                                    if current_page > int(page)+1:
                                        break
                                id = sqlCollect.select_ks_star_data(profile_id)
                                if id:
                                    break
                        except Exception as e:
                            print(f"{e}")
                            continue
            except Exception as e:
                print(f"{e}")
                continue

    """
    传播表现
    """
    @classmethod
    def ks_star_works_info(cls, star_id: str, profile_id: str, cookie: str):
        url = "https://k.kuaishou.com/rest/web/star/starWorksInfo"

        payload = json.dumps({
            "starId": star_id,
            "starType": 1
        })
        headers = {
            'Accept': 'application/json',
            'Accept-Language': 'zh-CN,zh;q=0.9',
            'Connection': 'keep-alive',
            'Content-Type': 'application/json',
            'Cookie': cookie,
            'Origin': 'https://k.kuaishou.com',
            'Referer': 'https://k.kuaishou.com/'
        }
        response = requests.request("POST", url, headers=headers, data=payload)
        response = response.json()
        result = response['result']
        if result == 1:
            try:
                star_work_report = response['starWorksInfo']['starWorkReport']
                pplay_median_data = star_work_report['pplayMedianData']  # 播放量中位数
                pplay_median_data_90 = pplay_median_data['90']  # 获取键 '90' 对应的值
                pplay_median_data_30 = pplay_median_data['30']  # 获取键 '30' 对应的值
                pphoto_cnt_data = star_work_report['pphotoCntData']  # 视频数量
                pphoto_cnt_data_90 = pphoto_cnt_data['90']  # 获取键 '90' 对应的值
                pphoto_cnt_data_30 = pphoto_cnt_data['30']  # 获取键 '30' 对应的值
                pavg_video_duration_data = star_work_report['pavgVideoDurationData']  # 平均时长
                # 处理 '90' 键的值
                value_90 = pavg_video_duration_data['90']
                pavg_video_duration_data_90 = f"{float(value_90):.3f}%" if value_90 else "0.000%"
                # 处理 '30' 键的值
                value_30 = pavg_video_duration_data['30']
                pavg_video_duration_data_30 = f"{float(value_30):.3f}%" if value_30 else "0.000%"
                pavg_like_cnt_data = star_work_report['pavgLikeCntData']  # 平均点赞
                star_work_report_90 = pavg_like_cnt_data['90']  # 获取键 '90' 对应的值
                star_work_report_30 = pavg_like_cnt_data['30']  # 获取键 '30' 对应的值
                pavg_share_cnt_data = star_work_report['pavgShareCntData']  # 平均分享
                pavg_share_cnt_90 = pavg_share_cnt_data['90']  # 获取键 '90' 对应的值
                pavg_share_cnt_30 = pavg_share_cnt_data['30']  # 获取键 '30' 对应的值
                sqlCollect.insert_ks_star_works_info(star_id, profile_id, pplay_median_data_90, pplay_median_data_30, pphoto_cnt_data_90, pphoto_cnt_data_30, pavg_video_duration_data_90, pavg_video_duration_data_30, star_work_report_90, star_work_report_30, pavg_share_cnt_90, pavg_share_cnt_30)
                print(f"{star_id}-{profile_id}获取传播表现入库成功")
            except Exception as e:
                print(f"{star_id}-{profile_id}:{e}")
                return
        else:
            print(f"{star_id}-{profile_id}获取传播表现失败")
            return

    """
    受众分析
    """
    @classmethod
    def ks_list_portrait_info(cls, star_id: str, profile_id: str, cookie: str):
        url = "https://k.kuaishou.com/rest/web/star/listPortrait"

        payload = json.dumps({
            "starId": star_id,
            "starType": 1
        })
        headers = {
            'Accept': 'application/json',
            'Accept-Language': 'zh-CN,zh;q=0.9',
            'Connection': 'keep-alive',
            'Content-Type': 'application/json',
            'Cookie': cookie,
            'Origin': 'https://k.kuaishou.com',
            'Referer': 'https://k.kuaishou.com/'
        }
        response = requests.request("POST", url, headers=headers, data=payload)
        response = response.json()
        result = response['result']
        if result == 1:
            viewer_portrait = response['data']['viewerPortrait']
            sex_percentage = viewer_portrait['sexPercentage']  # 性别占比
            try:
                for sex in sex_percentage:
                    sex_type = 'sex_percentage'
                    sex_name = sex['label']
                    value = sex['value']
                    sex_proportion = f"{float(value):.3f}%" if value else "0.000%"
                    sqlCollect.insert_ks_list_portrait_info(star_id, profile_id, sex_type, sex_name, sex_proportion)
                age_percentage = viewer_portrait['agePercentage']  # 年龄占比
                for age in age_percentage:
                    age_type = 'age_percentage'
                    age_name = age['label']
                    value = age['value']
                    age_proportion = f"{float(value):.3f}%" if value else "0.000%"
                    sqlCollect.insert_ks_list_portrait_info(star_id, profile_id, age_type, age_name, age_proportion)
                area_percentage = viewer_portrait['areaPercentage']  # 城市占比
                for area in area_percentage:
                    area_type = 'area_percentage'
                    area_name = area['label']
                    value = area['value']
                    area_proportion = f"{float(value):.3f}%" if value else "0.000%"
                    sqlCollect.insert_ks_list_portrait_info(star_id, profile_id, area_type, area_name, area_proportion)
                mobile_brand_percentage = viewer_portrait['mobileBrandPercentage']  # 设备占比
                for brand in mobile_brand_percentage:
                    brand_type = 'mobile_brand_percentage'
                    brand_name = brand['label']
                    value = brand['value']
                    brand_proportion = f"{float(value):.3f}%" if value else "0.000%"
                    sqlCollect.insert_ks_list_portrait_info(star_id, profile_id, brand_type, brand_name, brand_proportion)
            except Exception as e:
                print(f"{star_id}-{profile_id}:{e}")
                return
            mobile_price_percentage = viewer_portrait['mobilePricePercentage']  # 设备价格占比
            for price in mobile_price_percentage:
                price_type = 'mobile_price_percentage'
                price_name = price['label']
                value = price['value']
                price_proportion = f"{float(value):.3f}%" if value else "0.000%"
                sqlCollect.insert_ks_list_portrait_info(star_id, profile_id, price_type, price_name, price_proportion)
            print(f"{star_id}-{profile_id}获取受众分析入库成功")
            return
        else:
            print(f"{star_id}-{profile_id}获取受众分析失败")
            return




if __name__ == '__main__':
    QplKs.ks_star_info()