如何使用Tweepy实现指定账号关注新账号的实时通知

Twitter的Stream API（v2版本）确实不支持直接监听用户的关注/取关行为，所以你需要用「定时轮询+关注列表对比」的方案来实现需求，这也是目前这类场景的标准做法。

核心思路 #

1. 定期调用Tweepy的API获取目标用户的关注列表
2. 将最新列表与本地保存的历史列表对比，找出新增的关注账号
3. 针对新增账号触发通知逻辑
4. 更新本地历史列表，继续循环轮询

代码实现示例 #

import tweepy
import json
import time
from datetime import datetime

# 替换为你的Twitter API密钥
BEARER_TOKEN = "your_bearer_token"
# 替换为目标用户的ID（可通过client.get_user(username="xxx")获取）
TARGET_USER_ID = "123456789"
# 轮询间隔（单位：秒，建议≥300，避免触发API限流）
CHECK_INTERVAL = 300

# 初始化Tweepy客户端
client = tweepy.Client(bearer_token=BEARER_TOKEN)

def load_history_follows():
    """加载本地保存的历史关注列表"""
    try:
        with open("follows_history.json", "r", encoding="utf-8") as f:
            return set(json.load(f))
    except FileNotFoundError:
        # 首次运行时创建空集合
        return set()

def save_history_follows(follow_ids):
    """将当前关注列表保存到本地"""
    with open("follows_history.json", "w", encoding="utf-8") as f:
        json.dump(list(follow_ids), f, ensure_ascii=False)

def get_current_follows(user_id):
    """获取目标用户当前的所有关注账号ID"""
    current_follows = set()
    # 处理分页（Twitter API单次最多返回1000条关注）
    for page in tweepy.Paginator(
        client.get_users_following,
        user_id,
        max_results=1000,
        user_fields=["username", "name"]
    ):
        if page.data:
            current_follows.update([user.id for user in page.data])
    return current_follows

def check_new_follows():
    """检查新增关注并触发通知"""
    current_follows = get_current_follows(TARGET_USER_ID)
    history_follows = load_history_follows()

    # 计算新增的关注账号
    new_follows = current_follows - history_follows
    if new_follows:
        print(f"[{datetime.now().strftime('%Y-%m-%d %H:%M:%S')}] 发现新增关注:")
        for user_id in new_follows:
            # 获取新增用户的详细信息
            user_info = client.get_user(id=user_id, user_fields=["username", "name"]).data
            print(f"- 用户名: @{user_info.username} | 昵称: {user_info.name}")
            
            # 这里替换为你的通知逻辑，比如发邮件、Telegram推送等
            # send_notification(user_info)
        
        # 更新历史关注列表
        save_history_follows(current_follows)

# 主循环
if __name__ == "__main__":
    while True:
        check_new_follows()
        time.sleep(CHECK_INTERVAL)

关键注意事项 #

• API限流：Twitter v2 API对get_users_following的调用有次数限制，不要把轮询间隔设置得太短（建议5分钟以上）
• 用户ID获取：如果不知道目标用户的ID，可通过client.get_user(username="目标用户名")获取
• 通知逻辑扩展：你可以把示例中的send_notification函数替换为自己的需求，比如调用Telegram Bot API发送消息、发送邮件等

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