Python twython 模块，Twython() 实例源码

def tweet_arrival():
    options = UserOptions()
    options.read_settings()
    name = options.get('firstname') + ' ' + options.get('lastname')

    path = os.path.join(user_config_dir(), 'twitter.cfg')
    if not os.path.exists(path):
        print("Can't Tweet: {0} doesn't exist".format(path))
        return False
    cp = ConfigParser()
    cp.read(path)

    try:
        twitter = Twython(cp.get('twitter', 'app_key'),
                          cp.get('twitter', 'app_secret'),
                          cp.get('twitter', 'oauth_token'),
                          cp.get('twitter', 'oauth_token_secret'))
        twitter.update_status(
            status='Pleb %s has joined the botnet for good. #PlebNet #Cloudomate #Tribler #Bitcoin' % name)
        print("Tweeted arrival")
    except Exception, e:
        print e
        return False
    return True

def tweet_search(log, item, limit=50):
    log.debug("   Searching twitter for %s", item)
    check_twitter_config()
    if len(item) > 500:
        log.error("      Search string too long")
        raise Exception("Search string too long: %d", len(item))
    logging.captureWarnings(True)
    old_level = log.getEffectiveLevel()
    log.setLevel(logging.ERROR)
    twitter = Twython(APP_KEY, APP_SECRET, OAUTH_TOKEN, OAUTH_TOKEN_SECRET)
    try:
        result = twitter.search(q=item, count=limit)
    except TwythonAuthError, e:
        twitter_auth_issue(e)
        raise
    except:
        raise
    log.setLevel(old_level)
    return result

def check_relationship(log, id):
    my_screen_name = get_screen_name(log)
    if my_screen_name == "Unknown":
        raise("Couldn't get my own screen name")
    log.debug("      Checking relationship of %s with me (%s)", id, my_screen_name)
    check_twitter_config()
    logging.captureWarnings(True)
    old_level = log.getEffectiveLevel()
    log.setLevel(logging.ERROR)
    twitter = Twython(APP_KEY, APP_SECRET, OAUTH_TOKEN, OAUTH_TOKEN_SECRET)
    try:
        result = twitter.show_friendship(source_screen_name=my_screen_name, target_screen_name=id)
    except TwythonAuthError, e:
        log.setLevel(old_level)
        log.exception("   Problem trying to check relationship")
        twitter_auth_issue(e)
        raise
    except:
        raise
    log.setLevel(old_level)
    return result["relationship"]["source"]["following"], result["relationship"]["source"]["followed_by"]

def follow_twitter_user(log, id):
    log.debug("   Following %s", id)
    check_twitter_config()
    logging.captureWarnings(True)
    old_level = log.getEffectiveLevel()
    log.setLevel(logging.ERROR)
    twitter = Twython(APP_KEY, APP_SECRET, OAUTH_TOKEN, OAUTH_TOKEN_SECRET)
    try:
        twitter.create_friendship(screen_name=id)
    except TwythonAuthError, e:
        log.setLevel(old_level)
        log.exception("   Problem trying to follow twitter user")
        twitter_auth_issue(e)
        raise
    except:
        raise
    log.setLevel(old_level)

def unfollow_twitter_user(log, id):
    log.debug("   Unfollowing %s", id)
    check_twitter_config()
    logging.captureWarnings(True)
    old_level = log.getEffectiveLevel()
    log.setLevel(logging.ERROR)
    twitter = Twython(APP_KEY, APP_SECRET, OAUTH_TOKEN, OAUTH_TOKEN_SECRET)
    try:
        twitter.destroy_friendship(screen_name=id)
    except TwythonAuthError, e:
        log.setLevel(old_level)
        log.exception("Error unfollowing %s", id)
        twitter_auth_issue(e)
        raise
    except:
        log.exception("Error unfollowing %s", id)
    log.setLevel(old_level)

