"""A python interface into the Twitter API

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,

verify_ssl=None,

cert_ssl=None):

"""Instantiate a new twitter.Api object.

# 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

self.verify_ssl = verify_ssl

self.cert_ssl = cert_ssl

self.base_url = base_url or 'https://api.twitter.com/1.1'

self.stream_url = stream_url or 'https://stream.twitter.com/1.1'

self.upload_url = upload_url or 'https://upload.twitter.com/1.1'

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

self._session = requests.Session()

@staticmethod

def GetAppOnlyAuthToken(consumer_key, 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

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.

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.

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

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', data=parameters)

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.

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

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

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

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

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

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.

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.

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.

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

request_url = '%s/statuses/oembed.json' % (self.base_url)

parameters = {}

if status_id is not None:

try:

parameters['id'] = int(status_id)

except ValueError:

raise TwitterError({'message': "'status_id' must be an integer."})

elif url is not None:

parameters['url'] = url

else:

raise TwitterError({'message': "Must specify either 'status_id' or 'url'"})

if maxwidth is not None:

parameters['maxwidth'] = maxwidth

if hide_media is True:

parameters['hide_media'] = 'true'

if hide_thread is True:

parameters['hide_thread'] = 'true'

if omit_script is True:

parameters['omit_script'] = 'true'

if align is not None:

if align not in ('left', 'center', 'right', 'none'):

raise TwitterError({'message': "'align' must be 'left', 'center', 'right', or 'none'"})

parameters['align'] = align

if related:

if not isinstance(related, str):

raise TwitterError({'message': "'related' should be a string of comma separated screen names"})

parameters['related'] = related

if lang is not None:

if not isinstance(lang, str):

raise TwitterError({'message': "'lang' should be string instance"})

parameters['lang'] = lang

resp = self._RequestUrl(request_url, 'GET', data=parameters, enforce_auth=False)

data = self._ParseAndCheckTwitter(resp.content.decode('utf-8'))

return data

def DestroyStatus(self, status_id, trim_user=False):

"""Destroys the status specified by the required ID parameter.