星球慧答（源代码）

由于某人被打击了自信所以我来上传， python 源代码版本

可直接放入青龙面板执行 记得替换参数

不懂得就直接丢给ai吧 这里不解答问题哈

代码

import requests
import hashlib
import time
import datetime
import random
from requests.packages.urllib3.exceptions import InsecureRequestWarning

# 禁用SSL警告
requests.packages.urllib3.disable_warnings(InsecureRequestWarning)

# 核心配置
CONFIG = {
    # 接口地址
    "chaping_url": "https://admin.xqhd.shop/api/adver/chaping",
    "user_info_url": "https://admin.xqhd.shop/api/user/getUserInfo",
    # salt值（参考代码中的值）
    "salt": "4UdoAFDVNsFpdw/BybhpCuhbk/JQapIOirxxG",
    # user_id初始为空，脚本启动后自动获取
    "user_id": "",
    # 设备参数
    "device_brand": "Redmi",
    "device_model": "25102RKBEC",
    "device_id": "d5a805cf803d4826b6e5e3e5970367f6",
    # 完整请求头
    "headers": {
        "Host": "admin.xqhd.shop",
        "Connection": "keep-alive",
        "sec-ch-ua-platform": '"Android"',
        "sec-ch-ua": '"Chromium";v="140", "Not=A?Brand";v="24", "Android WebView";v="140"',
        "sec-ch-ua-mobile": "?1",
        "client-Type": "android",
        "company-code": "28",
        "User-Agent": "Mozilla/5.0 (Linux; Android 16; 25102RKBEC Build/BP2A.250605.031.A3; wv) AppleWebKit/537.36 (KHTML, like Gecko) Version/4.0 Chrome/140.0.7339.207 Mobile Safari/537.36 AgentWeb/5.0.8  UCBrowser/11.6.4.950",
        "Content-Type": "application/json",
        "app-id": "10",
        "token": "6f5524538f0c41cb0a4ee649befbd7e4f18ea56c",
        "Accept": "*/*",
        "Origin": "https://www.xqhd.shop",
        "X-Requested-With": "com.xqhd.fxkj",
        "Sec-Fetch-Site": "same-site",
        "Sec-Fetch-Mode": "cors",
        "Sec-Fetch-Dest": "empty",
        "Referer": "https://www.xqhd.shop/",
        "Accept-Encoding": "gzip, deflate, br, zstd",
        "Accept-Language": "zh-CN,zh;q=0.9,en-US;q=0.8,en;q=0.7"
    },
    # 多广告类型配置
    "ad_types": [
        {
            "networkPlacementId": "31608000009",
            "networkName": "kuaishou",
            "cate": 3,
            "ecpm_min": 4000,
            "ecpm_max": 6000
        },
        {
            "networkPlacementId": "9276955461732500",
            "networkName": "gdt",
            "cate": 2,
            "ecpm_min": 2000,
            "ecpm_max": 4000
        }
    ],
    # 目标金币数
    "target_coins": 6000,
    # 调用间隔（秒）
    "chaping_interval": 30,
    # 请求超时（秒）
    "request_timeout": 10
}

def get_current_time_str():
    """格式化当前时间（日志用）"""
    return datetime.datetime.now().strftime("%Y/%m/%d, %H:%M:%S")

def md5_encrypt(s):
    """MD5加密（大写）"""
    md5_obj = hashlib.md5()
    md5_obj.update(s.encode("utf-8"))
    return md5_obj.hexdigest().upper()