def get_screen_name(log):
    global MYSELF
    if not MYSELF or MYSELF == "Unknown":
        log.debug("   Getting current user screen name")
        check_twitter_config()
        logging.captureWarnings(True)
        old_level = log.getEffectiveLevel()
        log.setLevel(logging.ERROR)
        twitter = Twython(APP_KEY, APP_SECRET, OAUTH_TOKEN, OAUTH_TOKEN_SECRET)
        try:
            details = twitter.verify_credentials()
        except TwythonAuthError, e:
            log.setLevel(old_level)
            log.exception("   Problem trying to get screen name")
            twitter_auth_issue(e)
            raise
        except:
            log.exception("   Problem trying to get screen name")
            details = None
        log.setLevel(old_level)
        name = "Unknown"
        if details:
            name = details["screen_name"]
        MYSELF = name
    return MYSELF

def add_access_token(creds_file=None):
    """
    For OAuth 2, retrieve an access token for an app and append it to a
    credentials file.
    """
    if creds_file is None:
        path = os.path.dirname(__file__)
        creds_file = os.path.join(path, 'credentials2.txt')
    oauth2 = credsfromfile(creds_file=creds_file)
    app_key = oauth2['app_key']
    app_secret = oauth2['app_secret']

    twitter = Twython(app_key, app_secret, oauth_version=2)
    access_token = twitter.obtain_access_token()
    tok = 'access_token={}\n'.format(access_token)
    with open(creds_file, 'a') as infile:
        print(tok, file=infile)

def add_access_token(creds_file=None):
    """
    For OAuth 2, retrieve an access token for an app and append it to a
    credentials file.
    """
    if creds_file is None:
        path = os.path.dirname(__file__)
        creds_file = os.path.join(path, 'credentials2.txt')
    oauth2 = credsfromfile(creds_file=creds_file)
    app_key = oauth2['app_key']
    app_secret = oauth2['app_secret']

    twitter = Twython(app_key, app_secret, oauth_version=2)
    access_token = twitter.obtain_access_token()
    tok = 'access_token={}\n'.format(access_token)
    with open(creds_file, 'a') as infile:
        print(tok, file=infile)

def add_access_token(creds_file=None):
    """
    For OAuth 2, retrieve an access token for an app and append it to a
    credentials file.
    """
    if creds_file is None:
        path = os.path.dirname(__file__)
        creds_file = os.path.join(path, 'credentials2.txt')
    oauth2 = credsfromfile(creds_file=creds_file)
    app_key = oauth2['app_key']
    app_secret = oauth2['app_secret']

    twitter = Twython(app_key, app_secret, oauth_version=2)
    access_token = twitter.obtain_access_token()
    tok = 'access_token={}\n'.format(access_token)
    with open(creds_file, 'a') as infile:
        print(tok, file=infile)

def add_access_token(creds_file=None):
    """
    For OAuth 2, retrieve an access token for an app and append it to a
    credentials file.
    """
    if creds_file is None:
        path = os.path.dirname(__file__)
        creds_file = os.path.join(path, 'credentials2.txt')
    oauth2 = credsfromfile(creds_file=creds_file)
    app_key = oauth2['app_key']
    app_secret = oauth2['app_secret']

    twitter = Twython(app_key, app_secret, oauth_version=2)
    access_token = twitter.obtain_access_token()
    tok = 'access_token={}\n'.format(access_token)
    with open(creds_file, 'a') as infile:
        print(tok, file=infile)

def main():
    print(strftime("%a, %d %b %Y %H:%M:%S %Z", gmtime()), file=sys.stderr)
    print("Generating new talk", file=sys.stderr)

    talk = makeNewPage()

    print("Tweeting link to new talk at " + talk['url'], file=sys.stderr)

    #Tweet link here -- title is in talk['title']
    twitter = Twython(twitter_api, twitter_secret, twitter_access, twitter_access_secret)
    tweet = talk['title'] + ' ' + talk['url']
    twitter.update_status(status=tweet)

    print("Regenerating index file", file=sys.stderr)

    rebuildIndex()

    print("Done.", file=sys.stderr)

def add_access_token(creds_file=None):
    """
    For OAuth 2, retrieve an access token for an app and append it to a
    credentials file.
    """
    if creds_file is None:
        path = os.path.dirname(__file__)
        creds_file = os.path.join(path, 'credentials2.txt')
    oauth2 = credsfromfile(creds_file=creds_file)
    app_key = oauth2['app_key']
    app_secret = oauth2['app_secret']

    twitter = Twython(app_key, app_secret, oauth_version=2)
    access_token = twitter.obtain_access_token()
    tok = 'access_token={}\n'.format(access_token)
    with open(creds_file, 'a') as infile:
        print(tok, file=infile)

