def twittercallback():
verification = request.args["oauth_verifier"]
auth = tweepy.OAuthHandler(CONSUMER_KEY, CONSUMER_SECRET)
try:
auth.request_token = session["request_token"]
except KeyError:
flash("Please login again", "danger")
return redirect(url_for("bp.home"))
try:
auth.get_access_token(verification)
except tweepy.TweepError:
flash("Failed to get access token", "danger")
return redirect(url_for("bp.home"))
session["access_token"] = auth.access_token
session["access_token_secret"] = auth.access_token_secret
return render_template("twittercallback.html", form=HashtagForm())
python类OAuthHandler()的实例源码
def analyzetweets(self, access_token, access_token_secret, mytweets=False, q=None):
auth = tweepy.OAuthHandler(CONSUMER_KEY, CONSUMER_SECRET)
auth.set_access_token(access_token, access_token_secret)
api = tweepy.API(auth)
sentimentlist = []
subjectivitylist = []
number = NUMBER_OF_TWEETS
tweets = tweepy.Cursor(api.user_timeline).items() if mytweets else tweepy.Cursor(api.search, q=q).items(number)
for index, tweet in enumerate(tweets):
analysis = TextBlob(tweet.text).sentiment
sentimentlist.append(analysis.polarity)
subjectivitylist.append(analysis.subjectivity)
self.update_state(state="RUNNING", meta={"current": index + 1, "total": number})
sentimentavg = float(sum(sentimentlist) / max(len(sentimentlist), 1))
subjectivityavg = float(sum(subjectivitylist) / max(len(subjectivitylist), 1))
return {"current": number, "total": number, "subjectivityavg": subjectivityavg, "sentimentavg": sentimentavg}
def __init__(self, credentials, data):
super(Streamer, self).__init__(name='Streamer')
self.data = data
self.text_processing_queue = data['queues']['text_processing']
self.stoprequest = threading.Event()
self.filter_params = data['filters']
self.keyword_queue = data['queues']['keywords']
self.keywords = set()
self.auth = tweepy.OAuthHandler(credentials['consumer_key'],
credentials['consumer_secret'])
self.auth.set_access_token(credentials['access_token'],
credentials['access_token_secret'])
self.limit_queue = data['queues']['limit']
self.message_queue = data['queues']['messages']
self.last_connection = 0
self.min_reconnect_pause = 20
def __init__(self):
date_time_name = datetime.utcnow().strftime("%Y-%m-%d_%H-%M-%S")
logging.basicConfig(filename=date_time_name + '.log', level=logging.INFO)
path = os.path.dirname(os.path.abspath(inspect.getfile(inspect.currentframe())))
self.config = configparser.ConfigParser()
self.config.read(os.path.join(path, "configuration.txt"))
self.sleep_time = int(self.config.get("settings", "time_between_retweets"))
self.search_term = self.config.get("settings", "search_query")
self.tweet_language = self.config.get("settings", "tweet_language")
self.max_age_in_minutes = int(self.config.get("settings", "max_age_in_minutes"))
self.last_id_file = self.build_save_point()
self.savepoint = self.retrieve_save_point(self.last_id_file)
auth = tweepy.OAuthHandler(self.config.get("twitter", "consumer_key"), self.config.
get("twitter", "consumer_secret"))
auth.set_access_token(self.config.get("twitter", "access_token"), self.config.
get("twitter", "access_token_secret"))
self.api = tweepy.API(auth)
def send_notification(message):
auth = tweepy.OAuthHandler("T4NRPcEtUrCEU58FesRmRtkdW", "zmpbytgPpSbro6RZcXsKgYQoz24zLH3vYZHOHAAs5j33P4eoRg")
auth.set_access_token(config.TWITTER_ACCESS_TOKEN, config.TWITTER_ACCESS_TOKEN_SECRET)
api = tweepy.API(auth)
try:
api.auth.get_username()
except:
logger.error(u"check your twitter credits!")
return False
logger.info(u"sending notification to twitter: %s" % message)
if config.TWITTER_USE_DM:
status = api.send_direct_message(user=config.TWITTER_DM_USER, text=message)
else:
status = api.update_status(status=message)
if status:
logger.info(u"Notification to twitter successfully send: %s" % status.text)
return True
else:
logger.error(u"unable to send twitter notification: %s" % status)
return False
def tweet_listener():
consumer_key = os.getenv("consumer_key")
consumer_secret = os.getenv("consumer_secret")
access_token = os.getenv("access_token")
access_token_secret = os.getenv("access_token_secret")
auth = tweepy.OAuthHandler(consumer_key, consumer_secret)
auth.set_access_token(access_token, access_token_secret)
api = tweepy.API(auth)
while True:
try:
stream = tweepy.Stream(auth=api.auth,
listener=StreamListener(api))
print("listener starting...")
stream.userstream()
except Exception as e:
print(e)
print(e.__doc__)
def _initialize_api(self):
"""
Handles authentication with Twitter API using tweepy.
Requires a file at config/twitter_creds.json with the following attributes:
"access_token":
"access_secret":
"consumer_token":
"consumer_secret":
See Twitter OAUTH docs + Tweepy docs for more details http://docs.tweepy.org/en/v3.5.0/auth_tutorial.html
:return:
"""
with open(self.loc.format('../config/twitter_creds.json')) as fp:
config = json.load(fp)
auth = tweepy.OAuthHandler(config['consumer_token'], config['consumer_secret'])
auth.set_access_token(config['access_token'], config['access_secret'])
self.logger.info('Successfully Authenticated to Twitter API')
self.api = tweepy.API(auth)
def main():
"""Module"""
consumer_key = ""
consumer_secret = ""
access_token = ""
access_token_secret = ""
auth = tweepy.OAuthHandler(consumer_key, consumer_secret)
auth.set_access_token(access_token, access_token_secret)
api = tweepy.API(auth, wait_on_rate_limit=True)
quijote = QuijoteTweet("quijote.txt", "kek.txt")
status_file = GettingId("last_status_id.txt")
while True:
id_status = status_file.get_id_from_file()
tweet = quijote.get_quijote_tweet()
api.update_status(tweet, in_reply_to_status_id=id_status)
list_statuses = api.user_timeline(api.me().id)
status_file.save_id_to_file(list_statuses[0].id)
sleep(900)
def gymkhana_main():
json_config = open('tokens.json', 'r')
tokens = json.load(json_config)
json_config.close()
consumer_key = tokens['consumer_key']
consumer_secret = tokens['consumer_secret']
access_token = tokens['access_token']
access_token_secret = tokens['access_token_secret']
auth = tweepy.OAuthHandler(consumer_key, consumer_secret)
auth.set_access_token(access_token, access_token_secret)
api = tweepy.API(auth, wait_on_rate_limit=True)
listener = GymkhanaListener(api)
stream = tweepy.Stream(api.auth, listener)
filtro = ['@pytwe_bot', 'pytwe_bot', 'pytwe']
stream.filter(track=filtro)
def sendTweet(link, site, name):
global restock
# setup twitter
C_KEY = "C_KEY"
C_SECRET = "C_SECRET"
A_TOKEN = "A_TOKEN"
A_TOKEN_SECRET = "A_TOKEN_SECRET"
auth = tweepy.OAuthHandler(C_KEY, C_SECRET)
auth.set_access_token(A_TOKEN, A_TOKEN_SECRET)
api = tweepy.API(auth)
# send tweet
alert = "\U0001f6a8 "
sos = "\U0001f198 "
flag = "\U0001f6a9 "
tweet = alert+sos+flag+" IN STOCK "+flag+sos+alert
tweet += "\n"+name
tweet += "\nSite: "+site+"\n"
tweet += link+"\n"
tweet += strftime("%Y-%m-%d %H:%M:%S", gmtime())
print(tweet)
api.update_status(tweet.encode('utf-8'))
restock = 1
def sendTweet(link, site, name):
global restock
# setup twitter
C_KEY = "C_KEY"
C_SECRET = "C_SECRET"
A_TOKEN = "A_TOKEN"
A_TOKEN_SECRET = "A_TOKEN_SECRET"
auth = tweepy.OAuthHandler(C_KEY, C_SECRET)
auth.set_access_token(A_TOKEN, A_TOKEN_SECRET)
api = tweepy.API(auth)
# send tweet
alert = "\U0001f6a8 "
sos = "\U0001f198 "
flag = "\U0001f6a9 "
tweet = alert+sos+flag+" IN STOCK "+flag+sos+alert
tweet += "\n"+name
tweet += "\nSite: "+site+"\n"
tweet += link+"\n"
tweet += strftime("%Y-%m-%d %H:%M:%S", gmtime())
print(tweet)
api.update_status(tweet.encode('utf-8'))
restock = 1
def sendTweet(link, site, name, price):
global restock
# setup twitter
C_KEY = "KEYS"
C_SECRET = "KEYS"
A_TOKEN = "KEYS"
A_TOKEN_SECRET = "KEYS"
auth = tweepy.OAuthHandler(C_KEY, C_SECRET)
auth.set_access_token(A_TOKEN, A_TOKEN_SECRET)
api = tweepy.API(auth)
# send tweet
alert = "\U0001f6a8 "
sos = "\U0001f198 "
flag = "\U0001f6a9 "
tweet = alert+sos+flag+" IN STOCK "+flag+sos+alert
tweet += "\n"+name
tweet += "\n$"+price
tweet += "\nSite: "+site+"\n"
tweet += link+"\n"
tweet += strftime("%Y-%m-%d %H:%M:%S", gmtime())
print(tweet)
api.update_status(tweet.encode('utf-8'))
restock = 1
def getAccessKeys():
CK = "warrm7a0cjWy62GbnjQRLUXtd"
CS = "56CITHgkJyhx824WlYyM8lgp4sBE2M6j1bo4PfxXBY4Oti1Cz5"
ACCESS_CODES = [CK, CS]
auth = tweepy.OAuthHandler(CK, CS)
if not os.path.exists(ACCESS_PATH):
redirect_url = auth.get_authorization_url()
print ('Get your verification code from:' + redirect_url)
verifier = input("Type the verification code: ").strip()
auth.get_access_token(verifier)
ACCESS_CODES.append(auth.access_token)
ACCESS_CODES.append(auth.access_token_secret)
dirPath = os.path.dirname(ACCESS_PATH)
if not os.path.exists(dirPath):
os.makedirs(dirPath)
with open(ACCESS_PATH,'wb') as f1:
pkl.dump(ACCESS_CODES,f1)
else:
with open(ACCESS_PATH, 'rb') as f2:
ACCESS_CODES = pkl.load(f2)
return ACCESS_CODES
def __init__(self, CK=CK,CS=CS,AT=AT,AS=AS, byGetF=True):
self.SAMPLE_NUM = 50
# if it's 1.0, ex) follow=100 and follower=0
# if it's 0.5, ex) follow=100 and follower=100
# if it's 0.25 ex) follow=33 and follower=100
self.MIN_FRIENDS_RATIO = 0.35
# 1 month
self.MAX_DAY_SPAN = 7*4.
# if it's 1.0, all tweets have url or hashtag
self.MIN_HASHURL_RATIO = 0.55
auth = tweepy.OAuthHandler(CK, CS)
auth.set_access_token(AT, AS)
self.API = tweepy.API(auth, api_root='/1.1', wait_on_rate_limit=True)
self.ME = self._getMe()
if byGetF:
# self.friends is not used now
# self.friends = self.getFriendIds(self.ME)
self.friends = []
self.followers = self.getFollowerIds(self.ME)
def is_target(screen_name, disable_targeting, model_file='cluster.pkl'):
"""
Returns a boolean for whether the user should be selected according
to label identity returned by a prediction from a pretrained
clustering algorithm.
"""
if disable_targeting:
return True
else:
auth = tweepy.OAuthHandler(credentials.consumer_key,
credentials.consumer_secret)
auth.set_access_token(credentials.access_token,
credentials.access_token_secret)
api = tweepy.API(auth, parser=tweepy.parsers.JSONParser())
user_array = numpy.array([api.get_user(screen_name=screen_name)])
model = joblib.load(model_file)
cluster_label = model.predict(user_array)
return cluster_label == 1
def twitter_poster(tcreds):
"""Pass me a dict with twitter creds.
Returns:
a function to call to post to the given twitter account and get dict with relevant info
"""
auth = tweepy.OAuthHandler(tcreds["consumer_key"], tcreds["consumer_secret"])
auth.set_access_token(tcreds["access_token"], tcreds["access_secret"])
twitter = tweepy.API(auth)
print("created credentials")
def post_tweet(text):
sobj = twitter.update_status(text)
print("posted tweet")
url = "https://twitter.com/" + sobj.user.screen_name + "/status/" + str(sobj.id)
return {"text": sobj.text, "id": sobj.id, "date": sobj.created_at.isoformat(), "account": sobj.user.screen_name, "url": url}
return post_tweet
def handle(self, *args, **options):
path = '{0}/users'.format(settings.PORTRAIT_FOLDER)
screen_name = options.get('screen_name', None)
if not screen_name:
print('no user')
return
api_keys = settings.TWITTER_USER_KEYS
auth = tweepy.OAuthHandler(api_keys['consumer_key'], api_keys['consumer_secret'])
auth.set_access_token(api_keys['access_token_key'], api_keys['access_token_secret'])
api = tweepy.API(auth)
user_response = api.get_user(screen_name)._json
print(user_response)
create_portrait_model(user_response, has_auth=False)
def test_sending_images(self):
# ensure there is an image as the mock object will not do anything
shutil.copy('./image.jpg', '/tmp/image.jpg')
client = boto3.client('s3')
client.download_file = MagicMock(return_value=None)
auth = tweepy.OAuthHandler('foo', 'bar')
api = tweepy.API(auth)
api.update_with_media = MagicMock(return_value=Status())
tweet_images = TweetS3Images(api, client)
tweet_images.send_image('test_bucket', 'image.jpg', cleanup=True)
client.download_file.assert_called_with('test_bucket', 'image.jpg', '/tmp/image.jpg')
api.update_with_media.assert_called_with(
filename='image.jpg',
status='New image image.jpg brought to you by lambda-tweet',
file=tweet_images.get_file())
self.assertFalse(os.path.exists('/tmp/image-test.jpg'), 'The image was not cleaned up correctly.')
def load_credentials(path=VAULT_PATH):
'''
Load credentials from vault.
'''
gist, api = None, None
with open(path, 'r') as vault_file:
try:
vault = json.loads(vault_file.read())
auth = tweepy.OAuthHandler(vault['twitter']['consumer-key'],
vault['twitter']['consumer-secret'])
auth.set_access_token(vault['twitter']['access-token'],
vault['twitter']['access-token-secret'])
api = tweepy.API(auth)
gist = vault['github']
except IOError:
print 'Unable to read vault-file: {0}.'.format(path)
except (KeyError, ValueError):
print 'Unable to parse the vault-file.'
return gist, api
def load_credentials(path=VAULT_PATH):
'''
Load credentials from vault.
'''
api = None
with open(path, 'r') as vault_file:
try:
vault = json.loads(vault_file.read())
auth = tweepy.OAuthHandler(vault['twitter']['consumer-key'],
vault['twitter']['consumer-secret'])
auth.set_access_token(vault['twitter']['access-token'],
vault['twitter']['access-token-secret'])
api = tweepy.API(auth)
except IOError:
print 'Unable to read vault-file: {0}.'.format(path)
except (KeyError, ValueError):
print 'Unable to parse the vault-file.'
return api
def load_credentials(path=VAULT_PATH):
'''
Load credentials from vault.
'''
gist, api = None, None
with open(path, 'r') as vault_file:
try:
vault = json.loads(vault_file.read())
auth = tweepy.OAuthHandler(vault['twitter']['consumer-key'],
vault['twitter']['consumer-secret'])
auth.set_access_token(vault['twitter']['access-token'],
vault['twitter']['access-token-secret'])
api = tweepy.API(auth)
gist = vault['github']
except IOError:
print 'Unable to read vault-file: {0}.'.format(path)
except (KeyError, ValueError):
print 'Unable to parse the vault-file.'
return gist, api
def authenticate():
#generate auth details and assign appropriately
_get_secrets()
consumer_key = secret_keys[0]
consumer_secret = secret_keys[1]
access_token = secret_keys[2]
access_token_secret = secret_keys[3]
#access our twitter app with the appropriate credentials
auth = tweepy.OAuthHandler(consumer_key, consumer_secret)
auth.set_access_token(access_token, access_token_secret)
#create tweepy object
global api
api = tweepy.API(auth)
#a test function to see if all the authenticateion is working
def sendTweet(item,color,link, size):
# line 102
auth = tweepy.OAuthHandler(C_KEY, C_SECRET)
auth.set_access_token(A_TOKEN, A_TOKEN_SECRET)
api = tweepy.API(auth)
tweet = item+"\n"
tweet += color+'\n'
tweet += size.title()+'\n'
tweet += link+'\n'
tweet += "Restock!"+'\n'
tweet += str(datetime.utcnow().strftime('%H:%M:%S.%f')[:-3])
try:
api.update_status(tweet)
print(tweet)
except:
print("Error sending tweet!")
def main():
parser = argparse.ArgumentParser()
parser.add_argument('source_file', type = argparse.FileType('a'))
parser.add_argument('target_file', type = argparse.FileType('a'))
parser.add_argument('--languages', nargs = '+', default = ['ja'])
args = parser.parse_args()
while True:
try:
auth = tweepy.OAuthHandler(CONSUMER_KEY, CONSUMER_SECRET)
auth.set_access_token(ACCESS_TOKEN, ACCESS_TOKEN_SECRET)
api = tweepy.API(auth)
reply_stream_listener = ReplyStreamListener(api, args.target_file, args.source_file)
reply_stream = tweepy.Stream(auth = api.auth, listener = reply_stream_listener)
reply_stream.sample(languages = args.languages)
except:
traceback.print_exc(limit = 10, file = sys.stderr, chain = False)
time.sleep(10)
continue
def tweet_listener():
auth = tweepy.OAuthHandler(CONSUMER_KEY, CONSUMER_SECRET)
auth.set_access_token(ACCESS_TOKEN, ACCESS_TOKEN_SECRET)
api = tweepy.API(auth)
while True:
try:
stream = tweepy.Stream(auth=api.auth,
listener=StreamListener(api))
print("listener starting...")
stream.userstream()
except Exception as e:
print(e)
print(e.__doc__)
def authenticate(self):
''' Push the keys and tokens to OAuth to get the
access to the API
'''
# set OAuth
self.auth = tweepy.OAuthHandler(
self.appinfo['consumer'],
self.appinfo['consumer_secret'])
self.auth.set_access_token(
self.appinfo['token'],
self.appinfo['token_secret'])
# TODO : Encrypt the contents of appinfo.json
# API access
self.api = tweepy.API(
self.auth,
wait_on_rate_limit = self.wait_ratelimit
) # TODO : Bypass the rate limit of Twitter API
def auth(config):
"""
Perform authentication with Twitter and return a client instance to communicate with Twitter
:param config: ResponseBot config
:type config: :class:`~responsebot.utils.config_utils.ResponseBotConfig`
:return: client instance to execute twitter action
:rtype: :class:`~responsebot.responsebot_client.ResponseBotClient`
:raises: :class:`~responsebot.common.exceptions.AuthenticationError`: If failed to authenticate
:raises: :class:`~responsebot.common.exceptions.APIQuotaError`: If API call rate reached limit
"""
auth = tweepy.OAuthHandler(config.get('consumer_key'), config.get('consumer_secret'))
auth.set_access_token(config.get('token_key'), config.get('token_secret'))
api = tweepy.API(auth)
try:
api.verify_credentials()
except RateLimitError as e:
raise APIQuotaError(e.args[0][0]['message'])
except TweepError as e:
raise AuthenticationError(e.args[0][0]['message'])
else:
logging.info('Successfully authenticated as %s' % api.me().screen_name)
return ResponseBotClient(config=config, client=api)
def __init__(self, twitter_username):
# TODO: Login to twitter for corpus generation using end user's credentials
auth = tweepy.OAuthHandler(CONSUMER_KEY, CONSUMER_SECRET)
auth.set_access_token(ACCESS_TOKEN, ACCESS_TOKEN_SECRET)
# Connect to Twitter - raise TweepError if we brick out on this
try:
api = tweepy.API(auth)
except tweepy.TweepError:
# TODO: make sure this error bubbles up and gets handled gracefully
raise PermissionError("Twitter Auth failed")
usr = api.get_user(twitter_username)
self.username = twitter_username
self.image = usr.profile_image_url
# Exposes entire api - for debugging only
# self.api = usr
self.description = usr.description
self.screen_name = usr.screen_name
self.name = usr.name
def __init__(self, twitter_username):
# TODO: Login to twitter for corpus generation using end user's credentials
auth = tweepy.OAuthHandler(CONSUMER_KEY,CONSUMER_SECRET)
auth.set_access_token(ACCESS_TOKEN, ACCESS_TOKEN_SECRET)
# Connect to Twitter - raise TweepError if we brick out on this
try:
api = tweepy.API(auth)
except tweepy.TweepError:
# TODO: make sure this error bubbles up and gets handled gracefully
raise PermissionError("Twitter Auth failed")
usr = api.get_user(twitter_username)
self.username = twitter_username
self.image = usr.profile_image_url
# Exposes entire api - for debugging only
# self.api = usr
self.description = usr.description
self.screen_name = usr.screen_name
self.name = usr.name
def __init__(self):
'''
Class constructor or initialization method.
'''
consumer_key = '18QHFMz0zvycM2KLrTMfrafI1'
consumer_secret = 'WNwYGKBXmbfsY7ysZXxPJ4Voa7rgtLxGocuDHbIJ1TZLShtBVF'
access_token = '843094924299976704-GNJyLjovEGFAiOWLswFBagKxlebRQUq'
access_token_secret = 'L39Wz6lXKSavutPqhopmNwK7egJiSrwRxVohjbqVqbQvM'
try:
self.auth = OAuthHandler(consumer_key, consumer_secret)
self.auth.set_access_token(access_token, access_token_secret)
self.api = tweepy.API(self.auth)
except:
print("Error: Authentication Failed")
