github/plexpy
1
0
Fork 0
mirror of https://github.com/Tautulli/Tautulli.git synced 2025-07-05 20:51:15 -07:00
Code Issues Projects Releases Packages Wiki Activity Actions

Update python-twitter-3.5

Browse source
This commit is contained in:
JonnyWong16 2021-10-15 00:01:46 -07:00
parent cdeff390d9
commit 586f033347
No known key found for this signature in database
GPG key ID: B1F1F9807184697A
6 changed files with 134 additions and 49 deletions
Download patch file Download diff file Expand all files Collapse all files

9
lib/twitter/__init__.py
View file

@ -1,8 +1,7 @@
#!/usr/bin/env python
# -*- coding: utf-8 -*-
#
# vim: sw=2 ts=2 sts=2
#
# Copyright 2007 The Python-Twitter Developers
# Copyright 2007-2018 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.
@ -16,14 +15,14 @@
# 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"""
"""A library that provides a Python interface to the Twitter API."""
from __future__ import absolute_import
__author__ = 'The Python-Twitter Developers'
__email__ = 'python-twitter@googlegroups.com'
__copyright__ = 'Copyright (c) 2007-2016 The Python-Twitter Developers'
__license__ = 'Apache License 2.0'
__version__ = '3.4.1'
__version__ = '3.5'
__url__ = 'https://github.com/bear/python-twitter'
__download_url__ = 'https://pypi.python.org/pypi/python-twitter'
__description__ = 'A Python wrapper around the Twitter API'

78
lib/twitter/api.py
View file