def main():
    with open('../cleaned_quiltdata/items.json', 'r') as f:
        items = json.load(f)
        twitter = Twython(APP_KEY, APP_SECRET, TOKEN, TOKEN_SECRET)

        if len(sys.argv) > 1:
            block_numbers = [number.zfill(5) for number in sys.argv[1:]]
        else:
            blocks_left = len(os.listdir(PICTURE_DIR))
            block_numbers = [os.listdir(PICTURE_DIR)[random.randrange(0, blocks_left)].split('.')[0]]

        for block_number in block_numbers:
            block_data = next((item for item in items if item['block_number'] == block_number))
            block_path = PICTURE_DIR + block_number + '.jpg'
            block_image = open(block_path, 'rb')
            names = block_data['names']

            print("tweeting block", block_number)
            response = twitter.upload_media(media=block_image)
            last_tweet = twitter.update_status(status="", media_ids=[response['media_id']])

            split_names(names, twitter, last_tweet)
            print("deleting block", block_number)
            os.remove(block_path)

def add_access_token(creds_file=None):
    """
    For OAuth 2, retrieve an access token for an app and append it to a
    credentials file.
    """
    if creds_file is None:
        path = os.path.dirname(__file__)
        creds_file = os.path.join(path, 'credentials2.txt')
    oauth2 = credsfromfile(creds_file=creds_file)
    app_key = oauth2['app_key']
    app_secret = oauth2['app_secret']

    twitter = Twython(app_key, app_secret, oauth_version=2)
    access_token = twitter.obtain_access_token()
    tok = 'access_token={}\n'.format(access_token)
    with open(creds_file, 'a') as infile:
        print(tok, file=infile)

def add_access_token(creds_file=None):
    """
    For OAuth 2, retrieve an access token for an app and append it to a
    credentials file.
    """
    if creds_file is None:
        path = os.path.dirname(__file__)
        creds_file = os.path.join(path, 'credentials2.txt')
    oauth2 = credsfromfile(creds_file=creds_file)
    app_key = oauth2['app_key']
    app_secret = oauth2['app_secret']

    twitter = Twython(app_key, app_secret, oauth_version=2)
    access_token = twitter.obtain_access_token()
    tok = 'access_token={}\n'.format(access_token)
    with open(creds_file, 'a') as infile:
        print(tok, file=infile)

def connect(self, do_login=False):
        app_key = self.config["app_key"]
        app_secet = self.config["app_secret"]
        oauth_token = self.config["oauth_token"]
        oauth_token_secret = self.config["oauth_token_secret"]
        self.twitter = Twython(app_key,
                               app_secet,
                               oauth_token,
                               oauth_token_secret)
        return None

def auth():
    CONSUMER_KEY = '*'
    CONSUMER_SECRET = '*'
    ACCESS_KEY = '*'
    ACCESS_SECRET = '*'

    twitter = Twython(CONSUMER_KEY,CONSUMER_SECRET,ACCESS_KEY,ACCESS_SECRET)

    return twitter

def tweet_string(message, log, media=None):
    check_twitter_config()
    logging.captureWarnings(True)
    old_level = log.getEffectiveLevel()

    log.setLevel(logging.ERROR)
    twitter = Twython(APP_KEY, APP_SECRET, OAUTH_TOKEN, OAUTH_TOKEN_SECRET)

    retries = 0
    while retries < 5:
        log.setLevel(logging.ERROR)
        try:
            if media:
                photo = open(media, 'rb')
                media_ids = twitter.upload_media(media=photo)
                twitter.update_status(status=message.encode('utf-8').strip(), media_ids=media_ids['media_id'])
            else:
                twitter.update_status(status=message.encode('utf-8').strip())
            break
        except TwythonAuthError, e:
            log.setLevel(old_level)
            log.exception("   Problem trying to tweet string")
            twitter_auth_issue(e)
            return
        except:
            log.setLevel(old_level)
            log.exception("   Problem trying to tweet string")
        retries += 1
        s = random.randrange(5, 10 * retries)
        log.debug("   sleeping %d seconds for retry", s)
        time.sleep(s)

    log.setLevel(old_level)
    if retries == 5:
        log.error("Couldn't tweet string: %s with media: %s", message, media)