def get_user_id():
    """
    优先调用用户信息接口，获取真实的user_id（对应data.id）
    返回：成功返回user_id字符串，失败返回None
    """
    print(f"\n【{get_current_time_str()}】=== 第一步：获取user_id ===")
    try:
        response = requests.get(
            url=CONFIG["user_info_url"],
            headers=CONFIG["headers"],
            verify=False,
            timeout=CONFIG["request_timeout"],
            allow_redirects=False
        )
        # 打印原始响应（方便排查格式问题）
        print(f"【{get_current_time_str()}】用户信息接口原始响应：{response.text[:500]}")
        
        # 解析JSON，并校验类型
        response_data = response.json()
        if not isinstance(response_data, dict):
            raise Exception(f"响应数据类型错误（预期字典，实际：{type(response_data).__name__}）")
        
        # 校验接口响应是否正常
        if response_data.get("code") != 200:
            raise Exception(f"用户信息接口返回异常：{response_data.get('msg', '未知错误')}")
        
        # 提取user_id（对应接口返回的data.id）
        user_id = response_data.get("data", {}).get("id")
        if not user_id:
            raise Exception("用户信息接口未返回user_id（data.id为空）")
        
        # 转为字符串（拼接sign需要字符串格式）
        user_id_str = str(user_id)
        print(f"✅ 【{get_current_time_str()}】成功获取user_id：{user_id_str}")
        return user_id_str

    except requests.exceptions.Timeout:
        print(f"❌ 【{get_current_time_str()}】获取user_id失败：请求超时")
        return None
    except requests.exceptions.ConnectionError:
        print(f"❌ 【{get_current_time_str()}】获取user_id失败：网络不通")
        return None
    except ValueError:
        print(f"❌ 【{get_current_time_str()}】获取user_id失败：接口返回非JSON格式数据")
        return None
    except Exception as e:
        print(f"❌ 【{get_current_time_str()}】获取user_id失败：{str(e)}")
        return None

def call_chaping_api():
    """调用chaping接口（使用自动获取的user_id）"""
    # 校验user_id是否已获取
    if not CONFIG["user_id"]:
        print(f"\n⚠️ 【{get_current_time_str()}】user_id未获取，跳过chaping接口调用")
        return False

    try:
        # 1. 随机选择广告类型
        ad_type = random.choice(CONFIG["ad_types"])
        print(f"\n【{get_current_time_str()}】=== 本次选择的广告类型 ===")
        print(f"平台：{ad_type['networkName']} | placementId：{ad_type['networkPlacementId']} | ecpm范围：{ad_type['ecpm_min']}-{ad_type['ecpm_max']}")

        # 2. 生成对应范围的随机ecpm（字符串）
        ecpm = str(random.randint(ad_type["ecpm_min"], ad_type["ecpm_max"]))
        print(f"随机ecpm：{ecpm}")

        # 3. 生成实时10位时间戳
        timestamp = str(int(time.time()))
        print(f"实时时间戳：{timestamp}")

        # 4. 拼接sign_str：networkPlacementId + ecpm + timestamp + user_id + salt
        sign_str = (
            ad_type["networkPlacementId"]
            + ecpm
            + timestamp
            + CONFIG["user_id"]  # 使用自动获取的真实user_id
            + CONFIG["salt"]
        )
        print(f"加密字符串：{sign_str}")

        # 5. 生成sign
        sign = md5_encrypt(sign_str)
        print(f"生成sign：{sign}")

        # 6. 构造请求体
        request_body = {
            "type": 1,
            "vied_id": "",
            "ecpm": ecpm,
            "networkName": ad_type["networkName"],
            "networkPlacementId": ad_type["networkPlacementId"],
            "brand": CONFIG["device_brand"],
            "brand_model": CONFIG["device_model"],
            "device_id": CONFIG["device_id"],
            "loadid": "",
            "cate": ad_type["cate"],
            "time": int(timestamp),
            "sign": sign
        }

        # 7. 调用chaping接口
        print(f"\n【{get_current_time_str()}】调用chaping接口...")
        response = requests.post(
            url=CONFIG["chaping_url"],
            headers=CONFIG["headers"],
            json=request_body,
            verify=False,
            timeout=CONFIG["request_timeout"],
            allow_redirects=False
        )
        
        # 打印原始响应（关键：排查格式问题）
        print(f"【{get_current_time_str()}】chaping接口原始响应：{response.text[:500]}")

        # 8. 解析响应并校验类型
        try:
            response_data = response.json()
        except ValueError:
            print(f"❌ 【{get_current_time_str()}】chaping接口返回非JSON格式数据")
            return False
        
        # 校验响应是否为字典
        if not isinstance(response_data, dict):
            print(f"❌ 【{get_current_time_str()}】chaping响应类型错误（预期字典，实际：{type(response_data).__name__}）")
            return False

        # 解析业务响应
        print(f"【{get_current_time_str()}】接口响应：{response_data}")
        if response_data.get("data", {}).get("reason") == "err":
            print(f"⚠️ 【{get_current_time_str()}】业务验证失败（reason=err），请核对参数！")

        return True

    except requests.exceptions.Timeout:
        print(f"\n【{get_current_time_str()}】错误：chaping接口请求超时")
        return False
    except requests.exceptions.ConnectionError:
        print(f"\n【{get_current_time_str()}】错误：chaping接口网络不通")
        return False
    except Exception as e:
        print(f"\n【{get_current_time_str()}】错误：chaping接口调用失败 - {str(e)}")
        return False