@ -2,7 +2,7 @@
#
#
# Copyright 2007-2016 The Python-Twitter Developers
# Copyright 2007-2016, 2018 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.
@ -126,7 +126,7 @@ class Api(object):
>>> api.GetUserTimeline(user)
>>> api.GetHomeTimeline()
>>> api.GetStatus(status_id)
>>> def GetStatuses(status_ids)
>>> api.GetStatuses(status_ids)
>>> api.DestroyStatus(status_id)
>>> api.GetFriends(user)
>>> api.GetFollowers()
@ -292,6 +292,8 @@ class Api(object):
requests_log.setLevel(logging.DEBUG)
requests_log.propagate = True
self._session = requests.Session()
@staticmethod
def GetAppOnlyAuthToken(consumer_key, consumer_secret):
"""
@ -1158,9 +1160,13 @@ class Api(object):
else:
_, _, file_size, media_type = parse_media_file(media)
if file_size > self.chunk_size or media_type in chunked_types:
media_ids.append(self.UploadMediaChunked(media, media_additional_owners))
media_ids.append(self.UploadMediaChunked(
media, media_additional_owners, media_category=media_category
))
else:
media_ids.append(self.UploadMediaSimple(media, media_additional_owners))
media_ids.append(self.UploadMediaSimple(
media, media_additional_owners, media_category=media_category
))
parameters['media_ids'] = ','.join([str(mid) for mid in media_ids])
if latitude is not None and longitude is not None:
@ -1262,7 +1268,7 @@ class Api(object):
"""
url = '%s/media/upload.json' % self.upload_url
media_fp, filename, file_size, media_type = parse_media_file(media)
media_fp, filename, file_size, media_type = parse_media_file(media, async_upload=True)
if not all([media_fp, filename, file_size, media_type]):
raise TwitterError({'message': 'Could not process media file'})
@ -2819,8 +2825,6 @@ class Api(object):
if len(uids) > 100:
raise TwitterError("No more than 100 users may be requested per request.")
print(parameters)
resp = self._RequestUrl(url, 'GET', data=parameters)
data = self._ParseAndCheckTwitter(resp.content.decode('utf-8'))
@ -3003,30 +3007,48 @@ class Api(object):
Args:
text: The message text to be posted.
user_id:
The ID of the user who should receive the direct message. [Optional]
screen_name:
The screen name of the user who should receive the direct message. [Optional]
The ID of the user who should receive the direct message.
return_json (bool, optional):
If True JSON data will be returned, instead of twitter.User
If True JSON data will be returned, instead of twitter.DirectMessage
Returns:
A twitter.DirectMessage instance representing the message posted
"""
url = '%s/direct_messages/new.json' % self.base_url
data = {'text': text}
if user_id:
data['user_id'] = user_id
elif screen_name:
data['screen_name'] = screen_name
else:
raise TwitterError({'message': "Specify at least one of user_id or screen_name."})
url = '%s/direct_messages/events/new.json' % self.base_url
resp = self._RequestUrl(url, 'POST', data=data)
data = self._ParseAndCheckTwitter(resp.content.decode('utf-8'))
# Hack to allow some sort of backwards compatibility with older versions
# part of the fix for Issue #587
if user_id is None and screen_name is not None:
user_id = self.GetUser(screen_name=screen_name).id
event = {
'event': {
'type': 'message_create',
'message_create': {
'target': {
'recipient_id': user_id,
},
'message_data': {
'text': text
}
}
}
}
resp = self._RequestUrl(url, 'POST', json=event)
data = resp.json()
if return_json:
return data
else:
return DirectMessage.NewFromJsonDict(data)
dm = DirectMessage(
created_at=data['event']['created_timestamp'],
id=data['event']['id'],
recipient_id=data['event']['message_create']['target']['recipient_id'],
sender_id=data['event']['message_create']['sender_id'],
text=data['event']['message_create']['message_data']['text'],
)
dm._json = data
return dm
def DestroyDirectMessage(self, message_id, include_entities=True, return_json=False):
"""Destroys the direct message specified in the required ID parameter.
@ -4882,7 +4904,7 @@ class Api(object):
raise TwitterError({'message': "Exceeded connection limit for user"})
if "Error 401 Unauthorized" in json_data:
raise TwitterError({'message': "Unauthorized"})
raise TwitterError({'Unknown error: {0}'.format(json_data)})
raise TwitterError({'Unknown error': '{0}'.format(json_data)})
self._CheckForTwitterError(data)
return data
@ -4954,20 +4976,20 @@ class Api(object):
if data:
if 'media_ids' in data:
url = self._BuildUrl(url, extra_params={'media_ids': data['media_ids']})
resp = requests.post(url, data=data, auth=self.__auth, timeout=self._timeout, proxies=self.proxies)
resp = self._session.post(url, data=data, auth=self.__auth, timeout=self._timeout, proxies=self.proxies)
elif 'media' in data:
resp = requests.post(url, files=data, auth=self.__auth, timeout=self._timeout, proxies=self.proxies)
resp = self._session.post(url, files=data, auth=self.__auth, timeout=self._timeout, proxies=self.proxies)
else:
resp = requests.post(url, data=data, auth=self.__auth, timeout=self._timeout, proxies=self.proxies)
resp = self._session.post(url, data=data, auth=self.__auth, timeout=self._timeout, proxies=self.proxies)
elif json:
resp = requests.post(url, json=json, auth=self.__auth, timeout=self._timeout, proxies=self.proxies)
resp = self._session.post(url, json=json, auth=self.__auth, timeout=self._timeout, proxies=self.proxies)
else:
resp = 0 # POST request, but without data or json
elif verb == 'GET':
data['tweet_mode'] = self.tweet_mode
url = self._BuildUrl(url, extra_params=data)
resp = requests.get(url, auth=self.__auth, timeout=self._timeout, proxies=self.proxies)
resp = self._session.get(url, auth=self.__auth, timeout=self._timeout, proxies=self.proxies)
else:
resp = 0 # if not a POST or GET request

65
lib/twitter/debug.py Normal file
View file

@ -0,0 +1,65 @@
from twitter import Api, TwitterError
import requests
class Api(Api):
def DebugEndpoint(self, verb=None, endpoint=None, data=None):
""" Request a url and return raw data. For testing purposes only.
Args:
url:
The web location we want to retrieve.
verb:
Either POST or GET.
data:
A dict of (str, unicode) key/value pairs.
Returns:
data
"""
url = "{0}{1}".format(self.base_url, endpoint)
if verb == 'POST':
if 'media_ids' in data:
url = self._BuildUrl(
url,
extra_params={
'media_ids': data['media_ids']
}
)
print('POSTing url:', url)
if 'media' in data:
try:
print('POSTing url:', url)
raw_data = requests.post(
url,
files=data,
auth=self.__auth,
timeout=self._timeout
)
except requests.RequestException as e:
raise TwitterError(str(e))
else:
try:
print('POSTing url:', url)
raw_data = requests.post(
url,
data=data,
auth=self.__auth,
timeout=self._timeout
)
except requests.RequestException as e:
raise TwitterError(str(e))
if verb == 'GET':
url = self._BuildUrl(url, extra_params=data)
print('GETting url:', url)
try:
raw_data = requests.get(
url,
auth=self.__auth,
timeout=self._timeout)
except requests.RequestException as e:
raise TwitterError(str(e))
return raw_data._content