def get_following(log, id):
    log.debug("  Getting people %s is following", id)
    check_twitter_config()
    logging.captureWarnings(True)
    old_level = log.getEffectiveLevel()
    log.setLevel(logging.ERROR)
    twitter = Twython(APP_KEY, APP_SECRET, OAUTH_TOKEN, OAUTH_TOKEN_SECRET)
    log.setLevel(old_level)

    cursor = -1
    max_loops = 15
    while cursor != 0:
        try:
            log.setLevel(logging.ERROR)
            following = twitter.get_friends_list(screen_name=id, cursor=cursor, count=200)
            log.setLevel(old_level)
        except TwythonAuthError, e:
            log.exception("   Problem trying to get people following")
            twitter_auth_issue(e)
            raise
        except:
            raise
        for u in following["users"]:
            yield u["screen_name"]
        cursor = following["next_cursor"]
        if cursor:
            s = random.randint(55, 65)
            log.debug("      Sleeping %ds to avoid rate limit. Cursor: %s", s, cursor)
            time.sleep(s)
        else:
            log.debug("      Normal query end")

        max_loops -= 1
        if max_loops <= 0:
            log.debug("      Killing search due to max loops")
            break
    log.setLevel(old_level)

def get_followers(log, id):
    log.debug("  Getting people following % s", id)
    check_twitter_config()
    logging.captureWarnings(True)
    old_level = log.getEffectiveLevel()
    log.setLevel(logging.ERROR)
    twitter = Twython(APP_KEY, APP_SECRET, OAUTH_TOKEN, OAUTH_TOKEN_SECRET)
    log.setLevel(old_level)

    cursor = -1
    max_loops = 15
    while cursor != 0:
        try:
            log.setLevel(logging.ERROR)
            following = twitter.get_followers_list(screen_name=id, cursor=cursor, count=200)
            log.setLevel(old_level)
        except TwythonAuthError, e:
            log.exception("   Problem trying to get people following")
            twitter_auth_issue(e)
            raise
        except:
            raise
        for u in following["users"]:
            yield u
        cursor = following["next_cursor"]
        if cursor:
            s = random.randint(55, 65)
            log.debug("      Sleeping %ds to avoid rate limit. Cursor: %s", s, cursor)
            time.sleep(s)
        else:
            log.debug("      Normal query end")

        max_loops -= 1
        if max_loops <= 0:
            log.debug("      Killing search due to max loops")
            break
    log.setLevel(old_level)

def get_twitter_instance():
    app_key, app_secret, oauth_token, oauth_token_secret = get_twitter_creds()
    return Twython(app_key, app_secret, oauth_token, oauth_token_secret)

def create_api(self, user):
        """
        creates an instance of the twython api
        """
        twitter_account = user.twitteraccount
        return Twython(CONSUMER_KEY,
                       CONSUMER_SECRET,
                       twitter_account.access_token,
                       twitter_account.access_secret)

def get(self, request):
        twitter = Twython(CONSUMER_KEY, CONSUMER_SECRET)

        # build the callback url that contains the pk of the user
        callback_url = request.build_absolute_uri(reverse('twitter:callback'))
                                                  #+ "?user_id=%s" % user_id)

        auth_tokens = twitter.get_authentication_tokens(callback_url)

        request.session['twitter_request_token'] = auth_tokens
        request.session['twitter_next_url'] = request.GET.get('next', None)

        return HttpResponseRedirect(auth_tokens['auth_url'])

