如何避免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还是容易被检测,可以试试这两个方向:
- Scrapy + Playwright:Playwright比Selenium更难被反爬系统识别,Scrapy的框架更适合结构化数据抓取,两者结合可以高效爬取Twitter,同时模拟真实浏览器行为。
- 直接调用Twitter前端API:Twitter网页会调用内部API获取推文数据(比如面向用户的时间线API),你可以用
requests库模拟登录后的headers和cookies,直接请求这个API获取JSON格式的推文数据,这种方式比爬页面元素速度快10倍,反爬风险也更低。
备注:内容来源于stack exchange,提问作者HamidBee
相关产品推荐
相关产品推荐

