我正在使用免费的 Twitter / X 开发者帐户。
不可否认,OAuth 的东西让我很困惑。我的 .env 中有以下键值(已删除值),但不知道哪些实际上是必要的,哪些不是。
API_KEY=
API_SECRET=
ACCESS_TOKEN=
ACCESS_TOKEN_SECRET=
CLIENT_ID=
CLIENT_SECRET=
BEARER_TOKEN=
以下是一个最小且(半)完整的 Twitter / X Flask 客户端示例。它使用 Tweepy 和 OAuth2.0 通过身份验证 url 和回调流程向 Twitter/X 进行授权,获取授权用户的信息,并获取经过身份验证的用户最近 10 条推文。这些作品按预期工作。
问题是当我尝试使用删除流程(单击示例中的“删除”按钮)时,收到 401 未经授权的错误。
如何纠正此问题以便删除我的推文?
from flask import Flask, jsonify, redirect, request, session, render_template_string
import tweepy
import os
from dotenv import load_dotenv
load_dotenv()
app = Flask(__name__)
app.secret_key = os.getenv("FLASK_SECRET_KEY", "supersecretkey")
CLIENT_ID = os.getenv("CLIENT_ID")
CLIENT_SECRET = os.getenv("CLIENT_SECRET")
CALLBACK_URL = "http://127.0.0.1:5000/callback"
oauth2_handler = tweepy.OAuth2UserHandler(
client_id=CLIENT_ID,
client_secret=CLIENT_SECRET,
redirect_uri=CALLBACK_URL,
scope=["tweet.read", "tweet.write", "users.read", "offline.access"],
)
@app.route("/", methods=["GET"])
def auth_redirect():
try:
auth_url = oauth2_handler.get_authorization_url()
return redirect(auth_url)
except Exception as e:
return jsonify({"error": str(e)}), 500
@app.route("/callback", methods=["GET"])
def callback():
try:
code = request.args.get("code")
if not code:
return jsonify({"error": "Missing authorization code"}), 400
access_token = oauth2_handler.fetch_token(request.url)
session["access_token"] = access_token
client = tweepy.Client(access_token["access_token"], wait_on_rate_limit=True)
user = client.get_me(user_auth=False)
tweets = client.get_users_tweets(id=user.data.id, max_results=10)
tweet_list = (
"".join(
[
f"""
<li>{tweet.text}
<form action="/delete_tweet/{tweet.id}" method="POST">
<button type="submit">Delete</button>
</form>
</li>
"""
for tweet in (tweets.data or [])
]
)
or "<p>No tweets found.</p>"
)
return render_template_string(
f"""
<h1>Authorization Successful</h1>
<h2>User Information</h2>
<ul>
<li><strong>ID:</strong> {user.data.id}</li>
<li><strong>Username:</strong> {user.data.username}</li>
<li><strong>Name:</strong> {user.data.name}</li>
</ul>
<h2>Last 10 Tweets</h2>
<ul>{tweet_list}</ul>
"""
)
except Exception as e:
return jsonify({"error": str(e)}), 500
@app.route("/delete_tweet/<string:tweet_id>", methods=["POST"])
def delete_tweet(tweet_id):
try:
access_token = session.get("access_token")
if not access_token:
return jsonify({"error": "Access token not found"}), 400
client = tweepy.Client(
consumer_key=CLIENT_ID,
consumer_secret=CLIENT_SECRET,
access_token=access_token["access_token"],
wait_on_rate_limit=True,
)
response = client.delete_tweet(tweet_id)
if response.data:
return jsonify({"message": f"Tweet {tweet_id} deleted successfully"})
return jsonify({"error": "Failed to delete tweet"}), 500
except Exception as e:
return jsonify({"error": str(e)}), 500
if __name__ == "__main__":
app.run(debug=True)
这是一个简单的修复。正如我在问题中指出的,Twitter 的 OAuth 路径让我感到困惑,特别是允许/禁止哪个版本(OAuth1.0a 和 OAuth2.0)与哪个版本的 API(v1.1 与 v2.0)一起使用。
修复有两个方面:
tweepy.Client
初始化中,将密钥/秘密/令牌替换为单数 bearer_token
和 delete_tweet
的调用应包括 user_auth=False
。这使得方法
@app.route("/delete_tweet/<string:tweet_id>", methods=["POST"])
def delete_tweet(tweet_id):
try:
access_token = session.get("access_token")
if not access_token:
return jsonify({"error": "Access token not found"}), 400
client = tweepy.Client(
bearer_token=access_token["access_token"],
wait_on_rate_limit=True,
)
response = client.delete_tweet(tweet_id, user_auth=False)
if response.data:
return jsonify({"message": f"Tweet {tweet_id} deleted successfully"})
return jsonify({"error": "Failed to delete tweet"}), 500
except Exception as e:
return jsonify({"error": str(e)}), 500