Created
September 13, 2022 10:35
-
-
Save rougetimelord/04958d86badd32876d105bae8015bcb9 to your computer and use it in GitHub Desktop.
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
import requests, json | |
from requests_oauthlib import OAuth1, OAuth1Session | |
from requests.auth import AuthBase, HTTPBasicAuth | |
class authException(Exception): | |
""" Auth failure """ | |
class clientException(Exception): | |
""" Client failure """ | |
class AuthHandler: | |
def __init__(self, consumer_key, consumer_secret, user_key, user_secret,callback): | |
self.con_key = consumer_key | |
self.con_sec = consumer_secret | |
if user_key == None or user_secret == None: | |
raise authException("Provide a user token") | |
self.user_key = user_key | |
self.user_secret = user_secret | |
self.callback = callback | |
self.oauth = OAuth1Session(consumer_key, client_secret=consumer_secret, | |
callback_uri=self.callback) | |
def apply_auth(self): | |
return OAuth1( | |
self.con_key, client_secret=self.con_secret, | |
resource_owner_key=self.user_key, | |
resource_owner_secret=self.user_secret, decoding=None | |
) | |
class Client: | |
def __init__(self, consumer_key, consumer_secret, user_key=None, user_secret=None, callback=None): | |
self.auth = AuthHandler(consumer_key, consumer_secret, user_key, user_secret, callback) | |
self.headers = {} | |
self.headers["User-Agent"] = f"Requests/{requests.__version__} Tweepy/{tweepy.__version__}" | |
self.host = "https://twitter.com/i/api/1.1/" | |
self.session = requests.Session() | |
def request(self, method, endpoint, id=None): | |
url = f"{self.host}/1.1/{endpoint}.json" | |
auth = self.auth.apply_auth() | |
params = {} | |
if not id == None: | |
params["id"] = id | |
try: | |
resp = self.session.request(method, url, params=params, headers=self.headers, auth=auth) | |
except Exception as e: | |
raise clientException(f"Failed request with: {e}") | |
if not 200 <= resp.status_code < 300: | |
raise clientException(f"Failed with http code {resp.status_code}") | |
self.session.close() | |
return json.loads(resp.text) | |
def pin_tweet(self, id): | |
return self.request("POST", "account/pin_tweet", id=id) | |
def retweet(self, id): | |
return self.request("POST", f"statuses/retweet/{id}") | |
def main(): | |
# TODO: add token loading | |
client = Client(consumer_key, consumer_secret, user_key, user_secret) | |
tweet = input("Tweet link: ") | |
tweet = id.split("/")[-1] | |
resp = client.retweet(tweet) | |
client.pin_tweet(resp["id"]) |
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment