Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def post_update(status):
global config
consumer_key = config.get('twitter', 'consumer_key')
consumer_secret = config.get('twitter', 'consumer_secret')
access_token = config.get('twitter', 'access_token')
access_token_secret = config.get('twitter', 'access_token_secret')
auth = tweepy.OAuthHandler(consumer_key, consumer_secret)
auth.set_access_token(access_token, access_token_secret)
api = tweepy.API(auth)
try:
api.update_status(status)
except tweepy.error.TweepError, e:
print "Error occurred while updating status:", e
sys.exit(1)
else:
return True
self.config = config
self.data = {}
self.data['networks'] = {}
self.ns = {}
# cache url contents for 5 minutes, check for old entries every minute
self._urlcache = timeoutdict.TimeoutDict(timeout=300, pollinterval=60)
if not os.path.exists("data"):
os.mkdir("data")
# connect to twitter
if self.config['twitter']:
key = config['twitter'][0]
secret = config['twitter'][1]
try:
auth = tweepy.OAuthHandler(consumer_key, consumer_secret)
auth.set_access_token(key, secret)
self.twapi = tweepy.API(auth)
log.info('connection to TWITTER ok')
except:
log.info('could not connect to TWITTER')
self.dbCursor = self.getConn(
str.join('.', (self.config['nick'], 'db'))
)
Args:
None
"""
# User may not have configured twitter - don't initialize it until it's
# first used
if self.twitter_api is None:
self.logger.info("Initializing Twitter")
if cfg.TWITTER_CONSUMER_KEY == '' or cfg.TWITTER_CONSUMER_SECRET == '':
self.logger.error("Twitter consumer key/secret not specified - unable to Tweet!")
elif cfg.TWITTER_ACCESS_KEY == '' or cfg.TWITTER_ACCESS_SECRET == '':
self.logger.error("Twitter access key/secret not specified - unable to Tweet!")
else:
auth = tweepy.OAuthHandler(cfg.TWITTER_CONSUMER_KEY, cfg.TWITTER_CONSUMER_SECRET)
auth.set_access_token(cfg.TWITTER_ACCESS_KEY, cfg.TWITTER_ACCESS_SECRET)
self.twitter_api = tweepy.API(auth)
#!/usr/bin/python
# -*- coding: utf-8 -*-
# View Tweets
# v1.0
# kylebx
# kylebx
# View the latest 5 tweets from your Twitter timeline.
#
# python, tweepy, json
import tweepy, json
# establish api connection with keys
auth = tweepy.OAuthHandler('consumer_key', 'consumer_secret')
auth.set_access_token('access_token', 'access_token_secret')
api = tweepy.API(auth)
def getTweet():
try:
print "Twitter"
print "---"
tweet_list = api.home_timeline(count=5) # get the latest 5 tweets from the auntenicating user
for i in range(0,len(tweet_list)): # for 5 most recent tweets
status = tweet_list[i] # set the status equal to the first tweet in the array of tweets
json_str = json.dumps(status._json) # the json string is the dump of the status converted to a json like format
jsonTweet = json.loads(json_str) # json tweet becomes an actual json object using json.loads()
tweet = jsonTweet['text'] # tweet is the text of the said tweet
userInfo = jsonTweet['user'] # twitter handle is held within userUnfo so set that
tweeter = userInfo['screen_name'] # retrieve the twitter handle
print "@{u}: {tweet}".format(u=tweeter, tweet = tweet.encode('utf-8')) # print the username and the tweet
client = get_client(PROJECT_ID, service_account=SERVICE_ACCOUNT, private_key=KEY, readonly=False)
client.swallow_results = False
logger.info("client: %s" % client)
schema_str = Utils.read_file(GNIP_SCHEMA_FILE)
schema = json.loads(schema_str)
# create table BigQuery table
created = client.create_table(DATASET_ID, TABLE_ID, schema)
logger.info("created result: %s" % created)
# if (len(created) == 0):
# print "failed to create table"
# return
l = BigQueryGnipListener(client, DATASET_ID, TABLE_ID, logger=logger)
auth = tweepy.OAuthHandler(CONSUMER_KEY, CONSUMER_SECRET)
auth.set_access_token(ACCESS_TOKEN, ACCESS_TOKEN_SECRET)
while True:
stream = None
try:
getStream(l)
# stream = tweepy.Stream(auth, l, headers = {"Accept-Encoding": "deflate, gzip"})
#stream = tweepy.Stream(auth, l)
# Choose stream: filtered or sample
# stream.sample()
#stream.filter(track=TRACK_ITEMS) # async=True
except:
def initialize():
spark = SparkSession \
.builder \
.appName("search-flight-spark-ml-model") \
.getOrCreate()
sc = spark.sparkContext
auth = tweepy.OAuthHandler(CONSUMER_KEY, CONSUMER_SECRET)
auth.set_access_token(ACCESS_TOKEN, ACCESS_SECRET)
api = tweepy.API(auth)
important_fields = ['id', 'text', 'user']
schema = StructType([
StructField('id', LongType(), False),
StructField('text', StringType(), False),
StructField('username', StringType(), False)
])
tweetsDf = spark.createDataFrame(sc.emptyRDD(), schema)
for tweet in tweepy.Cursor(api.search, q='barajas', rpp=100, lang='en').items(MAX_TWEETS):
json_tweet = {k: tweet._json[k] for k in important_fields}
json_tweet['text'] = json_tweet['text'].replace("'", "").replace("\"", "").replace("\n", "")
tweetDf = spark.createDataFrame([
import sys
import tweepy
import argparse
consumer_key = '6s35FXsv4jD2ar0ZlDYjnt7jZ'
consumer_secret = 'oFAlNZr6JGHwCdYGrYNfS3plUSdxg8UlEP2RtiKg59uSYahWRk'
access_token = '1112070588-5bNvcWYSIowvzRbRnSp4jetaCbpLk0xNVFg8egv'
access_token_secret = 'rVX3YRtx2Qa5rO9PPqtWP1Fu3HHTK70EuSmBtJXmW7KjE'
auth = tweepy.OAuthHandler(consumer_key, consumer_secret)
auth.set_access_token(access_token, access_token_secret)
api = tweepy.API(auth)
if __name__ == '__main__':
argument_parser = argparse.ArgumentParser(description='')
argument_parser.add_argument('screen_name', help='The screen name, handle, or alias that this user identifies themselves with.\
screen_names are unique. Typically a maximum of 15 characters long,\
but some historical accounts may exist with longer names.')
argument_parser.add_argument('depth', help='', type=int)
args = argument_parser.parse_args()
def auth(self):
auth = tweepy.OAuthHandler(settings.TWITTER_CONSUMER_KEY, settings.TWITTER_CONSUMER_SECRET)
auth.set_access_token(settings.TWITTER_APP_TOKEN_KEY, settings.TWITTER_APP_TOKEN_SECRET)
self.api = tweepy.API(auth)
self.api.verify_credentials()
async def authenticate(self) -> tw.API:
"""Authenticate with Twitter's API"""
auth = tw.OAuthHandler(
await self.config.api.consumer_key(), await self.config.api.consumer_secret()
)
auth.set_access_token(
await self.config.api.access_token(), await self.config.api.access_secret()
)
return tw.API(
auth,
wait_on_rate_limit=True,
wait_on_rate_limit_notify=True,
retry_count=10,
retry_delay=5,
retry_errors=[500, 502, 503, 504],
)
import json
import imghdr
import io
from pprint import pprint
import tweepy
from tweepy import Stream
from tweepy import OAuthHandler
from tweepy.streaming import StreamListener
consumer_key='CxxA37fDXWlO4nTf57qu3QmI8'
consumer_secret='ADW3q82AMPZ0cREUFFsQIQ2sPFnCaCnZZYjXqfDUcrlITaUoqB'
access_token_key='888426634268745728-eO45HlfeiuETbGai6Cz9PbH1JhaYsBz'
access_token_secret='tLjtUSEleShoiu0RPrKlNunnXp4ijH3ufmfjb3lyKoZMd'
auth = OAuthHandler(consumer_key, consumer_secret)
auth.set_access_token(access_token_key, access_token_secret)
api = tweepy.API(auth)
def read_json():
d = json.loads(sys.stdin.read())
return d.get('text'), d.get('image')
def post_tweet(text, image):
if text is not None and len(text) > 140:
raise Exception("text length is greater than 140 characters")
if image is None:
api.update_status(status=text)