You need to enable JavaScript to run this app.
优惠活动
大模型
产品
解决方案
定价
更多

如何避免Twitter在爬虫刷新页面时拦截,同时优化Selenium或替代方案实现精准推文抓取?

如何避免Twitter在爬虫刷新页面时拦截,同时优化Selenium或替代方案实现精准推文抓取?

我来帮你一步步解决这个Twitter爬取的问题,你的需求很明确,咱们从最头疼的拦截问题和推文过滤的精准性开始拆解:

一、先解决推文过滤与去重的核心问题

你的当前脚本用推文文本去重+CSS选择器过滤的方式,确实容易误判和重复打印,咱们换更可靠的方案:

1. 精准识别并跳过 pinned 推文、转推

别只依赖单个CSS选择器,结合多个特征避免误判:

  • Pinned推文:除了找svg[aria-label="Pinned Tweet"],还可以检查推文容器里的专属标记(比如部分pinned推文的父元素会带data-testid="pinned"),或者看推文顶部是否有“Pinned”的文本标签,多重验证减少误判。
  • 转推(Retweet):直接检查推文的社交上下文元素[data-testid="socialContext"],如果里面包含“Retweeted”字样就判定为转推;同时搭配检查带有data-testid="retweet"的按钮元素,双重验证更稳妥。

2. 绝对避免重复打印推文

