403 Forbidden 453 - 即使我购买了 X 基本付费计划,您目前也可以访问 Twitter API V2 端点的子集和有限的 v1.1 端点

问题描述 投票:0回答:1

我有一个 Python 代码,用于检查 Twitter 的点赞、转发和评论。我购买了基本计划(100 美元/月),即使我有权访问这些端点,它仍然给我这个错误。这是我的代码:

import tweepy
import threading
import time
import re
from telegram import Update, ChatPermissions
from telegram.ext import Updater, CommandHandler, MessageHandler, Filters, ConversationHandler, CallbackContext
import logging

# Twitter API credentials
CONSUMER_KEY = 'API'
CONSUMER_SECRET = 'API'
ACCESS_TOKEN = 'API'
ACCESS_TOKEN_SECRET = 'API'

# Initialize Twitter API
auth = tweepy.OAuthHandler(CONSUMER_KEY, CONSUMER_SECRET)
auth.set_access_token(ACCESS_TOKEN, ACCESS_TOKEN_SECRET)
twitter_api = tweepy.API(auth)

# Set up logging
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

# Bot token from BotFather
TOKEN = ''

# States for conversation handler
LINK, LIKES, REPLIES, REPOSTS = range(4)

# Command handlers
def start(update: Update, context: CallbackContext):
    update.message.reply_text("Welcome to the Tweet Raider!")

def tweet(update: Update, context: CallbackContext):
    if is_user_admin(update, context):
        update.message.reply_text("Please send the Twitter post link.")
        return LINK
    else:
        update.message.reply_text("Sorry, you must be an admin to use this command.")
        return ConversationHandler.END

def link_handler(update: Update, context: CallbackContext):
    context.user_data['link'] = update.message.text
    update.message.reply_text("How many likes should it be?")
    return LIKES

def likes_handler(update: Update, context: CallbackContext):
    context.user_data['likes'] = int(update.message.text)
    update.message.reply_text("How many replies should it be?")
    return REPLIES

def replies_handler(update: Update, context: CallbackContext):
    context.user_data['replies'] = int(update.message.text)
    update.message.reply_text("How many reposts should it be?")
    return REPOSTS

def reposts_handler(update: Update, context: CallbackContext):
    context.user_data['reposts'] = int(update.message.text)
    update.message.reply_text("Tracking started. I will lock the chat now.")
    lock_chat(update, context, True)

    # Start a background thread to monitor the Twitter post
    thread = threading.Thread(target=monitor_twitter_post, args=(context.user_data['link'], context.user_data['likes'], context.user_data['replies'], context.user_data['reposts'], update, context))
    thread.start()

    return ConversationHandler.END

def cancel(update: Update, context: CallbackContext):
    update.message.reply_text('Operation cancelled. Chat is now open.')
    lock_chat(update, context, False)
    return ConversationHandler.END

def lock_chat(update: Update, context: CallbackContext, lock: bool):
    permissions = ChatPermissions(can_send_messages=not lock)
    context.bot.set_chat_permissions(chat_id=update.effective_chat.id, permissions=permissions)

last_update_message_id = None

