ks_automation/channel_ks/ks_analyze_photo.py

import json
from hashlib import md5

import requests
from flask import Flask, request, jsonify

app = Flask(__name__)

# def haiwai_tunnel_proxies():
#     tunnel = "c101.kdlfps.com:18866"
#     # 用户名密码方式
#     username = "f2801246645"
#     password = "q0i0ohnl"
#     proxies = {
#         "http": "http://%(user)s:%(pwd)s@%(proxy)s/" % {"user": username, "pwd": password, "proxy": tunnel},
#         "https": "http://%(user)s:%(pwd)s@%(proxy)s/" % {"user": username, "pwd": password, "proxy": tunnel}
#     }
#     return proxies

def process_range_data(data, label_key, value_key, selected_key=None, tips_key=None):
    processed_data = []
    for item in data:
        processed_item = {label_key: item['label'], value_key: item['value']}
        if selected_key and item['selected'] is not None:
            processed_item[selected_key] = item['selected']
        if tips_key and item['tips'] is not None:
            processed_item[tips_key] = item['tips']
        processed_data.append(processed_item)
    return processed_data


@app.route('/analyze_photo', methods=['GET'])
def analyze_photo():
    photo_id = request.args.get('photo_id')
    if not photo_id:
        return jsonify({"error": "photo_id is required"}), 400

    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',
    }
    query = {}
    data = {
        'photoId': photo_id,
        'client_key': '214c9979',
    }
    src = {}
    src.update(query)
    src.update(data)
    src = ''.join([f'{key}={src[key]}' for key in sorted(src.keys(), reverse=False)])
    salt = '08d8eece8e83'
    sig = md5(f'{src}{salt}'.encode()).hexdigest()
    query.update({'sig': sig})
    response = requests.post(url=url, headers=headers, params=query, data=data)

    if response.status_code != 200:
        return jsonify({"error": "Failed to analyze photo"}), response.status_code

    body = response.content.decode()
    json_body = json.loads(body)

    user_range = json_body['data']['play']['userRange']
    if len(user_range) == 0:
        return jsonify({photo_id: "无观众画像"})
    age_range = user_range['ageRange']
    sex_range = user_range['sexRange']
    province_range = user_range['provinceRange']

    age_range_data = process_range_data(age_range, '年龄范围', '占比', '挑选', '年龄范围占比')
    sex_range_data = process_range_data(sex_range, '性别', '占比', '挑选')
    province_range_data = process_range_data(province_range, '地域', '占比', '挑选')

    result = {
        "年龄范围": [item for item in age_range_data if item],
        "性别占比": [item for item in sex_range_data if item],
        "地域占比": [item for item in province_range_data if item]
    }

    return jsonify(result)


if __name__ == '__main__':
    app.run(host='0.0.0.0', port=8080)