我们从Python开源项目中，提取了以下50个代码示例，用于说明如何使用tweepy.OAuthHandler()。
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())
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, 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 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")
def _get_twitter_api(self): """ Since we are only reading public information from Twitter, we don't need access token/secret values. """ with open('secrets.json') as secrets_file: secrets = json.load(secrets_file) consumer_key = secrets['consumer_key'] consumer_secret = secrets['consumer_secret'] access_token = secrets['access_token'] access_secret = secrets['access_secret'] self.auth = OAuthHandler(consumer_key, consumer_secret) self.auth.set_access_token(access_token, access_secret) return tweepy.API(self.auth)
def create_api_instance(tokens): """return authenticated tweepy api instance. In order to do that you need to provide a dictionary which values are the four tokens provided by apps.twitter when registering an app. Parameters ---------- tokens : dict{'api_key': <api_key>, 'api_secret': <api_secret>, 'access': <access>, 'access_secret': <secret_access>} """ auth = tweepy.OAuthHandler(tokens['api_key'], tokens['api_secret']) auth.set_access_token(tokens['access'], tokens['access_secret']) if 'proxy' in tokens: api = tweepy.API(auth, proxy=tokens['proxy']) else: api = tweepy.API(auth) return api
def tweetit(msg): import tweepy access_token = "" access_token_secret = "" consumer_key = "" consumer_secret = "" if(msg != ""): print("Message Tweet") auth = tweepy.OAuthHandler(consumer_key, consumer_secret) auth.set_access_token(access_token, access_token_secret) api = tweepy.API(auth) tweet = msg print(tweet) if(debug == False): api.update_status(tweet) if(debug == True): print("Post not uploaded due to debug mode being on") if(msg == ""): print("Error no message")
def __init__(self, corpus, friends=[], commentary="None", black_list=[], local="world", hashtag_search=None): self.black_list = black_list self.local = local self.friends = friends self.corpus = corpus auth = tweepy.OAuthHandler(ConsumerKey, ConsumerSecret) auth.set_access_token(AccessToken, AccessTokenSecret) self.api = tweepy.API(auth) entry = [("Date", [get_date()]), ("Followers", [len(self.api.followers_ids())]), ("Following", [len(self.api.friends_ids())]), ("Commentary", [commentary])] self.df = pd.DataFrame.from_items(entry) self.log() if hashtag_search is None: self.hashtag_search = self.get_trends(self.local) else: self.hashtag_search = hashtag_search + self.get_trends(self.local)
def tweet_status(status, image_path=None): try: auth = tweepy.OAuthHandler(settings.CONSUMER_KEY, settings.CONSUMER_SECRET) auth.set_access_token(settings.ACCESS_TOKEN, settings.ACCESS_TOKEN_SECRET) api = tweepy.API(auth) if image_path: new_tweet = api.update_with_media(image_path, status=status) else: new_tweet = api.update_status(status) except tweepy.TweepError as ex: raise MeteoSangueException(ex) try: mention = '{0} {1}'.format( ' '.join([ass['twitter_id'] for ass in settings.BLOOD_ASSOCIATIONS if 'twitter_id' in ass]), 'Nuovo bollettino meteo ?', ) api.update_status(mention, in_reply_to_status_id=new_tweet.id) except tweepy.TweepError as ex: pass #Mention is allowed to fail silently
def setUp(self): self.data = { "DC_PEC": "", "DC_CODE": ''' import tweepy access_token = "1092294848-aHN7DcRP9B4VMTQIhwqOYiB14YkW92fFO8k8EPy" access_token_secret = "X4dHmhPfaksHcQ7SCbmZa2oYBBVSD2g8uIHXsp5CTaksx" consumer_key = "nZ6EA0FxZ293SxGNg8g8aP0HM" consumer_secret = "fJGEodwe3KiKUnsYJC3VRndj7jevVvXbK2D5EiJ2nehafRgA6i" auth = tweepy.OAuthHandler(consumer_key, consumer_secret) auth.set_access_token(access_token, access_token_secret) ''', "DC_SOLUTION": ''' import tweepy access_token = "1092294848-aHN7DcRP9B4VMTQIhwqOYiB14YkW92fFO8k8EPy" access_token_secret = "X4dHmhPfaksHcQ7SCbmZa2oYBBVSD2g8uIHXsp5CTaksx" consumer_key = "nZ6EA0FxZ293SxGNg8g8aP0HM" consumer_secret = "fJGEodwe3KiKUnsYJC3VRndj7jevVvXbK2D5EiJ2nehafRgA6i" auth = tweepy.OAuthHandler(consumer_key, consumer_secret) auth.set_access_token(access_token, access_token_secret) ''' }
def test_Pass(self): self.data["DC_SCT"] = ''' # Test: import tweepy import_msg = "Did you correctly import the required package?" test_import("tweepy", same_as = True, not_imported_msg = import_msg, incorrect_as_msg = import_msg) # Test: Predefined code predef_msg = "You don't have to change any of the predefined code." test_object("access_token", undefined_msg = predef_msg, incorrect_msg = predef_msg) test_object("access_token_secret", undefined_msg = predef_msg, incorrect_msg = predef_msg) test_object("consumer_key", undefined_msg = predef_msg, incorrect_msg = predef_msg) test_object("consumer_secret", undefined_msg = predef_msg, incorrect_msg = predef_msg) # Test: call to tweepy.OAuthHandler() and 'auth' variable test_object("auth", do_eval = False) test_function("tweepy.OAuthHandler") # Test: call to auth.set_access_token() test_function("auth.set_access_token") success_msg("Awesome!") ''' sct_payload = helper.run(self.data) self.assertTrue(sct_payload['correct'])
def main(args): if args.debug: logger.setLevel(logging.DEBUG) auth = tweepy.OAuthHandler(args.consumer_key, args.consumer_secret) auth.set_access_token(args.access_token, args.access_token_secret) api = tweepy.API(auth, wait_on_rate_limit=True) screen_name = api.me().screen_name if args.classifier == 'mock': classifier = classifiers.MockClassifier() elif args.classifier == 'local': classifier = classifiers.URLClassifier(classifiers.ImageClassifier(args.dataset_path, INPUT_SHAPE)) elif args.classifier == 'remote': classifier = classifiers.RemoteClassifier(args.remote_endpoint) stream = tweepy.Stream(auth=auth, listener=ReplyToTweet(screen_name, classifier, api, args.silent)) logger.info('Listening as {}'.format(screen_name)) stream.userstream(track=[screen_name])
def __init__(self): api_key = AppConfig.ENVIRONMENT_VARIABLE['twitter_api_key'] api_secret = AppConfig.ENVIRONMENT_VARIABLE['twitter_api_secret'] access_token = AppConfig.ENVIRONMENT_VARIABLE['twitter_access_token'] access_token_secret = AppConfig.ENVIRONMENT_VARIABLE['twitter_access_token_secret'] _check_environment_variable_set(api_key, raise_err = True) _check_environment_variable_set(api_secret, raise_err = True) _check_environment_variable_set(access_token, raise_err = True) _check_environment_variable_set(access_token_secret, raise_err = True) self.api_key = api_key self.api_secret = api_secret self.access_token = access_token self.access_token_secret = access_token_secret self.auth_handler = tweepy.OAuthHandler(self.api_key, self.api_secret) self.auth_handler.set_access_token(self.access_token, self.access_token_secret) self.api = tweepy.API(self.auth_handler)
def get_twitter_auth(): """Setup Twitter authentication. Return: tweepy.OAuthHandler object """ try: consumer_key = os.environ['TWITTER_CONSUMER_KEY'] consumer_secret = os.environ['TWITTER_CONSUMER_SECRET'] access_token = os.environ['TWITTER_ACCESS_TOKEN'] access_secret = os.environ['TWITTER_ACCESS_SECRET'] except KeyError: print("TWITTER_* environment variables not set\n") sys.stderr.write("TWITTER_* environment variables not set\n") sys.exit(1) auth = OAuthHandler(consumer_key, consumer_secret) auth.set_access_token(access_token, access_secret) return auth
def streaming(credentials, coins, queries, refresh, path, realtime=False, logTracker=True, logTweets=True, logSentiment=False, debug=True): # User Error Checks if len(coins) <= 0: print("Error: You must include at least one coin."); return if len(coins) >= 10: print("Warning: Fewer than ten coins recommended.") if len(queries) <= 0: print("Error: You must include at least one query."); return if len(queries) >= 20: print("Warning: Fewer than twenty queries recommended.") if refresh <= 0: print("Error: Refresh rate must be greater than 0"); return auth = tweepy.OAuthHandler(credentials[0], credentials[1]) auth.set_access_token(credentials[2], credentials[3]) if logSentiment: global SentimentIntensityAnalyzer from nltk.sentiment.vader import SentimentIntensityAnalyzer while True: # Start streaming ----------------------------- try: print("Streaming Now...") listener = CoinListener(auth, coins, queries, refresh, path, realtime, logTracker, logTweets, logSentiment, debug) stream = tweepy.Stream(auth, listener) stream.filter(track=queries) except (Timeout, ConnectionError, ReadTimeoutError): print("Reestablishing Connection...") with open("%sError_Log.txt" % path, "a") as outfile: outfile.write("%s Error: Connection Dropped\n" % time.strftime('%m/%d/%Y %H:%M')) time.sleep((15*60)+1) # Wait at least 15 minutes before restarting listener # ---------------------------------------------
def fetch_tweets_and_send_emails() -> None: logging.info('-- starting fetching tweets and sending emails') auth = tweepy.OAuthHandler(Settings.TWITTER_CONSUMER_KEY, Settings.TWITTER_CONSUMER_SECRET) auth.set_access_token(Settings.TWITTER_ACCESS_TOKEN, Settings.TWITTER_ACCESS_TOKEN_SECRET) twitter_api = tweepy.API(auth) with smtp_server() as server: for list_slug in LIST_SLUGS: last_tweet_id = load_last_tweet_id(list_slug) if last_tweet_id is None: logging.warning('no last tweet id was found for list %s' % list_slug) tweets = get_tweets(twitter_api, list_slug=list_slug, last_tweet_id=last_tweet_id) for tweet in reversed(tweets): kwargs = parse_tweet(tweet) kwargs['msg_subject'] = '[{list_slug}] {subject}'.format(list_slug=list_slug, subject=kwargs['msg_subject']) send_email(server, **kwargs) save_last_tweet_id(last_id=tweet.id_str, list_slug=list_slug)
def twittersignin(): auth = tweepy.OAuthHandler(CONSUMER_KEY, CONSUMER_SECRET) try: redirect_url = auth.get_authorization_url() session["request_token"] = auth.request_token except tweepy.TweepError: flash("Failed to get request token", "danger") return redirect(url_for("bp.home")) return redirect(redirect_url)
