mcvsoftware/twblue
1
0
Fork 0
mirror of https://github.com/MCV-Software/TWBlue.git synced 2024-11-22 19:28:09 -06:00
Code Issues Actions Packages Projects Releases Wiki Activity

Updated twython

Browse Source
This commit is contained in:
Manuel Cortez 2016-03-29 16:09:09 -06:00
parent edd45a1adf
commit ed95270d3b
3 changed files with 97 additions and 19 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
src/twython/__init__.py
View File

@ -19,7 +19,7 @@ Questions, comments? ryan@venodesigns.net
"""
__author__ = 'Ryan McGrath <ryan@venodesigns.net>'
__version__ = '3.2.0'
__version__ = '3.3.0'
from .api import Twython
from .streaming import TwythonStreamer

44
src/twython/api.py
View File

@ -10,6 +10,7 @@ dealing with the Twitter API
"""
import warnings
import re
import requests
from requests.auth import HTTPBasicAuth
@ -102,15 +103,10 @@ class Twython(EndpointsMixin, object):
auth = None
if oauth_version == 1:
# User Authentication is through OAuth 1
if self.app_key is not None and self.app_secret is not None and \
self.oauth_token is None and self.oauth_token_secret is None:
auth = OAuth1(self.app_key, self.app_secret)
if self.app_key is not None and self.app_secret is not None and \
self.oauth_token is not None and self.oauth_token_secret is \
not None:
if self.app_key is not None and self.app_secret is not None:
auth = OAuth1(self.app_key, self.app_secret,
self.oauth_token, self.oauth_token_secret)
self.oauth_token, self.oauth_token_secret)
elif oauth_version == 2 and self.access_token:
# Application Authentication is through OAuth 2
token = {'token_type': token_type,
@ -198,7 +194,10 @@ class Twython(EndpointsMixin, object):
retry_after=response.headers.get('X-Rate-Limit-Reset'))
try:
content = response.json()
if response.status_code == 204:
content = response.content
else:
content = response.json()
except ValueError:
raise TwythonError('Response was not valid JSON. \
Unable to decode.')
@ -528,7 +527,7 @@ class Twython(EndpointsMixin, object):
return str(text)
@staticmethod
def html_for_tweet(tweet, use_display_url=True, use_expanded_url=False):
def html_for_tweet(tweet, use_display_url=True, use_expanded_url=False, expand_quoted_status=False):
"""Return HTML for a tweet (urls, mentions, hashtags replaced with links)
:param tweet: Tweet object from received from Twitter API
@ -550,19 +549,22 @@ class Twython(EndpointsMixin, object):
entities = tweet['entities']
# Mentions
for entity in entities['user_mentions']:
for entity in sorted(entities['user_mentions'],
key=lambda mention: len(mention['screen_name']), reverse=True):
start, end = entity['indices'][0], entity['indices'][1]
mention_html = '<a href="https://twitter.com/%(screen_name)s" class="twython-mention">@%(screen_name)s</a>'
text = text.replace(tweet['text'][start:end],
mention_html % {'screen_name': entity['screen_name']})
text = re.sub(r'(?<!>)' + tweet['text'][start:end] + '(?!</a>)',
mention_html % {'screen_name': entity['screen_name']}, text)
# Hashtags
for entity in entities['hashtags']:
for entity in sorted(entities['hashtags'],
key=lambda hashtag: len(hashtag['text']), reverse=True):
start, end = entity['indices'][0], entity['indices'][1]
hashtag_html = '<a href="https://twitter.com/search?q=%%23%(hashtag)s" class="twython-hashtag">#%(hashtag)s</a>'
text = text.replace(tweet['text'][start:end], hashtag_html % {'hashtag': entity['text']})
text = re.sub(r'(?<!>)' + tweet['text'][start:end] + '(?!</a>)',
hashtag_html % {'hashtag': entity['text']}, text)
# Urls
for entity in entities['urls']:
@ -595,4 +597,16 @@ class Twython(EndpointsMixin, object):
text = text.replace(tweet['text'][start:end],
url_html % (entity['url'], shown_url))
if expand_quoted_status and tweet.get('is_quote_status'):
quoted_status = tweet['quoted_status']
text += '<blockquote class="twython-quote">%(quote)s<cite><a href="%(quote_tweet_link)s">' \
'<span class="twython-quote-user-name">%(quote_user_name)s</span>' \
'<span class="twython-quote-user-screenname">@%(quote_user_screen_name)s</span></a>' \
'</cite></blockquote>' % \
{'quote': Twython.html_for_tweet(quoted_status, use_display_url, use_expanded_url, False),
'quote_tweet_link': 'https://twitter.com/%s/status/%s' %
(quoted_status['user']['screen_name'], quoted_status['id_str']),
'quote_user_name': quoted_status['user']['name'],
'quote_user_screen_name': quoted_status['user']['screen_name']}
return text

70
src/twython/endpoints.py
View File

@ -14,7 +14,12 @@ This map is organized the order functions are documented at:
https://dev.twitter.com/docs/api/1.1
"""
import os
import warnings
try:
from StringIO import StringIO
except ImportError:
from io import StringIO
from .advisory import TwythonDeprecationWarning
@ -139,6 +144,65 @@ class EndpointsMixin(object):
"""
return self.post('https://upload.twitter.com/1.1/media/upload.json', params=params)
def set_description(self, **params):
return self.post('media/metadata/create', params=params)
def upload_video(self, media, media_type, size=None):
"""Uploads video file to Twitter servers in chunks. The file will be available to be attached
to a status for 60 minutes. To attach to a update, pass a list of returned media ids
to the 'update_status' method using the 'media_ids' param.
Upload happens in 3 stages:
- INIT call with size of media to be uploaded(in bytes). If this is more than 15mb, twitter will return error.
- APPEND calls each with media chunk. This returns a 204(No Content) if chunk is received.
- FINALIZE call to complete media upload. This returns media_id to be used with status update.
Twitter media upload api expects each chunk to be not more than 5mb. We are sending chunk of 1mb each.
Docs:
https://dev.twitter.com/rest/public/uploading-media#chunkedupload
"""
upload_url = 'https://upload.twitter.com/1.1/media/upload.json'
if not size:
media.seek(0, os.SEEK_END)
size = media.tell()
media.seek(0)
# Stage 1: INIT call
params = {
'command': 'INIT',
'media_type': media_type,
'total_bytes': size
}
response_init = self.post(upload_url, params=params)
media_id = response_init['media_id']
# Stage 2: APPEND calls with 1mb chunks
segment_index = 0
while True:
data = media.read(1*1024*1024)
if not data:
break
media_chunk = StringIO()
media_chunk.write(data)
media_chunk.seek(0)
params = {
'command': 'APPEND',
'media_id': media_id,
'segment_index': segment_index,
'media': media_chunk,
}
self.post(upload_url, params=params)
segment_index += 1
# Stage 3: FINALIZE call to complete upload
params = {
'command': 'FINALIZE',
'media_id': media_id
}
return self.post(upload_url, params=params)
def get_oembed_tweet(self, **params):
"""Returns information allowing the creation of an embedded
representation of a Tweet on third party sites.
@ -458,7 +522,7 @@ class EndpointsMixin(object):
Docs: https://dev.twitter.com/docs/api/1.1/get/users/lookup
"""
return self.post('users/lookup', params=params)
return self.get('users/lookup', params=params)
def show_user(self, **params):
"""Returns a variety of information about the user specified by the
@ -546,7 +610,7 @@ class EndpointsMixin(object):
list_mute_ids.iter_key = 'ids'
def create_mute(self, **params):
"""Mutes the specified user, preventing their tweets appearing
"""Mutes the specified user, preventing their tweets appearing
in the authenticating user's timeline.
Docs: https://dev.twitter.com/docs/api/1.1/post/mutes/users/create
@ -555,7 +619,7 @@ class EndpointsMixin(object):
return self.post('mutes/users/create', params=params)
def destroy_mute(self, **params):
"""Un-mutes the user specified in the user or ID parameter for
"""Un-mutes the user specified in the user or ID parameter for
the authenticating user.
Docs: https://dev.twitter.com/docs/api/1.1/post/mutes/users/destroy