调用API查询用户名可用性轮换代理仍遇429速率限制求助

现有代码的核心错误 #

• 代理格式错误：你读取的代理携带换行符，且为列表结构，不符合requests要求的字典格式（需指定http/https协议对应的代理地址），导致代理完全未生效，所有请求仍通过本地IP发出，必然触发速率限制。
• 代理赋值逻辑缺失：调用proxies()方法返回代理后，没有将返回值赋值给请求使用的self.proxy变量，也没有更新session的代理配置，换代理逻辑空转。
• 免费代理有效性差：你使用的公共免费代理多为透明代理，会在请求头中携带你本地真实IP，无法起到隐藏IP的作用，且很多代理本身已经被目标平台拉黑。
• 请求特征单一：仅更换IP没有同步轮换User-Agent等请求头特征，同一请求特征关联多个IP也会被平台风控识别。

修复方案 #

1. 修正代理读取与格式转换逻辑 #

# 读取代理时预处理
with open('external/proxies.txt', 'r') as f:
    proxyList = [line.strip() for line in f if line.strip()]

def proxies(self):
    try:
        proxy_addr = random.choice(self.proxyList)
        # 转换为requests要求的字典格式
        self.proxy = {
            "http": f"http://{proxy_addr}",
            "https": f"http://{proxy_addr}"
        }
        print(f"切换代理: {proxy_addr}")
    except Exception as e:
        print("无可用代理")

2. 完善请求逻辑 #

def checkAccounts(self):
    while not self.usernames.empty():
        name = self.usernames.get()
        url = f"https://public-ubiservices.ubi.com/v3/profiles?nameOnPlatform={name}&platformType=uplay"
        # 每100次请求主动换代理
        if self.checkedCount % 100 == 0:
            self.proxies()
        try:
            # 每个请求加随机延迟，避免请求过于密集
            time.sleep(random.uniform(1,3))
            # 传入正确的代理参数，同时轮换UA
            headers = self.headers.copy()
            headers['User-Agent'] = random.choice(self.ua_list) # 提前准备UA池
            r = requests.get(url, headers=headers, proxies=self.proxy, timeout=10)
            
            if r.status_code == 200:
                if len(r.json()['profiles']) != 0:
                    print(f"{Fore.LIGHTBLACK_EX}[+]{Fore.RESET} Taken:          {name}")
                else:
                    print(f"{Fore.LIGHTBLACK_EX}[+]{Fore.RESET} Available:      {name}")
                    self.availableCount += 1
                self.checkedCount += 1
            elif r.status_code == 429:
                self.errorCount += 1
                print(f"{Fore.LIGHTBLACK_EX}[+]{Fore.RESET} Error:          Rate limited")
                # 429时先等指定时间再换代理重试
                retry_after = int(r.headers.get('Retry-After', 15))
                time.sleep(retry_after)
                self.proxies()
                # 把当前用户名放回队列重试
                self.usernames.put(name)
            else:
                print(f"{Fore.LIGHTBLACK_EX}[+]{Fore.RESET} Error:          Check errors.txt")
                with open('external/errors.txt', "a") as errorFile:
                    errorFile.write(f'{self.currentTime} | Checking error {r.status_code}, Error message: {r.text}\n')
                self.errorCount +=1
        except Exception as e:
            # 代理失效时直接换代理重试
            self.proxies()
            self.usernames.put(name)

3. 额外优化建议 #

• 提前校验代理可用性：正式跑任务前先测试所有代理是否能正常访问目标API，筛掉失效、高延迟的代理，优先使用高匿付费代理，稳定性远高于免费代理。
• 增加代理失效标记：如果某个代理连续返回429或者连接失败，就将其从代理池中临时移除，避免重复使用无效代理。
• 若目标API需要认证，每切换3-5个代理同步更换一次认证token，避免同个账号关联多IP被风控。

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