Selenium Wire代理下Twitter爬虫重启会话后连接失败求助

1. Driver实例引用失效
   当触发速率限制后，reset_password内执行driver.quit()关闭了浏览器会话，但重新创建的driver是函数局部变量，process_usernames中持有的仍是已失效的旧实例。下一次循环调用reset_password时，使用的是已关闭的Driver，导致连接ChromeDriver端口被拒绝，出现WinError 10061错误。

2. 代理未真正轮换
   当前create_proxy_options返回固定代理IP（从config导入的proxy），重启浏览器后IP未更换，无法有效绕过Twitter的速率限制。

3. 多线程文件写入不安全
   多个线程同时写入output.txt，会导致内容覆盖或错乱，因为文件操作未加锁控制。

1. 修复Driver引用传递 #

修改reset_password函数，使其返回更新后的driver实例，让process_usernames中的driver变量同步更新。

2. 动态获取轮换代理 #

更新create_proxy_options，每次调用时获取新的代理IP（需根据你的旋转代理服务实现动态获取逻辑）。

3. 添加线程锁保证文件写入安全 #

在全局或模块级别定义线程锁，确保多线程写入文件时互斥执行。

4. 精简冗余代码并增强异常处理 #

移除不必要的auth_tokenJS赋值操作，添加元素查找、页面加载的异常捕获。

修改后的完整代码 #

import re
import threading
import argparse
from seleniumwire import webdriver
from selenium.webdriver.common.by import By
from selenium.webdriver.chrome.options import Options
import time
from config import get_new_proxy  # 替换为你的动态代理获取函数

# 线程锁，用于文件写入安全
file_lock = threading.Lock()

def create_proxy_options():
    # 每次调用获取新的代理IP
    proxy = get_new_proxy()
    return {
        'proxy': {
            'http': proxy,
            'https': proxy,
            'no_proxy': 'localhost,127.0.0.1'
        }
    }

def main():
    parser = argparse.ArgumentParser(description="Twitter 邮箱/号码提示爬取脚本")
    parser.add_argument('--threads', type=int, default=1, help="运行的线程数")
    args = parser.parse_args()

    with open('usernames.txt', 'r', encoding='utf-8') as file:
        usernames = [line.strip() for line in file if line.strip()]

    threads = []
    for i in range(args.threads):
        usernames_per_thread = usernames[i::args.threads]
        thread = threading.Thread(target=process_usernames, args=(usernames_per_thread,))
        threads.append(thread)
        thread.start()

    for thread in threads:
        thread.join()

def process_usernames(usernames):
    driver = None
    try:
        proxy_options = create_proxy_options()
        driver = webdriver.Chrome(seleniumwire_options=proxy_options)
        
        for username in usernames:
            # 接收reset_password返回的新driver实例
            driver = reset_password(driver, username)
            if not driver:
                # 如果返回None，说明当前线程无法继续，退出循环
                break
    except Exception as e:
        print(f"线程发生错误: {e}")
    finally:
        if driver:
            driver.quit()

def reset_password(driver, username):
    while True:
        try:
            driver.get("https://twitter.com/account/begin_password_reset")
            time.sleep(1)  # 增加短暂等待，确保页面加载完成

            # 查找账号输入框并填入用户名
            account_identifier_input = driver.find_element(By.NAME, 'account_identifier')
            account_identifier_input.clear()
            account_identifier_input.send_keys(username)

            # 提交按钮
            submit_button = driver.find_element(By.XPATH, "//input[@type='submit']")
            submit_button.click()
            time.sleep(2)  # 等待页面跳转

            page_source = driver.page_source
            if "Enter your email address to continue" in page_source:
                print(f"用户名 {username} 需要验证，跳过")
                return driver
            elif "You've exceeded the number of attempts" in page_source:
                print(f"用户名 {username} 触发速率限制，重启浏览器换IP")
                driver.quit()
                time.sleep(3)  # 延长等待，确保端口释放
                proxy_options = create_proxy_options()
                driver = webdriver.Chrome(seleniumwire_options=proxy_options)
            else:
                email_hint, number_hint = extract_email_and_number_hints(page_source)
                hint_text = f"{username}:{email_hint or ''}:{number_hint or ''}"
                print(hint_text)
                # 加锁写入文件
                with file_lock:
                    with open('output.txt', 'a', encoding='utf-8') as output_file:
                        output_file.write(hint_text + '\n')
                return driver
        except Exception as e:
            print(f"处理用户名 {username} 时出错: {e}")
            # 出错后重启driver
            driver.quit()
            time.sleep(3)
            proxy_options = create_proxy_options()
            driver = webdriver.Chrome(seleniumwire_options=proxy_options)

def extract_email_and_number_hints(page_source):
    email_hint = re.search(r'Send an email to <strong dir="ltr">(.*?)</strong>', page_source)
    number_hint = re.search(r'Text a code to the phone number ending in <strong dir="ltr">(.*?)</strong>', page_source)

    email_hint_text = email_hint.group(1) if email_hint else None
    number_hint_text = number_hint.group(1) if number_hint else None

    return email_hint_text, number_hint_text

if __name__ == "__main__":
    main()

关键修改说明 #

• Driver引用更新：reset_password返回当前有效的driver实例，process_usernames每次循环都会更新driver变量，避免使用失效实例。
• 动态代理：将原固定proxy替换为get_new_proxy()函数，需你实现该函数从旋转代理服务获取新IP。
• 线程锁：添加file_lock，确保多线程写入output.txt时不会冲突。
• 异常处理增强：在reset_password的核心操作环节添加try-except，出错后自动重启driver。
• 冗余代码移除：删除了不必要的auth_tokenJS赋值操作，原操作无实际意义。

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