Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def get_tweets(tweet_ids, consumer_key, consumer_secret, access_token, access_token_secret, nlp):
"""
Expands tweets from Twitter
:param tweet_ids: the list of tweet IDs to expand
:return: a dictionary of tweet ID to tweet text
"""
# Save tweets in a temporary file, in case the script stops working and re-starts
tweets = {}
if os.path.exists('tweet_temp'):
with codecs.open('tweet_temp', 'r', 'utf-8') as f_in:
lines = [tuple(line.strip().split('\t')) for line in f_in]
tweets = { tweet_id : tweet for (tweet_id, tweet) in lines }
api = twitter.Api(consumer_key=consumer_key, consumer_secret=consumer_secret, access_token_key=access_token,
access_token_secret=access_token_secret)
[sleeptime, resettime] = reset_sleep_time(api)
with codecs.open('tweet_temp', 'a', 'utf-8') as f_out:
for tweet_id in tweet_ids:
# We didn't download this tweet yet
if not tweet_id in tweets:
try:
curr_tweet = api.GetStatus(tweet_id, include_entities=False)
tweets[tweet_id] = clean_tweet(' '.join([t.lower_ for t in nlp(curr_tweet.text)]))
except twitter.TwitterError as err:
error = str(err)
import json
import os
import twitter
import boto3
ENDPOINT_NAME = os.getenv("SAGEMAKER_ENDPOINT_NAME", "whereml")
SSM_CREDS_NAME = os.getenv("SSM_CREDS_NAME", "/twitter/whereml")
MAX_PREDICTIONS = os.getenv("MAX_PREDICTIONS", 3)
WEBHOOK_PATH = os.getenv("WEBHOOK_PATH", "/twitter/whereml")
ssm = boto3.client('ssm')
sagemaker = boto3.client('runtime.sagemaker')
CREDS = ssm.get_parameter(Name=SSM_CREDS_NAME)['Parameter']['Value'].split(',')
CONSUMER_SECRET = CREDS[1]
twitter_api = twitter.Api(*CREDS)
TWITTER_SN = twitter_api.VerifyCredentials().screen_name
def sign_crc(crc):
h = hmac.new(
bytes(CONSUMER_SECRET, 'ascii'),
bytes(crc, 'ascii'),
digestmod=sha256
)
return json.dumps({
"response_token": "sha256="+b64encode(h.digest()).decode()
})
def verify_request(event):
crc = event['headers']['X-Twitter-Webhooks-Signature']
if __name__ == '__main__':
argc = len(sys.argv)
if argc < 5 or argc > 6:
sys.exit('Usage: python search_tweeter.py tweet statefile email passwd [code]')
tweet_new = sys.argv[1]
outfile = sys.argv[2]
email = sys.argv[3]
passwd = sys.argv[4]
code = None
if argc == 6:
code = sys.argv[5]
if tweet_new:
api = twitter.Api(consumer_key='',
consumer_secret='',
access_token_key='',
access_token_secret='')
history = get_history(email, passwd, code)
if len(history) > 0:
try:
most_recent = None
with open(outfile, 'r') as f:
most_recent = f.readline()
new_most_recent = history[0]
to_tweet = []
for search in history:
if search == most_recent:
break
def send_twitter_direct_message(sender, user_name, message, correlation_id=-1):
"""Sends a Twitter direct message to a given user, using the engine's account as sender.
Uses the TwitterSensor tool as back-end."""
# todo should be part of the TwitterSensor
try:
api = twitter.Api(username=ENGINE_TWITTER_USER_NAME, password=ENGINE_TWITTER_USER_PASSWORD)
status = api.PostUpdate("d %s %s" % (user_name, message))
return True
except:
context = "%s was trying to tell %s that %s" % (sender, user_name, message)
if correlation_id != -1:
context = "%s (processing cycle #%d)" % (context, correlation_id)
error_message = "Failed to send Twitter direct message: perhaps your connection is broken, or the whale is failing."
# todo extract the error
# error_message = " | ".join(sys.exc_info())
error_message = "%s. %s" % (context, error_message)
log_event("notification_failed", "AgentCell", sender.id, error_message, correlation_id)
return False
def forward(db, twitter_handle, mastodon_handle, debug, number=None, only_mark_as_seen=False):
t = twitter.Api(**yaml.safe_load(open("conf.yaml")))
mastodon_creds = "t2m_%s_creds.txt" % mastodon_handle
if not os.path.exists(mastodon_creds):
mastodon = Mastodon(client_id='./t2m_clientcred.txt')
print "Not credentials for mastodon account '%s', creating them (the password will NOT be saved)" % mastodon_handle
mastodon.log_in(
argh.io.safe_input("Email for mastodon account '%s': " % mastodon_handle).strip(),
getpass("Password for mastodon account of '%s': " % mastodon_handle),
to_file=mastodon_creds
)
mastodon = Mastodon(client_id='./t2m_clientcred.txt', access_token=mastodon_creds)
to_toot = []
# select tweets first
def run(self):
prev_day = '1970-05-01'
since_id = None if not self.settings['last_id'] else self.settings['last_id']
max_id = None if not self.settings['max_id'] else self.settings['max_id']
processed = 0
connection = twitter.Api(consumer_key=self.settings['app_key'],
consumer_secret=self.settings['app_secret'],
access_token_key=self.settings['token'],
access_token_secret=self.settings['token_secret'])
# Let's check if we really have some valid credentials
try:
connection.VerifyCredentials()
except twitter.error.TwitterError as error:
raise RunningError(colorama.Fore.RED + 'Twitter error: ' + error.message[0]['message'])
while processed <= self.settings['processing_limit']:
tweets = connection.GetUserTimeline(screen_name='dumpmon', max_id=max_id,
exclude_replies=True, include_rts=False, count=self.settings['limit'],
since_id=since_id)
import json as simplejson
except:
from django.utils import simplejson
from oauth import oauth
# Taken from oauth implementation at: http://github.com/harperreed/twitteroauth-python/tree/master
REQUEST_TOKEN_URL = 'https://twitter.com/oauth/request_token'
ACCESS_TOKEN_URL = 'https://twitter.com/oauth/access_token'
AUTHORIZATION_URL = 'http://twitter.com/oauth/authorize'
SIGNIN_URL = 'http://twitter.com/oauth/authenticate'
class OAuthApi(Api):
def __init__(self, consumer_key, consumer_secret, access_token=None):
if access_token:
Api.__init__(self,access_token.key, access_token.secret)
else:
Api.__init__(self)
self._Consumer = oauth.OAuthConsumer(consumer_key, consumer_secret)
self._signature_method = oauth.OAuthSignatureMethod_HMAC_SHA1()
self._access_token = access_token
def _GetOpener(self):
opener = self._urllib.build_opener()
return opener
def _FetchUrl(self,
url,
def get_twitter(self):
'''
returns user's latest twitter tweet
'''
twitter_conn = twitter.Api()
if self.twitter:
status = twitter_conn.GetUserTimeline(self.twitter)
for s in status:
return s.text
else:
return None
# teletext-twitter - creates pages for vbit2 teletext system
# (c) Mark Pentler 2018 (https://github.com/mpentler)
# see README.md for details on getting it running or run with -h
# main script
from output import *
import twitter
import time
import sys
import argparse
# Read config.py for our Twitter access keys etc
config = {}
exec(open("config.py").read(), config)
twitter_object = twitter.Api(access_token_key = config["access_key"],
access_token_secret = config["access_secret"],
consumer_key = config["consumer_key"],
consumer_secret = config["consumer_secret"],
sleep_on_rate_limit = True, # so we don't hit the rate limit and raise an exception
tweet_mode='extended')
def parse_args():
parser = argparse.ArgumentParser(description="Reads your Twitter timeline and turns it into teletext pages for your Raspberry Pi.")
parser.add_argument("-m", "--mode", type=str, help="choose between different modes - home, user or search")
parser.add_argument("-q", "--query", type=str, help="a search query, either a search term or a username. hashtags supported if you put quotes around the string")
parser.add_argument("-c", "--count", type=int, default=5, help="number of tweets to download (default is 5, capped at 200)")
parser.add_argument("-d", "--delay", type=int, default=60, help="seconds between timeline scrapes (default is 60 seconds - lower values have no effect)")
parser.add_argument("-n", "--no-repeat", action="store_true", default=False, help="only download tweets once - overrules -d switch")
parser.add_argument("-v", "--version", action="version", version="1.1.0")
parser.add_argument("-Q", "--quiet", action="store_true", default=False, help="suppresses all output to the terminal except warnings and errors")