def get(self, request):
        if not request.user.is_authenticated():
            return HttpResponseRedirect(login_url)

        default_url = '/'
        redirect_url = request.session.get('twitter_next_url')
        if not redirect_url:
            redirect_url = default_url

        verifier = request.GET.get('oauth_verifier')
        if not verifier:
            failure_url = '{}?status=error'.format(redirect_url)
            return HttpResponseRedirect(failure_url)

        # oauth token and secret are stored in the session
        oauth_token = request.session['twitter_request_token']['oauth_token']
        oauth_secret = request.session['twitter_request_token']['oauth_token_secret']
        # get the authorized tokens using the verifier
        twitter = Twython(CONSUMER_KEY, CONSUMER_SECRET,
                          oauth_token, oauth_secret)

        authorized_tokens = twitter.get_authorized_tokens(verifier)

        # create a TwitterAccount object to store token and secret
        try:
            # check if account already exists
            TwitterAccount.objects.get(user=request.user)
            success_url = '{}?status=success'.format(redirect_url)
            return HttpResponseRedirect(success_url)
        except:
            twitter_account = TwitterAccount()
            twitter_account.user = request.user
            twitter_account.access_token = authorized_tokens['oauth_token']
            twitter_account.access_secret = authorized_tokens['oauth_token_secret']
            twitter_account.save()

        success_url = '{}?status=success'.format(redirect_url)
        return HttpResponseRedirect(success_url)

def __init__(self):
        self.APP_KEY = "###"
        self.APP_SECRET = "###"

        self.twitter = Twython(self.APP_KEY, self.APP_SECRET, oauth_version=2)
        self.ACCESS_TOKEN = self.twitter.obtain_access_token()
        self.twitter = Twython(self.APP_KEY, access_token=self.ACCESS_TOKEN)

def __init__(self, username, query_string):
        super(TwitterFeedQuery, self).__init__(username, query_string)

        self.twitter = Twython(settings['CONSUMER_KEY'],
                               settings['CONSUMER_SECRET'],
                               settings['ACCESS_TOKEN_KEY'],
                               settings['ACCESS_TOKEN_SECRET'])

def __init__(self, user=None, callbacks=[], show_dupes=False, store=10):
        self.data = Storage()
        super(TwitterUserStream, self).__init__(*self.data.twitter)
        self.twitter = Twython(*self.data.twitter)

        self.error = "" # text of last error
        self._errors = 0
        self._max_errors = 5
        self._show_dupes = show_dupes
        self._follow = user
        self._lastn = list()
        self._store = store  # max size of _lastn
        self._callbacks = list()
        self.add_all(callbacks)

def __init__(self):

        self.configuration = ConfigParser.ConfigParser()
        self.directorycurrent = os.path.dirname(os.path.realpath(__file__))
        self.configuration.read(self.directorycurrent + '/configuration/credentials.config')
        self.consumer_key = self.configuration.get('twitter','consumer_key')
        self.consumer_secret = self.configuration.get('twitter','consumer_secret')
        self.access_token = self.configuration.get('twitter','access_token')
        self.access_token_secret = self.configuration.get('twitter','access_token_secret')
        self.twythonid = Twython(self.consumer_key, \
                                 self.consumer_secret, \
                                 self.access_token, \
                                 self.access_token_secret)

def get_user_timeline(screen_name, count=200):
    """Return list of most recent tweets posted by screen_name."""

    # ensure count is valid
    if count < 1 or count > 200:
        raise RuntimeError("invalid count")

    # ensure environment variables are set
    if not os.environ.get("API_KEY"):
        raise RuntimeError("API_KEY not set")
    if not os.environ.get("API_SECRET"):
        raise RuntimeError("API_SECRET not set")

    # get screen_name's most recent tweets
    # https://dev.twitter.com/rest/reference/get/users/lookup
    # https://dev.twitter.com/rest/reference/get/statuses/user_timeline
    # https://github.com/ryanmcgrath/twython/blob/master/twython/endpoints.py
    try:
        twitter = Twython(os.environ.get("API_KEY"), os.environ.get("API_SECRET"))
        user = twitter.lookup_user(screen_name=screen_name)
        if user[0]["protected"]:
            return None
        tweets = twitter.get_user_timeline(screen_name=screen_name, count=count)
        return [html.unescape(tweet["text"].replace("\n", " ")) for tweet in tweets]
    except TwythonAuthError as e:
        raise RuntimeError("invalid API_KEY and/or API_SECRET") from None
    except TwythonRateLimitError:
        raise RuntimeError("you've hit a rate limit") from None
    except TwythonError:
        return None

