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

shortner=None,

base_url=None,

stream_url=None,

upload_url=None,

use_gzip_compression=False,

debugHTTP=False,

timeout=None):

"""Instantiate a new twitter.Api object.

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._InitializeRequestHeaders(request_headers)

self._InitializeUserAgent()

self._InitializeDefaultParameters()

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

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.', file=sys.stderr)

print('If your using this library from a command line utility, please', file=sys.stderr)

print('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 GetHelpConfiguration(self):

if self._config is None:

url = '%s/help/configuration.json' % self.base_url

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

data = self._ParseAndCheckTwitter(json.content)

self._config = data

return self._config

def GetShortUrlLength(self, https=False):

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,

geocode=None,

since_id=None,

max_id=None,

until=None,

count=15,

lang=None,

locale=None,

result_type="mixed",

include_entities=None):

"""Return twitter search results for a given term.

# Build request parameters

parameters = {}

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 until:

parameters['until'] = until

if lang:

parameters['lang'] = lang

if locale:

parameters['locale'] = locale

if term is None and geocode is None:

return []

if term is not None:

parameters['q'] = term

if geocode is not None:

parameters['geocode'] = ','.join(map(str, geocode))

if include_entities:

parameters['include_entities'] = 1

try:

parameters['count'] = int(count)

except ValueError:

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

if result_type in ["mixed", "popular", "recent"]:

parameters['result_type'] = result_type

# Make and send requests

url = '%s/search/tweets.json' % self.base_url

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

data = self._ParseAndCheckTwitter(json.content)

# Return built list of statuses

