lm_code/test.py

import random
import time
from concurrent.futures import ThreadPoolExecutor

from curl_cffi import requests

from models.ips import Ips

cookies = {
    'PHPSESSID': 'a99b27b7-0c15-324d-49f0-29009b795783',
    'ipsb': 'Dn42KhiATwEbRov7et5J6FBspLaZdXVg',
}

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',
    'dnt': '1',
    'pragma': 'no-cache',
    'referer': 'https://ip.sb/',
    'sec-ch-ua': '"Chromium";v="142", "Microsoft Edge";v="142", "Not_A Brand";v="99"',
    'sec-ch-ua-mobile': '?0',
    'sec-ch-ua-platform': '"Windows"',
    'sec-fetch-dest': 'script',
    'sec-fetch-mode': 'no-cors',
    'sec-fetch-site': 'same-site',
    'user-agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/142.0.0.0 Safari/537.36 Edg/142.0.0.0',
    # 'cookie': 'PHPSESSID=a99b27b7-0c15-324d-49f0-29009b795783; ipsb=Dn42KhiATwEbRov7et5J6FBspLaZdXVg',
}


def main(sql_info):
    proxies = {
        'http': f'socks5://{sql_info.username}:{sql_info.password}@{sql_info.host}:{sql_info.port}',
        'https': f'socks5://{sql_info.username}:{sql_info.password}@{sql_info.host}:{sql_info.port}',
    }

    # socks5://92.113.106.14:30014:t5742:x335757
    for i in range(5):
        try:
            response = requests.get(
                'https://ipv4.ip.sb/addrinfo',
                headers=headers,
                proxies=proxies,
            )

            print(response.json()["address"])

            if response.json()["address"] == sql_info.host:
                sql_info.start = 1
                sql_info.save()

                break


        except:
            time.sleep(random.randint(1, 3))


if __name__ == '__main__':

    with ThreadPoolExecutor(max_workers=100) as executor:
        for i in Ips.select():
            executor.submit(main, i)
            time.sleep(random.randint(1, 3))

# https://x.com/Websea_MY