用推文唯一ID代替文本作为去重依据,这是最靠谱的方式:
每个推文都有唯一的ID,藏在推文的链接里(比如https://twitter.com/FabrizioRomano/status/123456789里的123456789)。你可以从推文元素的a标签href中提取这个ID,把ID存入集合,后续只要ID在集合里,不管推文怎么重现都跳过,彻底解决重复打印的问题。

二、解决Twitter页面拦截的关键措施

你遇到的“Something went wrong”本质是Twitter的反爬机制检测到了异常行为,咱们从行为模拟、工具配置两个方向优化:

1. 彻底改掉“频繁刷新页面”的操作

刷新页面是反爬检测的重灾区!换成滚动加载新内容的方式:
在同一个页面里,用JS代码滚动到页面底部,触发Twitter的动态加载,而不是每次调用driver.refresh()。示例代码:

self.driver.execute_script("window.scrollTo(0, document.body.scrollHeight);")
# 等待新内容加载完成
WebDriverWait(self.driver, 10).until(
    EC.presence_of_element_located((By.CSS_SELECTOR, 'article[role="article"]:last-child'))
)

2. 模拟更真实的人类行为

  • 拉长随机等待间隔:把固定的5秒刷新改成15-30分钟的随机间隔(毕竟你是抓最新推文,没必要刷这么勤),短等待也用random.uniform(5, 10)代替固定值,避免机械的时间规律。
  • 模拟人类打字/操作:登录输入账号密码时,不要一次性send_keys,拆成单个字符输入并加随机延迟:
    def type_slowly(self, element, text):
        for char in text:
            element.send_keys(char)
            time.sleep(random.uniform(0.1, 0.3))
    
    # 调用示例
    self.type_slowly(email_field, email)
    
  • 用Chrome新headless模式:放弃旧的--headless,改用--headless=new,新版headless模式更贴近真实浏览器的行为,不容易被检测。

3. 强化反爬规避配置

  • 旋转代理+随机UA:固定代理和UA很容易被标记,用fake_useragent库生成随机UA,搭配付费的旋转代理池(每次启动爬虫换一个新代理),避免IP被拉黑。
  • 复用登录会话:每次重启爬虫都重新登录会增加检测风险,你可以把登录后的cookies保存到本地文件,下次启动时直接加载cookies,跳过登录流程:
    # 登录成功后保存cookies
    import json
    with open("twitter_cookies.json", "w") as f:
        json.dump(self.driver.get_cookies(), f)
    
    # 下次启动时加载cookies
    self.driver.get("https://twitter.com")
    with open("twitter_cookies.json", "r") as f:
        cookies = json.load(f)
        for cookie in cookies:
            self.driver.add_cookie(cookie)
    self.driver.refresh()
    

三、优化后的Selenium脚本核心代码片段

我给你调整了关键部分的代码,重点优化了过滤、去重和反爬逻辑:

import undetected_chromedriver as uc
from selenium.webdriver.common.by import By
from selenium.webdriver.support.ui import WebDriverWait
from selenium.webdriver.support import expected_conditions as EC
import configparser
import time
import random
from fake_useragent import UserAgent

class TwitterScraper:
    def __init__(self, proxy=None):
        options = uc.ChromeOptions()
        # 用新版headless模式
        options.add_argument("--headless=new")
        # 随机生成UA
        ua = UserAgent()
        options.add_argument(f"user-agent={ua.random}")
        
        if proxy:
            options.add_argument(f"--proxy-server={proxy}")
        
        # 精简必要的浏览器配置
        options.add_argument("--disable-extensions")
        options.add_argument("--no-sandbox")
        options.add_argument("--disable-dev-shm-usage")
        options.add_argument("--lang=en-US")
        options.add_argument("--window-size=1920,1080")

        self.driver = uc.Chrome(options=options)
        # 用推文ID做去重依据
        self.printed_tweet_ids = set()

    def type_slowly(self, element, text):
        # 模拟人类打字速度
        for char in text:
            element.send_keys(char)
            time.sleep(random.uniform(0.1, 0.3))

    def login(self):
        config = configparser.ConfigParser()
        config.read(r"C:\Users\Gaming\Documents\Python Tweets\Account.ini")
        email = config.get("x", "email")
        username_value = config.get("x", "username")
        password_value = config.get("x", "password")

        self.driver.get("https://twitter.com/i/flow/login")
        time.sleep(random.uniform(3, 5))

        # 慢输入邮箱
        email_field = WebDriverWait(self.driver, 15).until(
            EC.visibility_of_element_located((By.CSS_SELECTOR, 'input[autocomplete="username"]'))
        )
        self.type_slowly(email_field, email)
        email_field.send_keys("\n")
        time.sleep(random.uniform(2, 4))

        # 慢输入用户名(如果需要)
        try:
            username_field = WebDriverWait(self.driver, 5).until(
                EC.visibility_of_element_located((By.CSS_SELECTOR, 'input[data-testid="ocfEnterTextTextInput"]'))
            )
            self.type_slowly(username_field, username_value)
            username_field.send_keys("\n")
            time.sleep(random.uniform(2, 4))
        except:
            print("No username prompt needed.")

        # 慢输入密码
        password_field = WebDriverWait(self.driver, 15).until(
            EC.visibility_of_element_located((By.CSS_SELECTOR, 'input[name="password"]'))
        )
        self.type_slowly(password_field, password_value)
        password_field.send_keys("\n")
        time.sleep(random.uniform(5, 8))
        print("Login successful.")

    def get_recent_tweets(self, num_tweets=3):
        try:
            WebDriverWait(self.driver, 10).until(
                EC.presence_of_element_located((By.CSS_SELECTOR, 'article[role="article"]'))
            )
            tweets = self.driver.find_elements(By.CSS_SELECTOR, 'article[role="article"]')
            valid_tweets = []

            for tweet in tweets:
                # 跳过pinned推文
                is_pinned = len(tweet.find_elements(By.CSS_SELECTOR, 'svg[aria-label="Pinned Tweet"]')) > 0
                if is_pinned:
                    continue

                # 跳过转推
                is_retweet = False
                try:
                    social_context = tweet.find_element(By.CSS_SELECTOR, '[data-testid="socialContext"]').text
                    if "Retweeted" in social_context:
                        is_retweet = True
                except:
                    pass
                # 双重验证转推
                if len(tweet.find_elements(By.CSS_SELECTOR, 'svg[aria-label="Retweet"]')) > 0:
                    is_retweet = True
                if is_retweet:
                    continue

                # 提取推文ID
                try:
                    tweet_link = tweet.find_element(By.CSS_SELECTOR, 'a[href*="/status/"]').get_attribute('href')
                    tweet_id = tweet_link.split("/")[-1]
                except:
                    continue

                # 提取推文文本
                try:
                    tweet_text = tweet.find_element(By.CSS_SELECTOR, 'div[data-testid="tweetText"]').text.strip()
                except:
                    continue

                valid_tweets.append((tweet_id, tweet_text))

            # 只取最新的指定数量推文
            valid_tweets = valid_tweets[:num_tweets]
            return valid_tweets
        except Exception as e:
            print(f"Error fetching tweets: {e}")
            return []

    def navigate_to_page(self, username):
        try:
            print(f"Navigating to @{username}'s Twitter page...")
            user_url = f"https://twitter.com/{username}"
            self.driver.get(user_url)
            time.sleep(random.uniform(3, 5))
        except Exception as e:
            print(f"Error navigating to @{username}'s page: {e}")

    def start(self, username):
        self.login()
        self.navigate_to_page(username)

        while True:
            recent_tweets = self.get_recent_tweets(num_tweets=3)

            for tweet_id, tweet_text in recent_tweets:
                if tweet_id not in self.printed_tweet_ids:
                    print(f"[Tweet ID: {tweet_id}] {tweet_text}")
                    self.printed_tweet_ids.add(tweet_id)

            # 滚动加载新内容,代替刷新页面
            self.driver.execute_script("window.scrollTo(0, document.body.scrollHeight);")
            # 随机等待15-25分钟再检查新内容
            wait_time = random.uniform(900, 1500)  # 转换为分钟是15-25
            print(f"Waiting {int(wait_time/60)} minutes for new tweets...")
            time.sleep(wait_time)

    def quit(self):
        self.driver.quit()

# 运行示例
if __name__ == "__main__":
    scraper = TwitterScraper()
    try:
        scraper.start(username="FabrizioRomano")
    except KeyboardInterrupt:
        print("Exiting...")
        scraper.quit()

四、替代Selenium的更优方案

如果Selenium还是容易被检测,可以试试这两个方向:

  1. Scrapy + Playwright:Playwright比Selenium更难被反爬系统识别,Scrapy的框架更适合结构化数据抓取,两者结合可以高效爬取Twitter,同时模拟真实浏览器行为。
  2. 直接调用Twitter前端API:Twitter网页会调用内部API获取推文数据(比如面向用户的时间线API),你可以用requests库模拟登录后的headers和cookies,直接请求这个API获取JSON格式的推文数据,这种方式比爬页面元素速度快10倍,反爬风险也更低。

备注:内容来源于stack exchange,提问作者HamidBee

相关产品推荐
方舟 Agent Plan

超全模态模型 × Harness 升级,最新支持 Deepseek-V4.1-Flash、GLM-5.3 系列、Doubao-Seedream-5.0-pro、Kimi-K3 (部分), 限时 9.9 元起

最近更新时间:2026.04.15 03:23:38