def get_user_unclaimed_coins():
    """调用用户信息接口，获取未领取金币"""
    try:
        print(f"\n【{get_current_time_str()}】=== 检查未领取金币 ===")
        response = requests.get(
            url=CONFIG["user_info_url"],
            headers=CONFIG["headers"],
            verify=False,
            timeout=CONFIG["request_timeout"],
            allow_redirects=False
        )
        # 打印原始响应
        print(f"【{get_current_time_str()}】金币查询接口原始响应：{response.text[:500]}")
        
        # 解析JSON并校验类型
        try:
            response_data = response.json()
        except ValueError:
            print(f"❌ 【{get_current_time_str()}】金币查询接口返回非JSON格式数据")
            return 0
        
        if not isinstance(response_data, dict):
            print(f"❌ 【{get_current_time_str()}】金币查询响应类型错误（预期字典，实际：{type(response_data).__name__}）")
            return 0
        
        if response_data.get("code") != 200:
            raise Exception(f"接口返回异常：{response_data.get('msg', '未知错误')}")
        
        unclaimed_coins = response_data.get("data", {}).get("unclaimed_coins", 0)
        print(f"当前未领取金币：{unclaimed_coins} / 目标6000")
        return unclaimed_coins

    except Exception as e:
        print(f"\n❌ 【{get_current_time_str()}】获取金币失败：{str(e)}")
        return 0

def main():
    """主流程：先获取user_id → 再循环调用chaping → 查金币"""
    # 第一步：强制获取user_id，失败则终止脚本
    user_id = get_user_id()
    if not user_id:
        print(f"\n🛑 【{get_current_time_str()}】user_id获取失败，脚本终止！")
        return
    
    # 更新CONFIG中的user_id
    CONFIG["user_id"] = user_id

    # 第二步：启动循环调用逻辑
    print(f"\n🚀 【{get_current_time_str()}】脚本正式启动！使用user_id：{user_id}")
    print(f"目标金币：{CONFIG['target_coins']} | 调用间隔：{CONFIG['chaping_interval']}秒")
    
    while True:
        # 调用chaping接口
        call_chaping_api()
        
        # 检查未领取金币
        current_coins = get_user_unclaimed_coins()
        
        # 金币达标则停止
        if current_coins >= CONFIG["target_coins"]:
            print(f"\n🎉 【{get_current_time_str()}】金币达标（{current_coins}≥6000），脚本停止！")
            break
        
        # 未达标，间隔指定时间后重试
        print(f"\n⏳ 【{get_current_time_str()}】{CONFIG['chaping_interval']}秒后再次调用接口...")
        time.sleep(CONFIG["chaping_interval"])

if __name__ == "__main__":
    try:
        main()
    except KeyboardInterrupt:
        print(f"\n🛑 【{get_current_time_str()}】脚本被手动终止！")
    except Exception as e:
        print(f"\n❌ 【{get_current_time_str()}】脚本运行异常：{str(e)}")