web3/models/daily_3dos.py

# -*- coding: utf-8 -*-

from odoo import models, fields
from httpx import Client
from concurrent.futures import ThreadPoolExecutor, as_completed
from fake_useragent import UserAgent
from odoo.exceptions import UserError


class Daily3dos(models.Model):
    _name = 'daily.3dos'
    _description = 'Daily 3dos'
    _order = 'id DESC'

    name = fields.Char(string='Name', default=fields.datetime.now())

    state = fields.Selection([
        ('draft', 'Draft'),
        ('done', 'Done'),
    ], string='State', default='draft')

    daily_3dos_line_ids = fields.One2many('daily.3dos.line', 'daily_3dos_id', string='Daily 3dos Lines')

    def btn_execute(self):
        # 获取所有3DOS账户配置
        if self.daily_3dos_line_ids:
            all_3dos_account = []
            for line in self.daily_3dos_line_ids:
                if line.line_state != 'done':
                    all_3dos_account.append(
                        self.env['daily.3dos.config'].search([('account', '=', line.account)], limit=1))
        else:
            all_3dos_account = self.env['daily.3dos.config'].search([('activate', '=', True)])

        if not all_3dos_account:
            raise UserError('先设置3dos账号!')

        # 创建字典, 记录每个账号的领取结果
        save_data = {}

        with ThreadPoolExecutor(max_workers=len(all_3dos_account)) as executor:
            futures = []
            for account in all_3dos_account:
                email, password = account.account, account.password
                futures.append(executor.submit(self.account_handle, email, password))

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

        for email, data in save_data.items():
            message, state = data
            # 判定 line 里面是否 name 是否 等于 email, 如果是, 更新当条数据, 如果不是, 创建数据
            email_exists = any(email == line.account for line in self.daily_3dos_line_ids)
            if email_exists:
                line = self.env['daily.3dos.line'].search([('account', '=', email), ('daily_3dos_id', '=', self.id)])
                line.write({
                    'line_state': state,
                    'message': message,
                })
            else:
                self.env['daily.3dos.line'].create({
                    'daily_3dos_id': self.id,
                    'line_state': state,
                    'account': email,
                    'message': message,
                })

        self.update({'state': 'done'})

    def account_handle(self, email, password):
        save_data = {}
        email = email
        password = password

        ua = UserAgent()
        user_agent = ua.random

        client = Client(proxy="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": user_agent
        }
        login_payload = {
            "email": email,
            "password": password
        }

        try:
            login_response = client.post(login_url, json=login_payload, headers=login_headers)
        except Exception as e:
            save_data[email] = [str(f"登录请求失败：{e}"), 'error']
            return save_data

        if login_response.status_code == 200:
            login_data = login_response.json()
            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:
                save_data[email] = [f"登录响应中未找到 Token 或 Token 类型！", 'error']
                return save_data

            authorization_header = f"{token_type} {token}"
        else:
            save_data[email] = [f"登录失败！状态码：{login_response.status_code}", 'error']
            return save_data

        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": user_agent
        }

        try:
            options_response = client.options(options_url, headers=options_headers)
        except Exception as e:
            save_data[email] = [f"OPTIONS 请求失败：{e}", 'error']
            return save_data

        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": user_agent
        }
        claim_payload = {
            "id": "daily-reward-api"
        }

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

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

        if claim_response.status_code not in [200, 429]:
            save_data[email] = [f"claim 报错: {message}, 状态码{claim_response.status_code}", 'error']
            return save_data

        save_data[email] = [message, 'done']
        print(f"{email}: {message}")
        return save_data


class Daily3dosLine(models.Model):
    _name = 'daily.3dos.line'
    _description = 'Daily 3dos Line'

    daily_3dos_id = fields.Many2one('daily.3dos', string='Daily 3dos')
    line_state = fields.Selection([
        ('draft', 'Draft'),
        ('error', 'Error'),
        ('done', 'Done'),
    ], string='Line State', default='draft')
    account = fields.Char(string='Account')
    password = fields.Char(string='Password')
    message = fields.Char(string='Message')

    def btn_execute_single(self):
        for retry in range(3):
            result = self.daily_3dos_id.account_handle(self.account, self.password)
            print(result)


class Daily3dosConfig(models.Model):
    _name = 'daily.3dos.config'
    _description = 'Daily 3dos Config'

    activate = fields.Boolean(string='Activate', default=True)
    account = fields.Char(string='Account')
    password = fields.Char(string='Password')
    description = fields.Text(string='Description')

    def btn_init_data(self):
        init_data = {
            "jack0210_@hotmail.com": "aaaAAA111!!!",
            "yujieccyj01@hotmail.com": "aaaAAA111!!!",
            "yujieccyj02@hotmail.com": "aaaAAA111!!!",
            "yujieccyj03@hotmail.com": "aaaAAA111!!!",
            "yujieccyj04@hotmail.com": "aaaAAA111!!!",
            "yujieccyj05@hotmail.com": "aaaAAA111!!!",
            "yujieccyj06@hotmail.com": "aaaAAA111!!!",
            "yujieccyj07@hotmail.com": "aaaAAA111!!!",
            "yujieccyj08@hotmail.com": "aaaAAA111!!!",
            "yujieccyj09@hotmail.com": "aaaAAA111!!!",
            "yujieccyj10@hotmail.com": "aaaAAA111!!!"
        }

        for account, password in init_data.items():
            created = self.search([('account', '=', account)], limit=1)
            if not created:
                self.create({
                    'account': account,
                    'password': password
                })