403禁用453 - 尽管我购买了X Basic付费计划,但我当前只能访问部分Twitter API V2和有限的v1.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基础套餐是为了能访问这些接口,但我还是收到了这个错误:
403 禁止访问 453 - 尽管我购买了X基础付费套餐,但你目前只能访问Twitter API V2的一部分接口和有限的v1.1接口。