"""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,

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

"""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 'Google App Engine' in os.environ.get('SERVER_SOFTWARE', ''):

import requests_toolbelt.adapters.appengine # Adapter ensures requests use app engine's urlfetch

requests_toolbelt.adapters.appengine.monkeypatch()

cache = None # App Engine does not like this caching strategy, disable caching

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

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

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 is not None and (access_token_key is None or

access_token_secret is None):

print('Twitter now requires an oAuth Access Token for API calls. '

'If you\'re using this library from a command line utility, '

'please run the included get_access_token.py tool to '

'generate one.', file=sys.stderr)

raise TwitterError({'message': "Twitter requires oAuth Access Token for all API access"})

self.SetCredentials(consumer_key, consumer_secret, access_token_key, access_token_secret)

if debugHTTP:

import logging

import http.client

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

def SetCredentials(self,

consumer_key,

consumer_secret,

access_token_key=None,

access_token_secret=None):

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

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 GetAccountSettings(self):

"""Returns dictionary of the current user's account settings.

url = "%s/account/settings.json" % self.base_url

resp = self._RequestUrl(url, 'GET')

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

return data

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.__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 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')

else:

resp = self._RequestUrl(url, 'GET', data=parameters)

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

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

if isinstance(category, Category):

slug = category.slug

elif isinstance(category, str):

slug = category

url = '%s/users/suggestions/%s.json' % (self.base_url, slug)

resp = self._RequestUrl(url, verb='GET')

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

users = []

for user in data.get('users', None):

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

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.

url = '%s/statuses/destroy/%s.json' % (self.base_url, status_id)

post_data = {

'id': enf_type('status_id', int, status_id),

'trim_user': enf_type('trim_user', bool, trim_user)

}

resp = self._RequestUrl(url, 'POST', data=post_data)

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

return Status.NewFromJsonDict(data)

def PostUpdate(self,

status,

media=None,

media_additional_owners=None,

media_category=None,

in_reply_to_status_id=None,

auto_populate_reply_metadata=False,

exclude_reply_user_ids=None,

attachment_url=None,

latitude=None,

longitude=None,

place_id=None,

display_coordinates=False,

trim_user=False,

verify_status_length=True):

"""Post a twitter status message from the authenticated user.