|
1 | | -import base64 |
2 | | -import hashlib |
| 1 | +""" |
| 2 | +Bookmarks Lookup - X API v2 |
| 3 | +=========================== |
| 4 | +Endpoint: GET https://api.x.com/2/users/:id/bookmarks |
| 5 | +Docs: https://developer.x.com/en/docs/twitter-api/bookmarks/api-reference/get-users-id-bookmarks |
| 6 | +
|
| 7 | +Authentication: OAuth 2.0 (User Context) |
| 8 | +Required env vars: CLIENT_ID, CLIENT_SECRET |
| 9 | +""" |
| 10 | + |
3 | 11 | import os |
4 | | -import re |
5 | 12 | import json |
6 | | -import requests |
7 | | -from requests.auth import AuthBase, HTTPBasicAuth |
8 | | -from requests_oauthlib import OAuth2Session |
| 13 | +from xdk import Client |
| 14 | +from xdk.oauth2_auth import OAuth2PKCEAuth |
9 | 15 |
|
10 | | -# First, you will need to enable OAuth 2.0 in your App’s auth settings in the Developer Portal to get your client ID. |
11 | | -# Inside your terminal you will need to set an enviornment variable |
12 | | -# export CLIENT_ID='your-client-id' |
| 16 | +# The code below sets the client ID and client secret from your environment variables |
| 17 | +# To set environment variables on macOS or Linux, run the export commands below from the terminal: |
| 18 | +# export CLIENT_ID='YOUR-CLIENT-ID' |
| 19 | +# export CLIENT_SECRET='YOUR-CLIENT-SECRET' |
13 | 20 | client_id = os.environ.get("CLIENT_ID") |
14 | | - |
15 | | -# If you have selected a type of App that is a confidential client you will need to set a client secret. |
16 | | -# Confidential Clients securely authenticate with the authorization server. |
17 | | - |
18 | | -# Inside your terminal you will need to set an enviornment variable |
19 | | -# export CLIENT_SECRET='your-client-secret' |
20 | | - |
21 | | -# Remove the comment on the following line if you are using a confidential client |
22 | | -# client_secret = os.environ.get("CLIENT_SECRET") |
| 21 | +client_secret = os.environ.get("CLIENT_SECRET") |
23 | 22 |
|
24 | 23 | # Replace the following URL with your callback URL, which can be obtained from your App's auth settings. |
25 | | -redirect_uri = "https://www.example.com" |
| 24 | +redirect_uri = "https://example.com" |
26 | 25 |
|
27 | 26 | # Set the scopes |
28 | 27 | scopes = ["bookmark.read", "tweet.read", "users.read", "offline.access"] |
29 | 28 |
|
30 | | -# Create a code verifier |
31 | | -code_verifier = base64.urlsafe_b64encode(os.urandom(30)).decode("utf-8") |
32 | | -code_verifier = re.sub("[^a-zA-Z0-9]+", "", code_verifier) |
33 | | - |
34 | | -# Create a code challenge |
35 | | -code_challenge = hashlib.sha256(code_verifier.encode("utf-8")).digest() |
36 | | -code_challenge = base64.urlsafe_b64encode(code_challenge).decode("utf-8") |
37 | | -code_challenge = code_challenge.replace("=", "") |
38 | | - |
39 | | -# Start an OAuth 2.0 session |
40 | | -oauth = OAuth2Session(client_id, redirect_uri=redirect_uri, scope=scopes) |
41 | | - |
42 | | -# Create an authorize URL |
43 | | -auth_url = "https://twitter.com/i/oauth2/authorize" |
44 | | -authorization_url, state = oauth.authorization_url( |
45 | | - auth_url, code_challenge=code_challenge, code_challenge_method="S256" |
46 | | -) |
47 | | - |
48 | | -# Visit the URL to authorize your App to make requests on behalf of a user |
49 | | -print( |
50 | | - "Visit the following URL to authorize your App on behalf of your Twitter handle in a browser:" |
51 | | -) |
52 | | -print(authorization_url) |
53 | | - |
54 | | -# Paste in your authorize URL to complete the request |
55 | | -authorization_response = input( |
56 | | - "Paste in the full URL after you've authorized your App:\n" |
57 | | -) |
58 | | - |
59 | | -# Fetch your access token |
60 | | -token_url = "https://api.x.com/2/oauth2/token" |
61 | | - |
62 | | -# The following line of code will only work if you are using a type of App that is a public client |
63 | | -auth = False |
64 | | - |
65 | | -# If you are using a confidential client you will need to pass in basic encoding of your client ID and client secret. |
66 | | - |
67 | | -# Please remove the comment on the following line if you are using a type of App that is a confidential client |
68 | | -# auth = HTTPBasicAuth(client_id, client_secret) |
69 | | - |
70 | | -token = oauth.fetch_token( |
71 | | - token_url=token_url, |
72 | | - authorization_response=authorization_response, |
73 | | - auth=auth, |
74 | | - client_id=client_id, |
75 | | - include_client_id=True, |
76 | | - code_verifier=code_verifier, |
77 | | -) |
78 | | - |
79 | | -# Your access token |
80 | | -access = token["access_token"] |
81 | | - |
82 | | -# Make a request to the users/me endpoint to get your user ID |
83 | | -user_me = requests.request( |
84 | | - "GET", |
85 | | - "https://api.x.com/2/users/me", |
86 | | - headers={"Authorization": "Bearer {}".format(access)}, |
87 | | -).json() |
88 | | -user_id = user_me["data"]["id"] |
89 | | - |
90 | | -# Make a request to the bookmarks url |
91 | | -url = "https://api.x.com/2/users/{}/bookmarks".format(user_id) |
92 | | -headers = { |
93 | | - "Authorization": "Bearer {}".format(access), |
94 | | - "User-Agent": "BookmarksSampleCode", |
95 | | -} |
96 | | -response = requests.request("GET", url, headers=headers) |
97 | | -if response.status_code != 200: |
98 | | - raise Exception( |
99 | | - "Request returned an error: {} {}".format(response.status_code, response.text) |
| 29 | +def main(): |
| 30 | + # Step 1: Create PKCE instance |
| 31 | + auth = OAuth2PKCEAuth( |
| 32 | + client_id=client_id, |
| 33 | + client_secret=client_secret, |
| 34 | + redirect_uri=redirect_uri, |
| 35 | + scope=scopes |
100 | 36 | ) |
101 | | -print("Response code: {}".format(response.status_code)) |
102 | | -json_response = response.json() |
103 | | -print(json.dumps(json_response, indent=4, sort_keys=True)) |
| 37 | + |
| 38 | + # Step 2: Get authorization URL |
| 39 | + auth_url = auth.get_authorization_url() |
| 40 | + print("Visit the following URL to authorize your App on behalf of your X handle in a browser:") |
| 41 | + print(auth_url) |
| 42 | + |
| 43 | + # Step 3: Handle callback |
| 44 | + callback_url = input("Paste the full callback URL here: ") |
| 45 | + |
| 46 | + # Step 4: Exchange code for tokens |
| 47 | + tokens = auth.fetch_token(authorization_response=callback_url) |
| 48 | + access_token = tokens["access_token"] |
| 49 | + |
| 50 | + # Step 5: Create client |
| 51 | + client = Client(access_token=access_token) |
| 52 | + |
| 53 | + # Step 6: Get authenticated user ID |
| 54 | + user_me = client.users.get_me() |
| 55 | + user_id = user_me.data["id"] |
| 56 | + |
| 57 | + # Step 7: Get bookmarks with automatic pagination |
| 58 | + all_bookmarks = [] |
| 59 | + for page in client.users.get_bookmarks( |
| 60 | + user_id, |
| 61 | + max_results=100, |
| 62 | + tweetfields=["created_at"] |
| 63 | + ): |
| 64 | + all_bookmarks.extend(page.data) |
| 65 | + print(f"Fetched {len(page.data)} bookmarks (total: {len(all_bookmarks)})") |
| 66 | + |
| 67 | + print(f"\nTotal Bookmarks: {len(all_bookmarks)}") |
| 68 | + print(json.dumps({"data": all_bookmarks[:5]}, indent=4, sort_keys=True)) # Print first 5 as example |
| 69 | + |
| 70 | +if __name__ == "__main__": |
| 71 | + main() |
0 commit comments