python类Twitter()的实例源码

def _twitter_reconnect(self):

        """Logs in to Twitter, using the stored OAuth. This function is
        intended for internal use, and should ONLY be called after
        twitter_login has been called.
        """

        # Report the reconnection attempt.
        self._message(u'_twitter_reconnect', \
            u"Attempting to reconnect to Twitter.")

        # Raise an Exception if the twitter library wasn't imported
        if not IMPTWITTER:
            self._error(u'_twitter_reconnect', u"The 'twitter' library could not be imported. Check whether it is installed correctly.")

        # Log in to a Twitter account
        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()

        # Report the reconnection success.
        self._message(u'_twitter_reconnect', \
            u"Successfully reconnected to Twitter!")

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()

def __init__(self):
        ACCESS_TOKEN = self.access_token
        ACCESS_SECRET = self.access_secret
        CONSUMER_KEY = self.consumer_key
        CONSUMER_SECRET = self.consumer_secret
        print ACCESS_TOKEN, ACCESS_SECRET, CONSUMER_KEY, CONSUMER_SECRET
        oauth = OAuth(ACCESS_TOKEN, ACCESS_SECRET, CONSUMER_KEY, CONSUMER_SECRET)
        self.twitter = Twitter(auth=oauth)

def post_tweet(text, image_path=None,
               user_token=None, user_secret=None,
               consumer_key=None, consumer_secret=None):
    """
    Post a tweet, optionally with an image attached.

    """
    if len(text) > 140:
        raise ValueError('tweet is too long')

    auth = twitter.OAuth(
        user_token or os.environ['TWITTER_USER_TOKEN'],
        user_secret or os.environ['TWITTER_USER_SECRET'],
        consumer_key or os.environ['TWITTER_CONSUMER_KEY'],
        consumer_secret or os.environ['TWITTER_CONSUMER_SECRET'])

    image_data, image_id = None, None
    if image_path:
        with open(image_path, 'rb') as image_file:
            image_data = image_file.read()
        t_up = twitter.Twitter(domain='upload.twitter.com', auth=auth)
        image_id = t_up.media.upload(media=image_data)['media_id_string']

    if not (text.strip() or image_id):
        raise ValueError('no text or images to tweet')

    t_api = twitter.Twitter(auth=auth)

    params = {
        'status': text,
        'trim_user': True,
    }

    if image_id:
        params.update({
            'media[]': base64.b64encode(image_data),
            '_base64': True,
        })
        return t_api.statuses.update_with_media(**params)
    else:
        return t_api.statuses.update(**params)

def __init__(self, bot):
        super(TweetPlugin, self).__init__(bot)
        self.command = "!tweet"
        self.twitter = twitter.Twitter(
            auth=twitter.OAuth(
                config["PLUGINS"]["TWITTER_ATOKEN"],
                config["PLUGINS"]["TWITTER_ASECRET"],
                config["PLUGINS"]["TWITTER_CKEY"],
                config["PLUGINS"]["TWITTER_CSECRET"]
            )
        )

def long_desc(self):
        return ("Interface to Twitter. The bot's account is at"
                " https://twitter.com/relay_bot . Every tweet will be signed"
                " with the nickname of the user who sent it. ")

def oauth_login():
  CONSUMER_KEY = "[your own key]"
  CONSUMER_SECRET = "[your own secret]"

  OAUTH_TOKEN = "[your own token]"
  OAUTH_TOKEN_SECRET = "[your own secret]"

  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))

    #Set the appropriate settings for each alert

def send_alert(self, alert, info):
        args = { "status": replace(alert['status'], info)}
        try_sending(log, self.connect, "Twitter", self.client.statuses.update, args)

    #Trigger an alert based on Pokemon info

def __init__(self):
        self.t = Twitter(auth=OAuth( \
                    config.TOKEN,
                    config.TOKEN_KEY,
                    config.CON_SECRET,
                    config.CON_SECRET_KEY)
        )

def main():
    # Checks if this is the first time running the script
    # Allows the user to choose the Twitter Handles to follow
    # Sets the TWITTER_HANDLES which is an empty list, to the new updated list
    if INITIAL_START:
        handles = initial_start()
    else:
        handles = TWITTER_HANDLES

    # Sets up share_output job
    twit = Twitter()
    schedule.every().day.at("18:17").do(twit.share_output)
    schedule.every(15).minutes.do(company.current_day)

    while True:
        for handle in handles:
            twitter = Twitter(handle)

            new_tweet = twitter.get_latest_tweet()

            # Checks if a new tweet has been posted
            # If it has, checks for companies within the tweet
            if new_tweet:
                matches = company.check_for_companies(new_tweet, handle)

                # If there is a company mentioned
                if matches:
                    # Gets the initial company info
                    company.get_initial_company_info()

                    # Outputs the matches via twitter/email
                    twitter.initial_tweet(matches)
                    email(handle, matches)

            # Checks mentions for sign ups/removals
            twitter.check_mentions()

        # Gets current share price for each company being monitored
        # Checks if there are any schedules to be run
        company.get_current_shares()
        schedule.run_pending()

        now = datetime.now()
        print(f'Running: {now.hour}:{now.minute} - {now.day}/{now.month}/{now.year}')
        sleep(10)