return [Status.NewFromJsonDict(x) for x in data['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

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

data = self._ParseAndCheckTwitter(json.content)

return [User.NewFromJsonDict(x) for x in data]

def GetTrendsCurrent(self, exclude=None):

"""Get the current top trending topics (global)

return self.GetTrendsWoeid(id=1, exclude=exclude)

def GetTrendsWoeid(self, id, exclude=None):

"""Return the top 10 trending topics for a specific WOEID, if trending

url = '%s/trends/place.json' % (self.base_url)

parameters = {'id': id}

if exclude:

parameters['exclude'] = exclude

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

data = self._ParseAndCheckTwitter(json.content)

trends = []

timestamp = data[0]['as_of']

for trend in data[0]['trends']:

trends.append(Trend.NewFromJsonDict(trend, timestamp=timestamp))

return trends

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

if not self.__auth:

raise TwitterError({'message': "API must be authenticated."})

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'

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

data = self._ParseAndCheckTwitter(json.content)

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=None,

exclude_replies=None):

"""Fetch the sequence of public Status messages for a single user.

parameters = {}

url = '%s/statuses/user_timeline.json' % (self.base_url)

if user_id:

parameters['user_id'] = user_id

elif screen_name:

parameters['screen_name'] = screen_name

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 count:

try:

parameters['count'] = int(count)

except ValueError:

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

if not include_rts:

parameters['include_rts'] = 0

if trim_user:

parameters['trim_user'] = 1

if exclude_replies:

parameters['exclude_replies'] = 1

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

data = self._ParseAndCheckTwitter(json.content)

return [Status.NewFromJsonDict(x) for x in data]

def GetStatus(self,

id,

trim_user=False,

include_my_retweet=True,

include_entities=True):

"""Returns a single status message, specified by the id parameter.

url = '%s/statuses/show.json' % (self.base_url)

if not self.__auth:

raise TwitterError({'message': "API must be authenticated."})

parameters = {}

try:

parameters['id'] = int(id)

except ValueError:

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

if trim_user:

parameters['trim_user'] = 1

if include_my_retweet:

parameters['include_my_retweet'] = 1

if not include_entities:

parameters['include_entities'] = 'none'

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

data = self._ParseAndCheckTwitter(json.content)

return Status.NewFromJsonDict(data)

def GetStatusOembed(self,

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)

if not self.__auth:

raise TwitterError({'message': "API must be authenticated."})

parameters = {}

if id is not None:

try:

parameters['id'] = int(id)

except ValueError:

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

elif url is not None:

parameters['url'] = url

else:

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

if maxwidth is not None:

parameters['maxwidth'] = maxwidth

if hide_media == True:

parameters['hide_media'] = 'true'

if hide_thread == True:

parameters['hide_thread'] = 'true'

if omit_script == 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

json = self._RequestUrl(request_url, 'GET', data=parameters)

data = self._ParseAndCheckTwitter(json.content)

return data

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

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

if not self.__auth:

raise TwitterError({'message': "API must be authenticated."})

try:

post_data = {'id': int(id)}

except ValueError:

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

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

if trim_user:

post_data['trim_user'] = 1

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

data = self._ParseAndCheckTwitter(json.content)

return Status.NewFromJsonDict(data)

@classmethod

def _calculate_status_length(cls, status, linksize=19):

dummy_link_replacement = 'https://-%d-chars%s/' % (linksize, '-' * (linksize - 18))

shortened = ' '.join([x if not (x.startswith('http://') or

x.startswith('https://'))

else

dummy_link_replacement

for x in status.split(' ')])

return len(shortened)

def PostUpdate(self,

status,

in_reply_to_status_id=None,

latitude=None,

longitude=None,

place_id=None,

display_coordinates=False,

trim_user=False):

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

if not self.__auth:

raise TwitterError({'message': "The twitter.Api instance must be authenticated."})

url = '%s/statuses/update.json' % self.base_url

if isinstance(status, str) or self._input_encoding is None:

u_status = status

else:

u_status = str(status, self._input_encoding)

# if self._calculate_status_length(u_status, self._shortlink_size) > CHARACTER_LIMIT:

# raise TwitterError("Text must be less than or equal to %d characters. "

# "Consider using PostUpdates." % CHARACTER_LIMIT)

data = {'status': status}

if in_reply_to_status_id:

data['in_reply_to_status_id'] = in_reply_to_status_id

if latitude is not None and longitude is not None:

data['lat'] = str(latitude)

data['long'] = str(longitude)

if place_id is not None:

data['place_id'] = str(place_id)

if display_coordinates:

data['display_coordinates'] = 'true'

if trim_user:

data['trim_user'] = 'true'

json = self._RequestUrl(url, 'POST', data=data)

data = self._ParseAndCheckTwitter(json.content)

return Status.NewFromJsonDict(data)

def PostMedia(self,

status,

media,

possibly_sensitive=None,

in_reply_to_status_id=None,

latitude=None,

longitude=None,

place_id=None,

display_coordinates=False):

"""Post a twitter status message from the user with a picture attached.

if not self.__auth:

raise TwitterError({'message': "The twitter.Api instance must be authenticated."})

url = '%s/statuses/update_with_media.json' % self.base_url

if isinstance(status, str) or self._input_encoding is None:

u_status = status

else:

u_status = str(status, self._input_encoding)

data = {'status': status}

if not hasattr(media, 'read'):

if media.startswith('http'):

data['media'] = urllib.request.urlopen(media).read()

else:

with open(str(media), 'rb') as f:

data['media'] = f.read()

else:

data['media'] = media.read()

if possibly_sensitive:

data['possibly_sensitive'] = 'true'

if in_reply_to_status_id:

data['in_reply_to_status_id'] = str(in_reply_to_status_id)

if latitude is not None and longitude is not None:

data['lat'] = str(latitude)

data['long'] = str(longitude)

if place_id is not None:

data['place_id'] = str(place_id)

if display_coordinates:

data['display_coordinates'] = 'true'

json = self._RequestUrl(url, 'POST', data=data)

data = self._ParseAndCheckTwitter(json.content)

return Status.NewFromJsonDict(data)

def PostMultipleMedia(self, status, media, possibly_sensitive=None,

in_reply_to_status_id=None, latitude=None,

longitude=None, place_id=None,

display_coordinates=False):

"""