14
lib/twitter/models.py
View file

@ -35,10 +35,10 @@ class TwitterModel(object):
raise TypeError('unhashable type: {} (no id attribute)'
.format(type(self)))
def AsJsonString(self):
def AsJsonString(self, ensure_ascii=True):
""" Returns the TwitterModel as a JSON string based on key/value
pairs returned from the AsDict() method. """
return json.dumps(self.AsDict(), sort_keys=True)
return json.dumps(self.AsDict(), ensure_ascii=ensure_ascii, sort_keys=True)
def AsDict(self):
""" Create a dictionary representation of the object. Please see inline
@ -185,21 +185,13 @@ class DirectMessage(TwitterModel):
self.param_defaults = {
'created_at': None,
'id': None,
'recipient': None,
'recipient_id': None,
'recipient_screen_name': None,
'sender': None,
'sender_id': None,
'sender_screen_name': None,
'text': None,
}
for (param, default) in self.param_defaults.items():
setattr(self, param, kwargs.get(param, default))
if 'sender' in kwargs:
self.sender = User.NewFromJsonDict(kwargs.get('sender', None))
if 'recipient' in kwargs:
self.recipient = User.NewFromJsonDict(kwargs.get('recipient', None))
def __repr__(self):
if self.text and len(self.text) > 140:
@ -208,7 +200,7 @@ class DirectMessage(TwitterModel):
text = self.text
return "DirectMessage(ID={dm_id}, Sender={sender}, Created={time}, Text='{text!r}')".format(
dm_id=self.id,
sender=self.sender_screen_name,
sender=self.sender_id,
time=self.created_at,
text=text)

2
lib/twitter/parse_tweet.py
View file

@ -96,5 +96,5 @@ class ParseTweet(object):
@staticmethod
def getURLs(tweet):
""" URL : [http://]?[\w\.?/]+"""
r""" URL : [http://]?[\w\.?/]+"""
return re.findall(ParseTweet.regexp["URL"], tweet)

15
lib/twitter/twitter_utils.py
View file

@ -216,12 +216,13 @@ def http_to_file(http):
return data_file
def parse_media_file(passed_media):
def parse_media_file(passed_media, async_upload=False):
""" Parses a media file and attempts to return a file-like object and
information about the media file.
Args:
passed_media: media file which to parse.
async_upload: flag, for validation media file attributes.
Returns:
file-like object, the filename of the media file, the file size, and
@ -229,9 +230,11 @@ def parse_media_file(passed_media):
"""
img_formats = ['image/jpeg',
'image/png',
'image/gif',
'image/bmp',
'image/webp']
long_img_formats = [
'image/gif'
]
video_formats = ['video/mp4',
'video/quicktime']
@ -266,9 +269,13 @@ def parse_media_file(passed_media):
if media_type is not None:
if media_type in img_formats and file_size > 5 * 1048576:
raise TwitterError({'message': 'Images must be less than 5MB.'})
elif media_type in video_formats and file_size > 15 * 1048576:
elif media_type in long_img_formats and file_size > 15 * 1048576:
raise TwitterError({'message': 'GIF Image must be less than 15MB.'})
elif media_type in video_formats and not async_upload and file_size > 15 * 1048576:
raise TwitterError({'message': 'Videos must be less than 15MB.'})
elif media_type not in img_formats and media_type not in video_formats:
elif media_type in video_formats and async_upload and file_size > 512 * 1048576:
raise TwitterError({'message': 'Videos must be less than 512MB.'})
elif media_type not in img_formats and media_type not in video_formats and media_type not in long_img_formats:
raise TwitterError({'message': 'Media type could not be determined.'})
return data_file, filename, file_size, media_type