Web3Tools/api_tools/3dos/daily_3dos.py

# -*- coding: utf-8 -*-
import os
from httpx import Client
from concurrent.futures import ThreadPoolExecutor, as_completed


def daily_claim_3dos(email, password):
    client = Client(proxies="http://127.0.0.1:7890")

    login_url = "https://api.dashboard.3dos.io/api/auth/login"
    login_headers = {
        "Accept": "application/json, text/plain, */*",
        "Accept-Encoding": "gzip, deflate, br, zstd",
        "Accept-Language": "zh-CN,zh;q=0.9",
        "Content-Type": "application/json",
        "Origin": "https://dashboard.3dos.io",
        "Priority": "u=1, i",
        "Referer": "https://dashboard.3dos.io/",
        "Sec-CH-UA": '"Chromium";v="130", "Google Chrome";v="130", "Not?A_Brand";v="99"',
        "Sec-CH-UA-Mobile": "?0",
        "Sec-CH-UA-Platform": '"Windows"',
        "Sec-Fetch-Dest": "empty",
        "Sec-Fetch-Mode": "cors",
        "Sec-Fetch-Site": "same-site",
        "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/130.0.0.0 Safari/537.36"
    }
    login_payload = {
        "email": email,
        "password": password
    }

    try:
        login_response = client.post(login_url, json=login_payload, headers=login_headers)
    except Exception as e:
        print(f"登录请求失败：{e}")
        return False

    if login_response.status_code == 200:
        # print(f"{email}: 登录成功！")
        login_data = login_response.json()
        # print("登录响应内容：", login_data)

        data = login_data.get("data")
        token = data.get("access_token") or data.get("token")
        token_type = data.get("token_type")
        if not token or not token_type:
            print("登录响应中未找到 Token 或 Token 类型！")
            return False

        authorization_header = f"{token_type} {token}"
    else:
        print("登录失败！")
        print("登录响应状态码：", login_response.status_code)
        print("登录响应内容：", login_response.json())
        return False

    options_url = "https://api.dashboard.3dos.io/api/claim-reward"
    options_headers = {
        "Accept": "*/*",
        "Accept-Encoding": "gzip, deflate, br, zstd",
        "Accept-Language": "zh-CN,zh;q=0.9",
        "Access-Control-Request-Headers": "authorization,cache-control,content-type,expires,pragma",
        "Access-Control-Request-Method": "POST",
        "Origin": "https://dashboard.3dos.io",
        "Priority": "u=1, i",
        "Referer": "https://dashboard.3dos.io/",
        "Sec-Fetch-Dest": "empty",
        "Sec-Fetch-Mode": "cors",
        "Sec-Fetch-Site": "same-site",
        "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/130.0.0.0 Safari/537.36"
    }

    try:
        options_response = client.options(options_url, headers=options_headers)
    except Exception as e:
        print(f"OPTIONS 请求失败：{e}")
        return False

    # print("OPTIONS 请求响应状态码：", options_response.status_code)

    claim_url = "https://api.dashboard.3dos.io/api/claim-reward"
    claim_headers = {
        "Accept": "application/json, text/plain, */*",
        "Authorization": authorization_header,
        "Cache-Control": "no-cache",
        "Content-Type": "application/json",
        "Expires": "0",
        "Origin": "https://dashboard.3dos.io",
        "Pragma": "no-cache",
        "Priority": "u=1, i",
        "Referer": "https://dashboard.3dos.io/",
        "Sec-CH-UA": '"Chromium";v="130", "Google Chrome";v="130", "Not?A_Brand";v="99"',
        "Sec-CH-UA-Mobile": "?0",
        "Sec-CH-UA-Platform": '"Windows"',
        "Sec-Fetch-Dest": "empty",
        "Sec-Fetch-Mode": "cors",
        "Sec-Fetch-Site": "same-site",
        "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/130.0.0.0 Safari/537.36"
    }
    claim_payload = {
        "id": "daily-reward-api"
    }

    try:
        claim_response = client.post(claim_url, json=claim_payload, headers=claim_headers)
    except Exception as e:
        print(f"{email} claim-reward 请求失败：{e}")
        return False

    message = claim_response.json().get("message")

    if claim_response.status_code not in [200, 429]:
        print(f"{email} claim 报错: {message}, 状态码{claim_response.status_code}")
        return False

    result = f"{email}: {message}"
    return result


def main():
    print("开始执行...")

    config_path = os.path.join(os.path.dirname(os.path.abspath(__file__)), "account.txt")
    try:
        with open(config_path, "r", encoding="utf-8") as file:
            account_list = [line.strip() for line in file.readlines()]
    except Exception as e:
        print(f"读取配置失败：{e}")
        return

    with ThreadPoolExecutor(max_workers=len(account_list)) as executor:
        futures = []
        for account in account_list:
            email, password = account.split("|||")
            futures.append(executor.submit(daily_claim_3dos, email, password))

        for future in as_completed(futures):
            result = future.result()
            if result:
                print(result)


if __name__ == "__main__":
    main()