def send_tweet():
    twitter = Twython(
        consumer_key,
        consumer_secret,
        access_token,
        access_token_secret
    )

    # Send the tweet
    message = "The All Seeing Pi saw you!"
    with open(output, 'rb') as photo:
        twitter.update_status_with_media(status=message, media=photo)

# Set up buttons

def send_tweet():
    twitter = Twython(
        consumer_key,
        consumer_secret,
        access_token,
        access_token_secret
    )

    # Send the tweet
    message = "The All Seeing Pi saw you!"
    with open(output, 'rb') as photo:
        twitter.update_status_with_media(status=message, media=photo)

# Set up buttons

def __init__(self):
        config = self.parse_config()
        self.twitter = Twython(config.get('consumer_key'), config.get('consumer_secret'),
                               config.get('access_token'), config.get('access_token_secret'))

def Run(self):
      '''
         All the high-level logic of the bot is driven from here:
         - load settings
         - connect to twitter
         - (let your derived bot class get set up)
         - either:
            - wait for events from the streaming API
            - do bot stuff:
               - maybe create one or more tweets
               - handle any mentions
               - handle any streaming API events that were saved
               - send tweets out
         - (let your derived bot class clean up)
      '''
      self.LoadSettings()


      # create the Twython object that's going to communicate with the
      # twitter API.
      appKey = self.settings.appKey
      appSecret = self.settings.appSecret
      accessToken = self.settings.accessToken
      accessTokenSecret = self.settings.accessTokenSecret
      if self.stream:
         self.twitter = NanobotStreamer(appKey, appSecret, accessToken, accessTokenSecret)
         self.twitter.SetOutputPath(self.botPath)
      else:
         self.twitter = Twython(appKey, appSecret, accessToken, accessTokenSecret)


      # give the derived bot class a chance to do whatever it needs
      # to do before we actually execute. 
      self.PreRun()
      if self.stream:
         if self.debug:
            print "About to stream from user account."
         try:
            # The call to user() will sit forever waiting for events on 
            # our user account to stream down. Those events will be handled 
            # for us by the BotStreamer object that we created above
            self.twitter.user()
         except KeyboardInterrupt:
            # disconnect cleanly from the server.
            self.twitter.disconnect()
      else:
         self.CreateUpdate()
         self.HandleMentions()
         self.HandleStreamEvents()
         self.SendTweets()

         # if anything we did changed the settings, make sure those changes 
         # get written out.
         self.settings.lastExecuted = str(datetime.now())
         self.settings.Write()

      # ...and let the derived bot class clean up as it needs to.
      self.PostRun()

def twitter(query):

    ckey = os.environ['consumer_key']
    csecret = os.environ['consumer_secret']
    atoken = os.environ['access_token']
    asecret = os.environ['access_secret']

    twitter = Twython(ckey, csecret, atoken, asecret)

    try:
        search_results = twitter.search(q=query, languages = ['pt'] ,count=100)
    except TwythonError as e:
        print (e)

    reviews = []

    tweets = []



    for tweet in search_results['statuses']:
        if tweet['lang'].encode("utf-8") == 'pt':
            tweets.append(tweet['text'])

    if tweets == []:
        return [], [], []

    sents = sentiment(tweets)
    both = zip(tweets,sents)
    overall_sentiment = []
    count_pos = 0
    count_neutral = 0
    count_neg = 0
    for i in range(len(both)):
        sent_dict = {}
        sent_dict['tweet'] = both[i][0]
        sent_dict['sentimento'] = both[i][1]
        if sent_dict['sentimento'] == 0:
            sent_dict['sentimento'] = "negative"
            overall_sentiment.append(-1.0)
            count_neg = count_neg + 1
        elif sent_dict['sentimento'] == 1:
            sent_dict['sentimento'] = "neutral"
            overall_sentiment.append(0.0)
            count_neutral = count_neutral + 1
        elif sent_dict['sentimento'] == 2:
            sent_dict['sentimento'] = "positive"
            overall_sentiment.append(1.0)
            count_pos = count_pos + 1

        reviews.append(sent_dict)

    overall_sentiment = sum(overall_sentiment)/len(overall_sentiment)
    data = [count_neg, count_neutral, count_pos]
    return reviews, overall_sentiment, data