lm_code/test2.py

import time
import random

import datetime
from loguru import *
from curl_cffi import requests

from models.combined_table import CombinedTable
from models.concrete_wallet import ConcreteWallet
from models.ips import Ips
from tools import get_evm_wallet_singed_message


class Credentials:
    def __init__(self, sql_info):

        self.sql_info = sql_info

        self.headers = {
            'accept': '*/*',
            'accept-language': 'zh-CN,zh;q=0.9,en;q=0.8,en-GB;q=0.7,en-US;q=0.6',
            'cache-control': 'no-cache',
            'content-type': 'application/json',
            'pragma': 'no-cache',
            'priority': 'u=1, i',
            'referer': 'https://points.concrete.xyz/home',
            'sec-ch-ua': '"Microsoft Edge";v="143", "Chromium";v="143", "Not A(Brand";v="24"',
            'sec-ch-ua-mobile': '?0',
            'sec-ch-ua-platform': '"Windows"',
            'sec-fetch-dest': 'empty',
            'sec-fetch-mode': 'cors',
            'sec-fetch-site': 'same-origin',
            'user-agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/143.0.0.0 Safari/537.36 Edg/143.0.0.0',
            # 'cookie': 'client-season=z2zi-tzc2; domain=https%3A%2F%2Fpoints.concrete.xyz; __Host-authjs.csrf-token=afc782765e3380ca973101d02f54c9516d317120fe83faea733390bfc2e46cb4%7Cdca5d6028cadbe52f05820005ed29862e6bd8fff72ae5514142c25bf0d0ad97f; __Secure-authjs.callback-url=https%3A%2F%2Fboost.absinthe.network; redirect-pathname=%2Fhome',
        }

        self.session = requests.Session()
        self.session.headers.update(self.headers)

    def csrf(self):
        for i in range(3):
            try:
                response = self.session.get('https://points.concrete.xyz/api/auth/csrf', )

                return response.json()["csrfToken"]
            except:
                time.sleep(random.randint(1, 5))

        return False

    def credentials(self, message, signature, csrfToken):
        params = ''

        data = {
            'message': message,
            'redirect': 'false',
            'signature': "0x" + signature,
            'csrfToken': csrfToken,
            'callbackUrl': 'https://points.concrete.xyz/home',
        }

        for i in range(3):
            try:
                response = self.session.post(
                    'https://points.concrete.xyz/api/auth/callback/credentials',
                    params=params,
                    data=data,
                )
                print(response.json())

                return True
            except:
                time.sleep(random.randint(1, 5))

        return False

    def add_ips_to_combined_table(self, ):
        # 获取 ips 表中的所有记录
        all_ips_records = Ips.select()
        for ips_record in all_ips_records:
            # 检查该记录是否已经存在于 CombinedTable 中
            existing_record = CombinedTable.get_or_none(
                CombinedTable.host == ips_record.host,
                CombinedTable.port == ips_record.port,
                CombinedTable.username == ips_record.username,
                CombinedTable.pwd == ips_record.pwd
            )
            if not existing_record:
                # 如果记录不存在，则将其添加到 CombinedTable 中
                # 这里假设 mnemonic 字段有默认值或者你可以根据需求传入具体值

                self.com_info.host = ips_record.host
                self.com_info.port = ips_record.port
                self.com_info.username = ips_record.username
                self.com_info.pwd = ips_record.pwd
                self.com_info.save()

                print("成功添加记录到 CombinedTable。")
                return True

        print("ips 表中所有数据在 CombinedTable 中都已存在，结束操作。")
        return False

    def action(self):

        self.com_info, type1 = CombinedTable.get_or_create(
            mnemonic=self.sql_info.mnemonic,
        )

        if not self.com_info.host:
            if self.add_ips_to_combined_table():
                logger.info("添加ip成功！！！")
            else:
                logger.error("没有ip了")
                return

        proxies = {
            'http': f'socks5://{self.com_info.username}:{self.com_info.pwd}@{self.com_info.host}:{self.com_info.port}',
            'https': f'socks5://{self.com_info.username}:{self.com_info.pwd}@{self.com_info.host}:{self.com_info.port}',
        }
        self.session.proxies.update(proxies)

        csrf_token = self.csrf()
        if csrf_token:
            logger.info("获取签名信息成功！！！")
        else:
            logger.error("获取签名信息失败！！！")

        # 获取当前 UTC 时间
        current_utc_time = datetime.datetime.now(datetime.timezone.utc)
        original_msg = f"points.concrete.xyz wants you to sign in with your Ethereum account:\n{self.sql_info.address}\n\nPlease sign with your account\n\nURI: https://points.concrete.xyz\nVersion: 1\nChain ID: 1\nNonce: {csrf_token}\nIssued At: {current_utc_time.strftime("%Y-%m-%dT%H:%M:%S.%f")[:-3] + 'Z'}\nResources:\n- connector://metaMask"
        signature = get_evm_wallet_singed_message(
            original_msg=original_msg,
            private_key=self.sql_info.private_key,

        )

        self.credentials(message=original_msg, signature=signature, csrfToken=csrf_token)


if __name__ == '__main__':
    for concrete_info in ConcreteWallet.select():
        credentials = Credentials(sql_info=concrete_info)
        credentials.action()