def load(self):
self.oauth = OAuth(
self.config.get(c.ACCESS_TOKEN_KEY),
self.config.get(c.ACCESS_TOKEN_SECRET_KEY),
self.config.get(c.CONSUMER_KEY_KEY),
self.config.get(c.CONSUMER_SECRET_KEY),
)
self.client = TwitterAPI(auth=self.oauth)
context = dict(plugin=self, bot=self.bot)
self.add_web_handlers(
(r'/twitter/oauth_token', OAuthTokenHandler, context),
(r'/twitter/authorize', OAuthAuthorizeHandler, context),
(r'/twitter/status', TwitterStatusHandler, context),
(r'/twitter/disconnect', TwitterDisconnectHandler, context),
)
self.monitored = set()
self.restore_monitored()
self.request_tokens = dict()
self.last_tweets = dict()
self.interactive_tweets = dict()
python类Twitter()的实例源码
def __init__(self, settings):
# Required Parameters
self.__token = require_and_remove_key('access_token', settings, "'Twitter' type alarms.")
self.__token_key = require_and_remove_key('access_secret', settings, "'Twitter' type alarms.")
self.__con_secret = require_and_remove_key('consumer_key', settings, "'Twitter' type alarms.")
self.__con_secret_key = require_and_remove_key('consumer_secret', settings, "'Twitter' type alarms.")
self.__client = None
# Optional Alarm Parameters
self.__startup_message = parse_boolean(settings.pop('startup_message', "True"))
# Optional Alert Parameters
self.__pokemon = self.create_alert_settings(settings.pop('pokemon', {}), self._defaults['pokemon'])
self.__pokestop = self.create_alert_settings(settings.pop('pokestop', {}), self._defaults['pokestop'])
self.__gym = self.create_alert_settings(settings.pop('gym', {}), self._defaults['gym'])
self.__egg = self.create_alert_settings(settings.pop('egg', {}), self._defaults['egg'])
self.__raid = self.create_alert_settings(settings.pop('raid', {}), self._defaults['raid'])
# Warn user about leftover parameters
reject_leftover_parameters(settings, "'Alarm level in Twitter alarm.")
log.info("Twitter Alarm has been created!")
# Establish connection with Twitter
def twitter_callback(self, tweet):
"""Analyzes Trump tweets, trades stocks, and tweets about it."""
# Initialize the Analysis, Logs, Trading, and Twitter instances inside
# the callback to create separate httplib2 instances per thread.
analysis = Analysis(logs_to_cloud=LOGS_TO_CLOUD)
logs = Logs(name="main-callback", to_cloud=LOGS_TO_CLOUD)
# Analyze the tweet.
companies = analysis.find_companies(tweet)
logs.info("Using companies: %s" % companies)
if not companies:
return
# Trade stocks.
trading = Trading(logs_to_cloud=LOGS_TO_CLOUD)
trading.make_trades(companies)
# Tweet about it.
twitter = Twitter(logs_to_cloud=LOGS_TO_CLOUD)
twitter.tweet(companies, tweet)
def initial_start():
print("Welcome to the Twitter Stock Monitor!")
twitter_handles = input("Enter the Twitter handles you want this bot to follow, separated by spaces: \n").split()
with open('./Files/config.json') as json_f:
json_data = json.load(json_f)
json_data["TwitterAuth"]["Handles"] = twitter_handles
json_data["InitialStart"] = False
with open('./Files/config.json', 'w') as json_f:
json.dump(json_data, json_f, sort_keys=True, indent=4, ensure_ascii=False)
print("Creating files needed..")
for handle in twitter_handles:
with open(f'./Files/LatestTweets/{handle}.txt', "w") as f:
continue
print(f'Files created! This bot will now begin to monitor: {twitter_handles}\n\n\n')
return twitter_handles
def twitter_autoreply_stop(self):
"""Stops the Thread that replies to all tweets that match the target
string.
For an explanation of the target string, see the Twitter dev site:
https://dev.twitter.com/streaming/overview/request-parameters#track
"""
# Raise an Exception if the twitter library wasn't imported
if not IMPTWITTER:
self._error(u'twitter_autoreply_stop', \
u"The 'twitter' library could not be imported. Check whether it is installed correctly.")
# Update the autoreply parameters
self._autoreply_database = None
self._targetstring = None
self._keywords = None
self._tweetprefix = None
self._tweetsuffix = None
# Signal the _autoreply thread to pause
self._autoreplying = False
def twitter_autoreply_stop(self):
"""Stops the Thread that replies to all tweets that match the target
string.
For an explanation of the target string, see the Twitter dev site:
https://dev.twitter.com/streaming/overview/request-parameters#track
"""
# Raise an Exception if the twitter library wasn't imported
if not IMPTWITTER:
self._error(u'twitter_autoreply_stop', \
u"The 'twitter' library could not be imported. Check whether it is installed correctly.")
# Update the autoreply parameters
self._autoreply_database = None
self._targetstring = None
self._keywords = None
self._tweetprefix = None
self._tweetsuffix = None
# Signal the _autoreply thread to pause
self._autoreplying = False
def __init__(self, settings):
#Service Info
self.token = settings['access_token']
self.token_key = settings['access_secret']
self.con_secret = settings['consumer_key']
self.con_secret_key = settings['consumer_secret']
self.startup_message = settings.get('startup_message', "True")
self.startup_list = settings.get('startup_list', "True")
#Set Alerts
self.pokemon = self.set_alert(settings.get('pokemon', {}), self._defaults['pokemon'])
self.pokestop = self.set_alert(settings.get('pokestop', {}), self._defaults['pokestop'])
self.gym = self.set_alert(settings.get('gyms', {}), self._defaults['gym'])
#Connect and send startup messages
self.connect()
timestamps = get_timestamps(datetime.utcnow());
if parse_boolean(self.startup_message):
self.client.statuses.update(status="%s - PokeAlarm has intialized!" % timestamps[2])
if parse_boolean(self.startup_list):
for line in notify_list_multi_msgs(config["NOTIFY_LIST"],140, "We will tweet about the following pokemon", timestamps[2]):
self.client.statuses.update(status="%s" % line)
log.info("Twitter Alarm intialized.")
#Establish connection with Twitter
def create_api():
"""create a twitter api"""
from twitter import OAuth, Twitter
from app import APP_INSTANCE as app
access_token = app.get_config('api.twitter.access_token')
access_secret = app.get_config('api.twitter.access_secret')
consumer_key = app.get_config('api.twitter.consumer_key')
consumer_secret = app.get_config('api.twitter.consumer_secret')
# temporary fix for certificate error
ssl._create_default_https_context = ssl._create_unverified_context
oauth = OAuth(access_token, access_secret, consumer_key, consumer_secret)
# Initiate the connection to Twitter API
return Twitter(auth=oauth)
def __init__(self, consumer_key, consumer_secret, user_credentials):
credentials = json.loads(user_credentials)
auth = OAuth(credentials['oauth_token'],
credentials['oauth_token_secret'],
consumer_key,
consumer_secret)
self.api = Twitter(auth=auth)
def __init__(self, consumer_key, consumer_secret):
bearer_token = oauth2_dance(consumer_key, consumer_secret)
auth = OAuth2(bearer_token=bearer_token)
self.api = Twitter(auth=auth)
def authenticate():
CONSUMER_KEY = 'BAKDTiFR9tvGl9HwUqHn3VfwW'
CONSUMER_SECRET = 'VD0CBzlxxMCV3ya746drzG9Z9VLWp55YuOxmCMunmIN8E154iv'
OAUTH_TOKEN = '324805546-8vOSeTDQQLsSZhPh1pNI1eELsU0MCU7E3fUBdbbh'
OAUTH_TOKEN_SECRET = 'wZ8vNzKe5K7n3YyZ7LKITFtmveRtHQ2s2CTYHNMGdjNBc'
auth = twitter.oauth.OAuth(OAUTH_TOKEN, OAUTH_TOKEN_SECRET,
CONSUMER_KEY, CONSUMER_SECRET)
twitter_api = twitter.Twitter(auth=auth)
return twitter_api
def authenticate():
CONSUMER_KEY = 'BAKDTiFR9tvGl9HwUqHn3VfwW'
CONSUMER_SECRET = 'VD0CBzlxxMCV3ya746drzG9Z9VLWp55YuOxmCMunmIN8E154iv'
OAUTH_TOKEN = '324805546-8vOSeTDQQLsSZhPh1pNI1eELsU0MCU7E3fUBdbbh'
OAUTH_TOKEN_SECRET = 'wZ8vNzKe5K7n3YyZ7LKITFtmveRtHQ2s2CTYHNMGdjNBc'
auth = twitter.oauth.OAuth(OAUTH_TOKEN, OAUTH_TOKEN_SECRET,
CONSUMER_KEY, CONSUMER_SECRET)
twitter_api = twitter.Twitter(auth=auth)
return twitter_api
def authenticate():
CONSUMER_KEY = 'BAKDTiFR9tvGl9HwUqHn3VfwW'
CONSUMER_SECRET = 'VD0CBzlxxMCV3ya746drzG9Z9VLWp55YuOxmCMunmIN8E154iv'
OAUTH_TOKEN = '324805546-8vOSeTDQQLsSZhPh1pNI1eELsU0MCU7E3fUBdbbh'
OAUTH_TOKEN_SECRET = 'wZ8vNzKe5K7n3YyZ7LKITFtmveRtHQ2s2CTYHNMGdjNBc'
auth = twitter.oauth.OAuth(OAUTH_TOKEN, OAUTH_TOKEN_SECRET,
CONSUMER_KEY, CONSUMER_SECRET)
twitter_api = twitter.Twitter(auth=auth)
return twitter_api
def connect(self):
self.__client = Twitter(
auth=OAuth(self.__token, self.__token_key, self.__con_secret, self.__con_secret_key))
# Send a start up tweet
def startup_message(self):
if self.__startup_message:
timestamps = get_time_as_str(datetime.utcnow())
args = {"status": "{}- PokeAlarm activated!" .format(timestamps[2])}
try_sending(log, self.connect, "Twitter", self.send_tweet, args)
log.info("Startup tweet sent!")
# Set the appropriate settings for each alert
def create_alert_settings(self, settings, default):
alert = {
'status': settings.pop('status', default['status'])
}
reject_leftover_parameters(settings, "'Alert level in Twitter alarm.")
return alert
def send_alert(self, alert, info):
limit = 140
status = alert['status']
if status.endswith("<gmaps>"):
limit = 117 # Save 23 characters for the google maps
status = status[:-7] # Truncate gmaps
status = replace(status[:limit], info) # Truncate status
if limit == 117:
status += info['gmaps'] # Add in gmaps link
args = {"status": status}
try_sending(log, self.connect, "Twitter", self.send_tweet, args)
# Trigger an alert based on Pokemon info
def request_token(self, user):
tweaked_twitter_client = TwitterAPI(
auth = self.oauth,
format = '',
api_version = None
)
try:
response = tweaked_twitter_client.oauth.request_token(
oauth_callback = Twitter.AUTHORIZE_CALLBACK
)
except Exception as e:
self.error('Error while generating oauth token: {}'.format(e))
else:
params = dict(
(key, value) for key, value in (
component.split('=')
for component in response.split('&')
)
)
try:
token = params['oauth_token']
secret = params['oauth_token_secret']
except:
return
else:
self.info('Generated request token for {}'.format(user.name))
self.request_tokens[token] = secret
return token
def save_user(self, user, oauth_token, oauth_verifier):
oauth = OAuth(
oauth_token,
self.request_tokens[oauth_token],
self.config.get(c.CONSUMER_KEY_KEY),
self.config.get(c.CONSUMER_SECRET_KEY),
)
tweaked_twitter_client = TwitterAPI(
auth = oauth,
format = '',
api_version = None
)
try:
response = tweaked_twitter_client.oauth.access_token(
oauth_verifier = oauth_verifier
)
except Exception as e:
self.error(e)
pass
else:
params = dict(
(key, value) for key, value in (
component.split('=')
for component in response.split('&')
)
)
with self.transaction() as trans:
trans.execute(q.add_user, dict(
id = user.id,
token = params['oauth_token'],
secret = params['oauth_token_secret']
))
def get_user_api(self, oauth_user):
oauth = OAuth(
oauth_user.token,
oauth_user.secret,
self.config.get(c.CONSUMER_KEY_KEY),
self.config.get(c.CONSUMER_SECRET_KEY),
)
return TwitterAPI(auth = oauth)
def get_twitter_client():
return twitter.Twitter(
auth=twitter.OAuth(
os.environ.get('ACCESS_TOKEN'),
os.environ.get('ACCESS_TOKEN_SECRET'),
os.environ.get('CONSUMER_KEY'),
os.environ.get('CONSUMER_SECRET'),
)
)
def tweet_rsscontent(self, rsscontent):
credentials = self.data['twitter']
twitter = Twitter(**credentials)
body_length = self._calculate_tweet_length()
tweet_body = rsscontent.title[:body_length]
tweet_url = rsscontent.url
tweet_hashtag = self.data['hashtags']
tweet_text = "{} {} {}".format(tweet_body, tweet_url, tweet_hashtag)
twitter.update_status(tweet_text)
rsscontent.published=True
self.session.flush()
def get_login_url(callback='oob', consumer_key=None, consumer_secret=None):
twitter = Twitter(
auth=OAuth('', '', consumer_key, consumer_secret),
format='', api_version=None)
resp = url_decode(twitter.oauth.request_token(oauth_callback=callback))
oauth_token = resp['oauth_token']
oauth_token_secret = resp['oauth_token_secret']
token = OAuthToken(token=oauth_token, token_secret=oauth_token_secret)
db.session.merge(token)
db.session.commit()
return (
"https://api.twitter.com/oauth/authenticate?oauth_token=%s"
% (oauth_token,))
def receive_verifier(oauth_token, oauth_verifier,
consumer_key=None, consumer_secret=None):
temp_token = OAuthToken.query.get(oauth_token)
if not temp_token:
raise Exception("OAuth token has expired")
twitter = Twitter(
auth=OAuth(temp_token.token, temp_token.token_secret,
consumer_key, consumer_secret),
format='', api_version=None)
resp = url_decode(
twitter.oauth.access_token(oauth_verifier=oauth_verifier))
db.session.delete(temp_token)
new_token = OAuthToken(token=resp['oauth_token'],
token_secret=resp['oauth_token_secret'])
new_token = db.session.merge(new_token)
new_twitter = Twitter(
auth=OAuth(new_token.token, new_token.token_secret,
consumer_key, consumer_secret))
remote_acct = new_twitter.account.verify_credentials()
acct = account_from_api_user_object(remote_acct)
acct = db.session.merge(acct)
new_token.account = acct
db.session.commit()
return new_token
def get_twitter_for_acc(account):
consumer_key = app.config['TWITTER_CONSUMER_KEY']
consumer_secret = app.config['TWITTER_CONSUMER_SECRET']
tokens = (OAuthToken.query.with_parent(account)
.order_by(db.desc(OAuthToken.created_at)).all())
for token in tokens:
t = Twitter(
auth=OAuth(token.token, token.token_secret,
consumer_key, consumer_secret))
try:
t.account.verify_credentials()
return t
except TwitterHTTPError as e:
if e.e.code == 401:
# token revoked
if sentry:
sentry.captureMessage(
'Twitter auth revoked', extra=locals())
db.session.delete(token)
db.session.commit()
else:
raise TemporaryError(e)
except URLError as e:
raise TemporaryError(e)
return None
def get_tweet(tweet_id):
"""Looks up data for a single tweet."""
twitter = Twitter(logs_to_cloud=False)
return twitter.get_tweet(tweet_id)
def __init__(self, logs_to_cloud):
self.logs = Logs(name="analysis", to_cloud=logs_to_cloud)
self.gcnl_client = language.Client()
self.twitter = Twitter(logs_to_cloud=logs_to_cloud)
def __init__(self):
self.logs = Logs(name="main", to_cloud=LOGS_TO_CLOUD)
self.twitter = Twitter(logs_to_cloud=LOGS_TO_CLOUD)
def twitter():
return Twitter(logs_to_cloud=False)
def twitter_login(self, cons_key, cons_secret, access_token, \
access_token_secret):
"""Logs in to Twitter, using the provided access keys. You can get
these for your own Twitter account at apps.twitter.com
Arguments
cons_key - String of your Consumer Key (API Key)
cons_secret - String of your Consumer Secret (API Secret)
access_token - String of your Access Token
access_token_secret
- String of your Access Token Secret
"""
# Raise an Exception if the twitter library wasn't imported
if not IMPTWITTER:
self._error(u'twitter_login', u"The 'twitter' library could not be imported. Check whether it is installed correctly.")
# Log in to a Twitter account
self._oauth = twitter.OAuth(access_token, access_token_secret, \
cons_key, cons_secret)
self._t = twitter.Twitter(auth=self._oauth)
self._ts = twitter.TwitterStream(auth=self._oauth)
self._loggedin = True
# Get the bot's own user credentials
self._credentials = self._t.account.verify_credentials()