def monitor_twitter_post(tweet_url, target_likes, target_replies, target_reposts, update, context):
    global last_update_message_id
    tweet_id = extract_tweet_id(tweet_url)
    image_url = 'URL_TO_MEDIA'  #mediaurl

    while True:
        retweets, likes = get_tweet_metrics(tweet_id)
        replies = get_replies_count(tweet_id, twitter_api)
        # Prepare status symbols based on goals
        likes_status = '✅' if likes >= target_likes else '❌'
        replies_status = '✅' if replies >= target_replies else '❌'
        reposts_status = '✅' if retweets >= target_reposts else '❌'
        # Send update to chat with media
        if likes >= target_likes and replies >= target_replies and retweets >= target_reposts:
            context.bot.send_photo(chat_id=update.effective_chat.id, photo=image_url, 
                                   caption="Great job Tweeters! Unlocking the chat.")
            lock_chat(update, context, False)
            break
        else:
            context.bot.send_photo(chat_id=update.effective_chat.id, photo=image_url, 
                                   caption=f"Locking chat until the tweet has {target_likes} likes, {target_replies} replies, and {target_reposts} reposts.\n\n"
                                           f"{likes_status} Current Likes: {likes} | 🎯 {target_likes}\n"
                                           f"{replies_status} Current Replies: {replies} | 🎯 {target_replies}\n"
                                           f"{reposts_status} Current Reposts: {retweets} | 🎯 {target_reposts}\n\n"
                                           f"Check the tweet here:\n{tweet_url}")
        # Delete last updated message
        if last_update_message_id:
            try:
                context.bot.delete_message(chat_id=update.effective_chat.id, message_id=last_update_message_id)
            except Exception as e:
                logger.error(f"Error deleting message: {e}")

        # Send new updated message
        new_message = context.bot.send_photo(chat_id=update.effective_chat.id, photo=image_url, 
                                             caption=f"Your caption here")
        last_update_message_id = new_message.message_id
    
    time.sleep(3600)  # 3600 = 1 Hour
            
def is_user_admin(update: Update, context: CallbackContext) -> bool:
    chat_id = update.effective_chat.id
    user_id = update.effective_user.id
            
    try:
        chat_administrators = context.bot.get_chat_administrators(chat_id)
        return any(admin.user.id == user_id for admin in chat_administrators)
    except Exception as e:
        logger.error(f"Error checking if user is admin: {e}")
        return False
            
def extract_tweet_id(tweet_url):
    match = re.search(r'/status/(\d+)', tweet_url)
    return match.group(1) if match else None
            
def get_tweet_metrics(tweet_id):
    try:
        tweet = twitter_api.get_status(tweet_id)
        retweets = tweet.retweet_count
        likes = tweet.favorite_count
        return retweets, likes
    except Exception as e:
        logger.error(f"Error fetching tweet metrics: {e}")
        return 0, 0
            
def get_replies_count(tweet_id, twitter_api):
    original_tweet = twitter_api.get_status(tweet_id)
    user_screen_name = original_tweet.user.screen_name
    query = f"to:{user_screen_name}"
    replies_count = 0
            
    try:
        for page in tweepy.Cursor(twitter_api.search, q=query, since_id=tweet_id, tweet_mode='extended').pages():
            for tweet in page:
                if hasattr(tweet, 'in_reply_to_status_id_str'):
                    if tweet.in_reply_to_status_id_str == tweet_id:
                        replies_count += 1
    except Exception as e:
        logger.error(f"Error fetching replies: {e}")
            
    return replies_count
            
# Main function to set up the bot           
def main():
    updater = Updater(TOKEN, use_context=True)
    dp = updater.dispatcher
    dp.add_handler(CommandHandler('start', start))

    conv_handler = ConversationHandler(
        entry_points=[CommandHandler('tweet', tweet)],
        states={
            LINK: [MessageHandler(Filters.text & ~Filters.command, link_handler)],
            LIKES: [MessageHandler(Filters.text & ~Filters.command, likes_handler)],
            REPLIES: [MessageHandler(Filters.text & ~Filters.command, replies_handler)],
            REPOSTS: [MessageHandler(Filters.text & ~Filters.command, reposts_handler)],
        },
        fallbacks=[CommandHandler('cancel', cancel)]
    )
    dp.add_handler(conv_handler)

    updater.start_polling()
    updater.idle()

if __name__ == '__main__':
    main()

基本上我购买了 X Basic 来访问这些端点,但我仍然收到该错误:

403 Forbidden 453 - 您当前可以访问 Twitter API V2 端点的子集和有限的 v1.1 端点,即使我购买了 X 基本付费计划

python tweepy twitter-oauth
1个回答
1
投票

您的代码使用的是 Twitter APIv1 兼容的 API 对象,而不是 Twitter APIv2 兼容的 Client 对象。

v1 Twitter API 已失效。除非您向 Twitter 支付很多钱以通过企业帐户继续访问它,否则您的 Tweepy 代码应该使用 v2 客户端。

© www.soinside.com 2019 - 2024. All rights reserved.