如何用Tweepy监听特定Twitter账号并触发自动化操作

嘿，我帮你搞定这个问题！你之前的监听器是全局过滤关键词，所以会收到整个平台的内容对吧？其实只需要调整两个核心逻辑：只监听你自己的账号，再检查推文末尾的特殊代码就行，下面是具体实现方案：

步骤1：先获取你的用户ID #

首先得拿到自己的Twitter用户ID，用来在监听器里识别是不是你发的推文。用Tweepy的API就能快速获取：

import tweepy

# 填入你的Twitter开发者配置
cfg = {
    "consumer_key": "你的consumer_key",
    "consumer_secret": "你的consumer_secret",
    "access_token": "你的access_token",
    "access_token_secret": "你的access_token_secret"
}

# 初始化API连接
auth = tweepy.OAuthHandler(cfg["consumer_key"], cfg["consumer_secret"])
auth.set_access_token(cfg["access_token"], cfg["access_token_secret"])
api = tweepy.API(auth)

# 获取你的用户ID和用户名
my_user_id = api.me().id
my_screen_name = api.me().screen_name
print(f"你的用户ID：{my_user_id}，用户名：{my_screen_name}")

运行这段代码就能得到你的用户ID，记下来后面要用。

步骤2：自定义监听器，精准过滤你的推文 #

接下来修改你的监听器逻辑，去掉全局关键词过滤，改成先判断推文作者是不是你，再检查末尾的特殊代码：

import tweepy
import sqlite3  # 示例用SQLite，可替换为你实际使用的数据库

# 你的配置信息
cfg = {
    "consumer_key": "你的consumer_key",
    "consumer_secret": "你的consumer_secret",
    "access_token": "你的access_token",
    "access_token_secret": "你的access_token_secret"
}

# 填入步骤1获取的你的用户ID
MY_USER_ID = 123456789  # 替换成你的实际ID

# 初始化数据库连接（示例）
conn = sqlite3.connect('my_tweets.db')
cursor = conn.cursor()
# 创建存储表（如果不存在）
cursor.execute('''CREATE TABLE IF NOT EXISTS saved_tweets
                  (id INTEGER PRIMARY KEY AUTOINCREMENT, content TEXT, post_time DATETIME)''')
conn.commit()

class MyTweetListener(tweepy.StreamListener):
    def on_status(self, status):
        # 1. 先判断推文是不是你发的，不是就直接跳过
        if status.user.id != MY_USER_ID:
            return True
        
        # 2. 获取完整推文内容（避免长推文被截断）
        tweet_content = status.full_text if hasattr(status, 'full_text') else status.text
        
        # 3. 检查推文末尾是否带有特殊代码（比如“…”）
        special_code = "…"
        if tweet_content.strip().endswith(special_code):
            # 提取前序内容（去掉末尾的特殊代码和多余空格）
            processed_content = tweet_content.strip().rstrip(special_code).strip()
            
            # 存入数据库
            try:
                cursor.execute('''INSERT INTO saved_tweets (content, post_time)
                                  VALUES (?, ?)''', (processed_content, status.created_at))
                conn.commit()
                print(f"成功存入推文：{processed_content}")
            except Exception as e:
                print(f"存入数据库失败：{str(e)}")
                conn.rollback()
        
        return True

    def on_error(self, status_code):
        print(f"监听出错，错误码：{status_code}")
        if status_code == 420:
            # 触发限流，直接断开监听
            return False

# 启动流监听
auth = tweepy.OAuthHandler(cfg["consumer_key"], cfg["consumer_secret"])
auth.set_access_token(cfg["access_token"], cfg["access_token_secret"])
tweet_stream = tweepy.Stream(auth=auth, listener=MyTweetListener())

# 核心：只监听你自己的推文，follow参数传入你的用户ID字符串
tweet_stream.filter(follow=[str(MY_USER_ID)])

关键细节说明 #

• follow=[str(MY_USER_ID)]：这是解决全局监听问题的核心！用这个参数替代原来的关键词过滤，Tweepy只会推送你指定用户（也就是你自己）的所有推文，不用再处理整个平台的内容，效率大幅提升。
• status.full_text：如果你的推文超过140字符，status.text会被截断，所以优先用full_text获取完整内容，记得先判断属性是否存在。
• 特殊代码匹配：用endswith()精准判断末尾字符，rstrip()用来移除末尾的特殊代码，再用strip()清理前后多余空格。
• 数据库适配：示例用了SQLite，你可以换成MySQL、PostgreSQL等，只需要修改数据库连接和对应的SQL语句即可。

额外优化（可选） #

如果你只想处理原创推文（排除转发和回复），可以在on_status方法开头加这两段判断：

# 跳过转发内容
if hasattr(status, 'retweeted_status'):
    return True
# 跳过回复内容
if status.in_reply_to_status_id is not None:
    return True

这样就完美实现你想要的自动化功能啦！

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