#!/usr/bin/env python
#
#
# Copyright 2007-2016 The Python-Twitter Developers
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
"""A library that provides a Python interface to the Twitter API"""
from __future__ import division
from __future__ import print_function
import json
import sys
import gzip
import time
import base64
import re
import logging
import requests
from requests_oauthlib import OAuth1, OAuth2
import io
import warnings
from uuid import uuid4
import os
try:
# python 3
from urllib.parse import urlparse, urlunparse, urlencode, quote_plus
from urllib.request import __version__ as urllib_version
except ImportError:
from urlparse import urlparse, urlunparse
from urllib import urlencode, quote_plus
from urllib import __version__ as urllib_version
from twitter import (
__version__,
_FileCache,
Category,
DirectMessage,
List,
Status,
Trend,
User,
UserStatus,
)
from twitter.ratelimit import RateLimit
from twitter.twitter_utils import (
calc_expected_status_length,
is_url,
parse_media_file,
enf_type,
parse_arg_list)
from twitter.error import (
TwitterError,
PythonTwitterDeprecationWarning330,
)
if sys.version_info > (3,):
long = int # pylint: disable=invalid-name,redefined-builtin
CHARACTER_LIMIT = 280
# A singleton representing a lazily instantiated FileCache.
DEFAULT_CACHE = object()
logger = logging.getLogger(__name__)
class Api(object):
"""A python interface into the Twitter API
By default, the Api caches results for 1 minute.
Example usage:
To create an instance of the twitter.Api class, with no authentication:
>>> import twitter
>>> api = twitter.Api()
To fetch a single user's public status messages, where "user" is either
a Twitter "short name" or their user id.
>>> statuses = api.GetUserTimeline(user)
>>> print([s.text for s in statuses])
To use authentication, instantiate the twitter.Api class with a
consumer key and secret; and the oAuth key and secret:
>>> api = twitter.Api(consumer_key='twitter consumer key',
consumer_secret='twitter consumer secret',
access_token_key='the_key_given',
access_token_secret='the_key_secret')
To fetch your friends (after being authenticated):
>>> users = api.GetFriends()
>>> print([u.name for u in users])
To post a twitter status message (after being authenticated):
>>> status = api.PostUpdate('I love python-twitter!')
>>> print(status.text)
I love python-twitter!
There are many other methods, including:
>>> api.PostUpdates(status)
>>> api.PostDirectMessage(user, text)
>>> api.GetUser(user)
>>> api.GetReplies()
>>> api.GetUserTimeline(user)
>>> api.GetHomeTimeline()
>>> api.GetStatus(status_id)
>>> api.GetStatuses(status_ids)
>>> api.DestroyStatus(status_id)
>>> api.GetFriends(user)
>>> api.GetFollowers()
>>> api.GetFeatured()
>>> api.GetDirectMessages()
>>> api.GetSentDirectMessages()
>>> api.PostDirectMessage(user, text)
>>> api.DestroyDirectMessage(message_id)
>>> api.DestroyFriendship(user)
>>> api.CreateFriendship(user)
>>> api.LookupFriendship(user)
>>> api.VerifyCredentials()
"""
DEFAULT_CACHE_TIMEOUT = 60 # cache for 1 minute
_API_REALM = 'Twitter API'
def __init__(self,
consumer_key=None,
consumer_secret=None,
access_token_key=None,
access_token_secret=None,
application_only_auth=False,
input_encoding=None,
request_headers=None,
cache=DEFAULT_CACHE,
base_url=None,
stream_url=None,
upload_url=None,
chunk_size=1024 * 1024,
use_gzip_compression=False,
debugHTTP=False,
timeout=None,
sleep_on_rate_limit=False,
tweet_mode='compat',
proxies=None):
"""Instantiate a new twitter.Api object.
Args:
consumer_key (str):
Your Twitter user's consumer_key.
consumer_secret (str):
Your Twitter user's consumer_secret.
access_token_key (str):
The oAuth access token key value you retrieved
from running get_access_token.py.
access_token_secret (str):
The oAuth access token's secret, also retrieved
from the get_access_token.py run.
application_only_auth:
Use Application-Only Auth instead of User Auth.
Defaults to False [Optional]
input_encoding (str, optional):
The encoding used to encode input strings.
request_header (dict, optional):
A dictionary of additional HTTP request headers.
cache (object, optional):
The cache instance to use. Defaults to DEFAULT_CACHE.
Use None to disable caching.
base_url (str, optional):
The base URL to use to contact the Twitter API.
Defaults to https://api.twitter.com.
stream_url (str, optional):
The base URL to use for streaming endpoints.
Defaults to 'https://stream.twitter.com/1.1'.
upload_url (str, optional):
The base URL to use for uploads. Defaults to 'https://upload.twitter.com/1.1'.
chunk_size (int, optional):
Chunk size to use for chunked (multi-part) uploads of images/videos/gifs.
Defaults to 1MB. Anything under 16KB and you run the risk of erroring out
on 15MB files.
use_gzip_compression (bool, optional):
Set to True to tell enable gzip compression for any call
made to Twitter. Defaults to False.
debugHTTP (bool, optional):
Set to True to enable debug output from urllib2 when performing
any HTTP requests. Defaults to False.
timeout (int, optional):
Set timeout (in seconds) of the http/https requests. If None the
requests lib default will be used. Defaults to None.
sleep_on_rate_limit (bool, optional):
Whether to sleep an appropriate amount of time if a rate limit is hit for
an endpoint.
tweet_mode (str, optional):
Whether to use the new (as of Sept. 2016) extended tweet mode. See docs for
details. Choices are ['compatibility', 'extended'].
proxies (dict, optional):
A dictionary of proxies for the request to pass through, if not specified
allows requests lib to use environmental variables for proxy if any.
"""
# check to see if the library is running on a Google App Engine instance
# see GAE.rst for more information
if os.environ:
if 'APPENGINE_RUNTIME' in os.environ.keys():
# Adapter ensures requests use app engine's urlfetch
import requests_toolbelt.adapters.appengine
requests_toolbelt.adapters.appengine.monkeypatch()
# App Engine does not like this caching strategy, disable caching
cache = None
self.SetCache(cache)
self._cache_timeout = Api.DEFAULT_CACHE_TIMEOUT
self._input_encoding = input_encoding
self._use_gzip = use_gzip_compression
self._debugHTTP = debugHTTP
self._shortlink_size = 19
if timeout and timeout < 30:
warnings.warn("Warning: The Twitter streaming API sends 30s keepalives, the given timeout is shorter!")
self._timeout = timeout
self.__auth = None
self._InitializeRequestHeaders(request_headers)
self._InitializeUserAgent()
self._InitializeDefaultParameters()
self.rate_limit = RateLimit()
self.sleep_on_rate_limit = sleep_on_rate_limit
self.tweet_mode = tweet_mode
self.proxies = proxies
if base_url is None:
self.base_url = 'https://api.twitter.com/1.1'
else:
self.base_url = base_url
if stream_url is None:
self.stream_url = 'https://stream.twitter.com/1.1'
else:
self.stream_url = stream_url
if upload_url is None:
self.upload_url = 'https://upload.twitter.com/1.1'
else:
self.upload_url = upload_url
self.chunk_size = chunk_size
if self.chunk_size < 1024 * 16:
warnings.warn((
"A chunk size lower than 16384 may result in too many "
"requests to the Twitter API when uploading videos. You are "
"strongly advised to increase it above 16384"))
if (consumer_key and not
(application_only_auth or all([access_token_key, access_token_secret]))):
raise TwitterError({'message': "Missing oAuth Consumer Key or Access Token"})
self.SetCredentials(consumer_key, consumer_secret, access_token_key, access_token_secret,
application_only_auth)
if debugHTTP:
try:
import http.client as http_client # python3
except ImportError:
import httplib as http_client # python2
http_client.HTTPConnection.debuglevel = 1
logging.basicConfig() # you need to initialize logging, otherwise you will not see anything from requests
logging.getLogger().setLevel(logging.DEBUG)
requests_log = logging.getLogger("requests.packages.urllib3")
requests_log.setLevel(logging.DEBUG)
requests_log.propagate = True
@staticmethod
def GetAppOnlyAuthToken(consumer_key, consumer_secret):
"""
Generate a Bearer Token from consumer_key and consumer_secret
"""
key = quote_plus(consumer_key)
secret = quote_plus(consumer_secret)
bearer_token = base64.b64encode('{}:{}'.format(key, secret).encode('utf8'))
post_headers = {
'Authorization': 'Basic {0}'.format(bearer_token.decode('utf8')),
'Content-Type': 'application/x-www-form-urlencoded;charset=UTF-8'
}
res = requests.post(url='https://api.twitter.com/oauth2/token',
data={'grant_type': 'client_credentials'},
headers=post_headers)
bearer_creds = res.json()
return bearer_creds
def SetCredentials(self,
consumer_key,
consumer_secret,
access_token_key=None,
access_token_secret=None,
application_only_auth=False):
"""Set the consumer_key and consumer_secret for this instance
Args:
consumer_key:
The consumer_key of the twitter account.
consumer_secret:
The consumer_secret for the twitter account.
access_token_key:
The oAuth access token key value you retrieved
from running get_access_token.py.
access_token_secret:
The oAuth access token's secret, also retrieved
from the get_access_token.py run.
application_only_auth:
Whether to generate a bearer token and use Application-Only Auth
"""
self._consumer_key = consumer_key
self._consumer_secret = consumer_secret
self._access_token_key = access_token_key
self._access_token_secret = access_token_secret
if application_only_auth:
self._bearer_token = self.GetAppOnlyAuthToken(consumer_key, consumer_secret)
self.__auth = OAuth2(token=self._bearer_token)
else:
auth_list = [consumer_key, consumer_secret,
access_token_key, access_token_secret]
if all(auth_list):
self.__auth = OAuth1(consumer_key, consumer_secret,
access_token_key, access_token_secret)
self._config = None
def GetHelpConfiguration(self):
"""Get basic help configuration details from Twitter.
Args:
None
Returns:
dict: Sets self._config and returns dict of help config values.
"""
if self._config is None:
url = '%s/help/configuration.json' % self.base_url
resp = self._RequestUrl(url, 'GET')
data = self._ParseAndCheckTwitter(resp.content.decode('utf-8'))
self._config = data
return self._config
def GetShortUrlLength(self, https=False):
"""Returns number of characters reserved per URL included in a tweet.
Args:
https (bool, optional):
If True, return number of characters reserved for https urls
or, if False, return number of character reserved for http urls.
Returns:
(int): Number of characters reserved per URL.
"""
config = self.GetHelpConfiguration()
if https:
return config['short_url_length_https']
else:
return config['short_url_length']
def ClearCredentials(self):
"""Clear any credentials for this instance
"""
self._consumer_key = None
self._consumer_secret = None
self._access_token_key = None
self._access_token_secret = None
self._bearer_token = None
self.__auth = None # for request upgrade
def GetSearch(self,
term=None,
raw_query=None,
geocode=None,
since_id=None,
max_id=None,
until=None,
since=None,
count=15,
lang=None,
locale=None,
result_type="mixed",
include_entities=None,
return_json=False):
"""Return twitter search results for a given term. You must specify one
of term, geocode, or raw_query.
Args:
term (str, optional):
Term to search by. Optional if you include geocode.
raw_query (str, optional):
A raw query as a string. This should be everything after the "?" in
the URL (i.e., the query parameters). You are responsible for all
type checking and ensuring that the query string is properly
formatted, as it will only be URL-encoded before be passed directly
to Twitter with no other checks performed. For advanced usage only.
*This will override any other parameters passed*
since_id (int, optional):
Returns results with an ID greater than (that is, more recent
than) the specified ID. There are limits to the number of
Tweets which can be accessed through the API. If the limit of
Tweets has occurred since the since_id, the since_id will be
forced to the oldest ID available.
max_id (int, optional):
Returns only statuses with an ID less than (that is, older
than) or equal to the specified ID.
until (str, optional):
Returns tweets generated before the given date. Date should be
formatted as YYYY-MM-DD.
since (str, optional):
Returns tweets generated since the given date. Date should be
formatted as YYYY-MM-DD.
geocode (str or list or tuple, optional):
Geolocation within which to search for tweets. Can be either a
string in the form of "latitude,longitude,radius" where latitude
and longitude are floats and radius is a string such as "1mi" or
"1km" ("mi" or "km" are the only units allowed). For example:
>>> api.GetSearch(geocode="37.781157,-122.398720,1mi").
Otherwise, you can pass a list of either floats or strings for
lat/long and a string for radius:
>>> api.GetSearch(geocode=[37.781157, -122.398720, "1mi"])
>>> # or:
>>> api.GetSearch(geocode=(37.781157, -122.398720, "1mi"))
>>> # or:
>>> api.GetSearch(geocode=("37.781157", "-122.398720", "1mi"))
count (int, optional):
Number of results to return. Default is 15 and maxmimum that
Twitter returns is 100 irrespective of what you type in.
lang (str, optional):
Language for results as ISO 639-1 code. Default is None
(all languages).
locale (str, optional):
Language of the search query. Currently only 'ja' is effective.
This is intended for language-specific consumers and the default
should work in the majority of cases.
result_type (str, optional):
Type of result which should be returned. Default is "mixed".
Valid options are "mixed, "recent", and "popular".
include_entities (bool, optional):
If True, each tweet will include a node called "entities".
This node offers a variety of metadata about the tweet in a
discrete structure, including: user_mentions, urls, and
hashtags.
return_json (bool, optional):
If True JSON data will be returned, instead of twitter.Userret
Returns:
list: A sequence of twitter.Status instances, one for each message
containing the term, within the bounds of the geocoded area, or
given by the raw_query.
"""
url = '%s/search/tweets.json' % self.base_url
parameters = {}
if since_id:
parameters['since_id'] = enf_type('since_id', int, since_id)
if max_id:
parameters['max_id'] = enf_type('max_id', int, max_id)
if until:
parameters['until'] = enf_type('until', str, until)
if since:
parameters['since'] = enf_type('since', str, since)
if lang:
parameters['lang'] = enf_type('lang', str, lang)
if locale:
parameters['locale'] = enf_type('locale', str, locale)
if term is None and geocode is None and raw_query is None:
return []
if term is not None:
parameters['q'] = term
if geocode is not None:
if isinstance(geocode, list) or isinstance(geocode, tuple):
parameters['geocode'] = ','.join([str(geo) for geo in geocode])
else:
parameters['geocode'] = enf_type('geocode', str, geocode)
if include_entities:
parameters['include_entities'] = enf_type('include_entities',
bool,
include_entities)
parameters['count'] = enf_type('count', int, count)
if result_type in ["mixed", "popular", "recent"]:
parameters['result_type'] = result_type
if raw_query is not None:
url = "{url}?{raw_query}".format(
url=url,
raw_query=raw_query)
resp = self._RequestUrl(url, 'GET')
else:
resp = self._RequestUrl(url, 'GET', data=parameters)
data = self._ParseAndCheckTwitter(resp.content.decode('utf-8'))
if return_json:
return data
else:
return [Status.NewFromJsonDict(x) for x in data.get('statuses', '')]
def GetUsersSearch(self,
term=None,
page=1,
count=20,
include_entities=None):
"""Return twitter user search results for a given term.
Args:
term:
Term to search by.
page:
Page of results to return. Default is 1
[Optional]
count:
Number of results to return. Default is 20
[Optional]
include_entities:
If True, each tweet will include a node called "entities,".
This node offers a variety of metadata about the tweet in a
discrete structure, including: user_mentions, urls, and hashtags.
[Optional]
Returns:
A sequence of twitter.User instances, one for each message containing
the term
"""
# Build request parameters
parameters = {}
if term is not None:
parameters['q'] = term
if page != 1:
parameters['page'] = page
if include_entities:
parameters['include_entities'] = 1
try:
parameters['count'] = int(count)
except ValueError:
raise TwitterError({'message': "count must be an integer"})
# Make and send requests
url = '%s/users/search.json' % self.base_url
resp = self._RequestUrl(url, 'GET', data=parameters)
data = self._ParseAndCheckTwitter(resp.content.decode('utf-8'))
return [User.NewFromJsonDict(x) for x in data]
def GetTrendsCurrent(self, exclude=None):
"""Get the current top trending topics (global)
Args:
exclude:
Appends the exclude parameter as a request parameter.
Currently only exclude=hashtags is supported. [Optional]
Returns:
A list with 10 entries. Each entry contains a trend.
"""
return self.GetTrendsWoeid(woeid=1, exclude=exclude)
def GetTrendsWoeid(self, woeid, exclude=None):
"""Return the top 10 trending topics for a specific WOEID, if trending
information is available for it.
Args:
woeid:
the Yahoo! Where On Earth ID for a location.
exclude:
Appends the exclude parameter as a request parameter.
Currently only exclude=hashtags is supported. [Optional]
Returns:
A list with 10 entries. Each entry contains a trend.
"""
url = '%s/trends/place.json' % (self.base_url)
parameters = {'id': woeid}
if exclude:
parameters['exclude'] = exclude
resp = self._RequestUrl(url, verb='GET', data=parameters)
data = self._ParseAndCheckTwitter(resp.content.decode('utf-8'))
trends = []
timestamp = data[0]['as_of']
for trend in data[0]['trends']:
trends.append(Trend.NewFromJsonDict(trend, timestamp=timestamp))
return trends
def GetUserSuggestionCategories(self):
""" Return the list of suggested user categories, this can be used in
GetUserSuggestion function
Returns:
A list of categories
"""
url = '%s/users/suggestions.json' % (self.base_url)
resp = self._RequestUrl(url, verb='GET')
data = self._ParseAndCheckTwitter(resp.content.decode('utf-8'))
categories = []
for category in data:
categories.append(Category.NewFromJsonDict(category))
return categories
def GetUserSuggestion(self, category):
""" Returns a list of users in a category
Args:
category:
The Category object to limit the search by
Returns:
A list of users in that category
"""
url = '%s/users/suggestions/%s.json' % (self.base_url, category.slug)
resp = self._RequestUrl(url, verb='GET')
data = self._ParseAndCheckTwitter(resp.content.decode('utf-8'))
users = []
for user in data['users']:
users.append(User.NewFromJsonDict(user))
return users
def GetHomeTimeline(self,
count=None,
since_id=None,
max_id=None,
trim_user=False,
exclude_replies=False,
contributor_details=False,
include_entities=True):
"""Fetch a collection of the most recent Tweets and retweets posted
by the authenticating user and the users they follow.
The home timeline is central to how most users interact with Twitter.
Args:
count:
Specifies the number of statuses to retrieve. May not be
greater than 200. Defaults to 20. [Optional]
since_id:
Returns results with an ID greater than (that is, more recent
than) the specified ID. There are limits to the number of
Tweets which can be accessed through the API. If the limit of
Tweets has occurred since the since_id, the since_id will be
forced to the oldest ID available. [Optional]
max_id:
Returns results with an ID less than (that is, older than) or
equal to the specified ID. [Optional]
trim_user:
When True, each tweet returned in a timeline will include a user
object including only the status authors numerical ID. Omit this
parameter to receive the complete user object. [Optional]
exclude_replies:
This parameter will prevent replies from appearing in the
returned timeline. Using exclude_replies with the count
parameter will mean you will receive up-to count tweets -
this is because the count parameter retrieves that many
tweets before filtering out retweets and replies. [Optional]
contributor_details:
This parameter enhances the contributors element of the
status response to include the screen_name of the contributor.
By default only the user_id of the contributor is included. [Optional]
include_entities:
The entities node will be disincluded when set to false.
This node offers a variety of metadata about the tweet in a
discreet structure, including: user_mentions, urls, and
hashtags. [Optional]
Returns:
A sequence of twitter.Status instances, one for each message
"""
url = '%s/statuses/home_timeline.json' % self.base_url
parameters = {}
if count is not None:
try:
if int(count) > 200:
raise TwitterError({'message': "'count' may not be greater than 200"})
except ValueError:
raise TwitterError({'message': "'count' must be an integer"})
parameters['count'] = count
if since_id:
try:
parameters['since_id'] = int(since_id)
except ValueError:
raise TwitterError({'message': "'since_id' must be an integer"})
if max_id:
try:
parameters['max_id'] = int(max_id)
except ValueError:
raise TwitterError({'message': "'max_id' must be an integer"})
if trim_user:
parameters['trim_user'] = 1
if exclude_replies:
parameters['exclude_replies'] = 1
if contributor_details:
parameters['contributor_details'] = 1
if not include_entities:
parameters['include_entities'] = 'false'
resp = self._RequestUrl(url, 'GET', data=parameters)
data = self._ParseAndCheckTwitter(resp.content.decode('utf-8'))
return [Status.NewFromJsonDict(x) for x in data]
def GetUserTimeline(self,
user_id=None,
screen_name=None,
since_id=None,
max_id=None,
count=None,
include_rts=True,
trim_user=False,
exclude_replies=False):
"""Fetch the sequence of public Status messages for a single user.
The twitter.Api instance must be authenticated if the user is private.
Args:
user_id (int, optional):
Specifies the ID of the user for whom to return the
user_timeline. Helpful for disambiguating when a valid user ID
is also a valid screen name.
screen_name (str, optional):
Specifies the screen name of the user for whom to return the
user_timeline. Helpful for disambiguating when a valid screen
name is also a user ID.
since_id (int, optional):
Returns results with an ID greater than (that is, more recent
than) the specified ID. There are limits to the number of
Tweets which can be accessed through the API. If the limit of
Tweets has occurred since the since_id, the since_id will be
forced to the oldest ID available.
max_id (int, optional):
Returns only statuses with an ID less than (that is, older
than) or equal to the specified ID.
count (int, optional):
Specifies the number of statuses to retrieve. May not be
greater than 200.
include_rts (bool, optional):
If True, the timeline will contain native retweets (if they
exist) in addition to the standard stream of tweets.
trim_user (bool, optional):
If True, statuses will only contain the numerical user ID only.
Otherwise a full user object will be returned for each status.
exclude_replies (bool, optional)
If True, this will prevent replies from appearing in the returned
timeline. Using exclude_replies with the count parameter will mean you
will receive up-to count tweets - this is because the count parameter
retrieves that many tweets before filtering out retweets and replies.
This parameter is only supported for JSON and XML responses.
Returns:
A sequence of Status instances, one for each message up to count
"""
url = '%s/statuses/user_timeline.json' % (self.base_url)
parameters = {}
if user_id:
parameters['user_id'] = enf_type('user_id', int, user_id)
elif screen_name:
parameters['screen_name'] = screen_name
if since_id:
parameters['since_id'] = enf_type('since_id', int, since_id)
if max_id:
parameters['max_id'] = enf_type('max_id', int, max_id)
if count:
parameters['count'] = enf_type('count', int, count)
parameters['include_rts'] = enf_type('include_rts', bool, include_rts)
parameters['trim_user'] = enf_type('trim_user', bool, trim_user)
parameters['exclude_replies'] = enf_type('exclude_replies', bool, exclude_replies)
resp = self._RequestUrl(url, 'GET', data=parameters)
data = self._ParseAndCheckTwitter(resp.content.decode('utf-8'))
return [Status.NewFromJsonDict(x) for x in data]
def GetStatus(self,
status_id,
trim_user=False,
include_my_retweet=True,
include_entities=True,
include_ext_alt_text=True):
"""Returns a single status message, specified by the status_id parameter.
Args:
status_id:
The numeric ID of the status you are trying to retrieve.
trim_user:
When set to True, each tweet returned in a timeline will include
a user object including only the status authors numerical ID.
Omit this parameter to receive the complete user object. [Optional]
include_my_retweet:
When set to True, any Tweets returned that have been retweeted by
the authenticating user will include an additional
current_user_retweet node, containing the ID of the source status
for the retweet. [Optional]
include_entities:
If False, the entities node will be disincluded.
This node offers a variety of metadata about the tweet in a
discreet structure, including: user_mentions, urls, and
hashtags. [Optional]
Returns:
A twitter.Status instance representing that status message
"""
url = '%s/statuses/show.json' % (self.base_url)
parameters = {
'id': enf_type('status_id', int, status_id),
'trim_user': enf_type('trim_user', bool, trim_user),
'include_my_retweet': enf_type('include_my_retweet', bool, include_my_retweet),
'include_entities': enf_type('include_entities', bool, include_entities),
'include_ext_alt_text': enf_type('include_ext_alt_text', bool, include_ext_alt_text)
}
resp = self._RequestUrl(url, 'GET', data=parameters)
data = self._ParseAndCheckTwitter(resp.content.decode('utf-8'))
return Status.NewFromJsonDict(data)
def GetStatuses(self,
status_ids,
trim_user=False,
include_entities=True,
map=False):
"""Returns a list of status messages, specified by the status_ids parameter.
Args:
status_ids:
A list of the numeric ID of the statuses you are trying to retrieve.
trim_user:
When set to True, each tweet returned in a timeline will include
a user object including only the status authors numerical ID.
Omit this parameter to receive the complete user object. [Optional]
include_entities:
If False, the entities node will be disincluded.
This node offers a variety of metadata about the tweet in a
discreet structure, including: user_mentions, urls, and
hashtags. [Optional]
map:
If True, returns a dictionary with status id as key and returned
status data (or None if tweet does not exist or is inaccessible)
as value. Otherwise returns an unordered list of successfully
retrieved Tweets. [Optional]
Returns:
A dictionary or unordered list (depending on the parameter 'map') of
twitter Status instances representing the status messages.
"""
url = '%s/statuses/lookup.json' % (self.base_url)
map = enf_type('map', bool, map)
if map:
result = {}
else:
result = []
offset = 0
parameters = {
'trim_user': enf_type('trim_user', bool, trim_user),
'include_entities': enf_type('include_entities', bool, include_entities),
'map': map
}
while offset < len(status_ids):
parameters['id'] = ','.join([str(enf_type('status_id', int, status_id)) for status_id in status_ids[offset:offset + 100]])
resp = self._RequestUrl(url, 'GET', data=parameters)
data = self._ParseAndCheckTwitter(resp.content.decode('utf-8'))
if map:
result.update({int(key): (Status.NewFromJsonDict(value) if value else None) for key, value in data['id'].items()})
else:
result += [Status.NewFromJsonDict(dataitem) for dataitem in data]
offset += 100
return result
def GetStatusOembed(self,
status_id=None,
url=None,
maxwidth=None,
hide_media=False,
hide_thread=False,
omit_script=False,
align=None,
related=None,
lang=None):
"""Returns information allowing the creation of an embedded representation of a
Tweet on third party sites.
Specify tweet by the id or url parameter.
Args:
status_id:
The numeric ID of the status you are trying to embed.
url:
The url of the status you are trying to embed.
maxwidth:
The maximum width in pixels that the embed should be rendered at.
This value is constrained to be between 250 and 550 pixels. [Optional]
hide_media:
Specifies whether the embedded Tweet should automatically expand images. [Optional]
hide_thread:
Specifies whether the embedded Tweet should automatically show the original
message in the case that the embedded Tweet is a reply. [Optional]
omit_script:
Specifies whether the embedded Tweet HTML should include a