如何为现有Ubisoft API用户名检测请求代码实现多线程优化

改造注意点 #

• 线程数建议控制在5~20区间，避免触发Ubisoft API的速率限制，可根据实际测试结果调整数值
• 原代码的全局计数器是线程不安全的，多线程场景下直接累加会出现计数丢失，需要加锁保证操作原子性
• 原逻辑中无限重试的写法遇到API持续报错会陷入死循环，建议增加最大重试次数规避该问题
• requests库本身是线程安全的，无需额外适配可直接在多线程中调用

改造后完整代码 #

import requests
import threading
from concurrent.futures import ThreadPoolExecutor, as_completed
from colorama import Fore

# 全局复用请求头，避免每个线程重复生成
HEADERS = {
    'Method':'GET',
    'Authority':'public-ubiservices.ubi.com',
    'referer':'https://lb-prod-account-pdc.ubisoft.com',
    'Ubi-AppId':'c5393f10-7ac7-4b4f-90fa-21f8f3451a04',
    'Authorization': authToken, # 替换为你自己的有效授权令牌
    'User-Agent':'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/90.0.4430.212 Safari/537.36',
    'Ubi-RequestedPlatformType':'uplay'
}

# 计数器与线程锁
count_lock = threading.Lock()
checkedCount = 0
availableCount = 0
# 最大重试次数，可按需调整
MAX_RETRY = 5

# 单个用户名检测逻辑抽为独立函数
def check_single_name(name):
    global checkedCount, availableCount
    url = f"https://public-ubiservices.ubi.com/v3/profiles?nameOnPlatform={name}&platformType=uplay"
    retry_cnt = 0
    while retry_cnt < MAX_RETRY:
        r = requests.get(url, headers=HEADERS)
        if r.status_code == 200:
            break
        retry_cnt += 1
        # 若经常触发限流可打开下方注释，重试前等待1秒降低请求频率
        # import time
        # time.sleep(1)
    else:
        print(f"{Fore.RED}[!]{Fore.RESET} {name} 检测失败，重试次数耗尽")
        with count_lock:
            checkedCount += 1
        return False
    
    is_available = not r.json()['profiles']
    # 计数操作加锁保证线程安全
    with count_lock:
        checkedCount += 1
        if is_available:
            availableCount += 1
    if is_available:
        print(f"{Fore.CYAN}[$]{Fore.RESET} {name} 可用")
    else:
        print(f"{Fore.CYAN}[$]{Fore.RESET} {name} 不可用")
    return is_available

def check(thread_num=10):
    # thread_num为线程池大小，可按需调整
    with ThreadPoolExecutor(max_workers=thread_num) as executor:
        # 批量提交所有检测任务
        tasks = [executor.submit(check_single_name, name) for name in usernameList]
        # 等待所有任务执行完成
        for future in as_completed(tasks):
            pass
    # 输出最终统计结果
    print(f"{Fore.GREEN}[*]{Fore.RESET} 全部检测完成，共检测{checkedCount}个用户名，可用{availableCount}个")

调用方式 #

直接调用check()函数即可，需要调整线程数时传入对应参数即可，例如开15个线程就调用check(thread_num=15)。如果不想用全局变量，也可以收集所有check_single_name的返回值最后统一统计，就不需要加锁逻辑。

内容的提问来源于stack exchange，提问作者user16427319