# -*- coding: utf-8 -*-
from __future__ import unicode_literals
import json
from calendar import timegm
try:
from rfc822 import parsedate
except ImportError:
from email.utils import parsedate
[docs]class List(TwitterModel):
"""A class representing the List structure used by the twitter API. """
def __init__(self, **kwargs):
self.param_defaults = {
'description': None,
'following': None,
'full_name': None,
'id': None,
'member_count': None,
'mode': None,
'name': None,
'slug': None,
'subscriber_count': None,
'uri': None,
'user': None,
}
for (param, default) in self.param_defaults.items():
setattr(self, param, kwargs.get(param, default))
if 'user' in kwargs:
self.user = User.NewFromJsonDict(kwargs.get('user'))
def __repr__(self):
return "List(ID={list_id}, FullName={full_name!r}, Slug={slug}, User={user})".format(
list_id=self.id,
full_name=self.full_name,
slug=self.slug,
user=self.user.screen_name)
[docs]class Category(TwitterModel):
"""A class representing the suggested user category structure. """
def __init__(self, **kwargs):
self.param_defaults = {
'name': None,
'size': None,
'slug': None,
}
for (param, default) in self.param_defaults.items():
setattr(self, param, kwargs.get(param, default))
def __repr__(self):
return "Category(Name={name!r}, Slug={slug}, Size={size})".format(
name=self.name,
slug=self.slug,
size=self.size)
[docs]class DirectMessage(TwitterModel):
"""A class representing a Direct Message. """
def __init__(self, **kwargs):
self.param_defaults = {
'created_at': None,
'id': None,
'recipient_id': None,
'sender_id': None,
'text': None,
}
for (param, default) in self.param_defaults.items():
setattr(self, param, kwargs.get(param, default))
def __repr__(self):
if self.text and len(self.text) > 140:
text = "{text}[...]".format(text=self.text[:140])
else:
text = self.text
return "DirectMessage(ID={dm_id}, Sender={sender}, Created={time}, Text='{text!r}')".format(
dm_id=self.id,
sender=self.sender_id,
time=self.created_at,
text=text)
[docs]class Trend(TwitterModel):
""" A class representing a trending topic. """
def __init__(self, **kwargs):
self.param_defaults = {
'events': None,
'name': None,
'promoted_content': None,
'query': None,
'timestamp': None,
'url': None,
'tweet_volume': None,
}
for (param, default) in self.param_defaults.items():
setattr(self, param, kwargs.get(param, default))
def __repr__(self):
return "Trend(Name={0!r}, Time={1}, URL={2})".format(
self.name,
self.timestamp,
self.url)
@property
def volume(self):
return self.tweet_volume
[docs]class Hashtag(TwitterModel):
""" A class representing a twitter hashtag. """
def __init__(self, **kwargs):
self.param_defaults = {
'text': None
}
for (param, default) in self.param_defaults.items():
setattr(self, param, kwargs.get(param, default))
def __repr__(self):
return "Hashtag(Text={text!r})".format(
text=self.text)
[docs]class Url(TwitterModel):
""" A class representing an URL contained in a tweet. """
def __init__(self, **kwargs):
self.param_defaults = {
'expanded_url': None,
'url': None}
for (param, default) in self.param_defaults.items():
setattr(self, param, kwargs.get(param, default))
def __repr__(self):
return "URL(URL={url}, ExpandedURL={eurl})".format(
url=self.url,
eurl=self.expanded_url)
[docs]class UserStatus(TwitterModel):
""" A class representing the UserStatus structure. This is an abbreviated
form of the twitter.User object. """
_connections = {'following': False,
'followed_by': False,
'following_received': False,
'following_requested': False,
'blocking': False,
'muting': False}
def __init__(self, **kwargs):
self.param_defaults = {
'blocking': False,
'followed_by': False,
'following': False,
'following_received': False,
'following_requested': False,
'id': None,
'id_str': None,
'muting': False,
'name': None,
'screen_name': None,
}
for (param, default) in self.param_defaults.items():
setattr(self, param, kwargs.get(param, default))
if 'connections' in kwargs:
for param in self._connections:
if param in kwargs['connections']:
setattr(self, param, True)
@property
def connections(self):
return {'following': self.following,
'followed_by': self.followed_by,
'following_received': self.following_received,
'following_requested': self.following_requested,
'blocking': self.blocking,
'muting': self.muting}
def __repr__(self):
connections = [param for param in self.connections if getattr(self, param)]
return "UserStatus(ID={uid}, ScreenName={sn}, Connections=[{conn}])".format(
uid=self.id,
sn=self.screen_name,
conn=", ".join(connections))
[docs]class User(TwitterModel):
"""A class representing the User structure. """
def __init__(self, **kwargs):
self.param_defaults = {
'contributors_enabled': None,
'created_at': None,
'default_profile': None,
'default_profile_image': None,
'description': None,
'email': None,
'favourites_count': None,
'followers_count': None,
'following': None,
'friends_count': None,
'geo_enabled': None,
'id': None,
'id_str': None,
'lang': None,
'listed_count': None,
'location': None,
'name': None,
'notifications': None,
'profile_background_color': None,
'profile_background_image_url': None,
'profile_background_image_url_https': None,
'profile_background_tile': None,
'profile_banner_url': None,
'profile_image_url': None,
'profile_image_url_https': None,
'profile_link_color': None,
'profile_sidebar_border_color': None,
'profile_sidebar_fill_color': None,
'profile_text_color': None,
'profile_use_background_image': None,
'protected': None,
'screen_name': None,
'status': None,
'statuses_count': None,
'time_zone': None,
'url': None,
'utc_offset': None,
'verified': None,
'withheld_in_countries': None,
'withheld_scope': None,
}
for (param, default) in self.param_defaults.items():
setattr(self, param, kwargs.get(param, default))
def __repr__(self):
return "User(ID={uid}, ScreenName={sn})".format(
uid=self.id,
sn=self.screen_name)
[docs] @classmethod
def NewFromJsonDict(cls, data, **kwargs):
from twitter import Status
if data.get('status', None):
status = Status.NewFromJsonDict(data.get('status'))
return super(cls, cls).NewFromJsonDict(data=data, status=status)
else:
return super(cls, cls).NewFromJsonDict(data=data)
[docs]class Status(TwitterModel):
"""A class representing the Status structure used by the twitter API.
"""
def __init__(self, **kwargs):
self.param_defaults = {
'contributors': None,
'coordinates': None,
'created_at': None,
'current_user_retweet': None,
'favorite_count': None,
'favorited': None,
'full_text': None,
'geo': None,
'hashtags': None,
'id': None,
'id_str': None,
'in_reply_to_screen_name': None,
'in_reply_to_status_id': None,
'in_reply_to_user_id': None,
'lang': None,
'location': None,
'media': None,
'place': None,
'possibly_sensitive': None,
'quoted_status': None,
'quoted_status_id': None,
'quoted_status_id_str': None,
'retweet_count': None,
'retweeted': None,
'retweeted_status': None,
'scopes': None,
'source': None,
'text': None,
'truncated': None,
'urls': None,
'user': None,
'user_mentions': None,
'withheld_copyright': None,
'withheld_in_countries': None,
'withheld_scope': None,
}
for (param, default) in self.param_defaults.items():
setattr(self, param, kwargs.get(param, default))
if kwargs.get('full_text', None):
self.tweet_mode = 'extended'
else:
self.tweet_mode = 'compatibility'
@property
def created_at_in_seconds(self):
""" Get the time this status message was posted, in seconds since
the epoch (1 Jan 1970).
Returns:
int: The time this status message was posted, in seconds since
the epoch.
"""
return timegm(parsedate(self.created_at))
def __repr__(self):
""" A string representation of this twitter.Status instance.
The return value is the ID of status, username and datetime.
Returns:
string: A string representation of this twitter.Status instance with
the ID of status, username and datetime.
"""
if self.tweet_mode == 'extended':
text = self.full_text
else:
text = self.text
if self.user:
return "Status(ID={0}, ScreenName={1}, Created={2}, Text={3!r})".format(
self.id,
self.user.screen_name,
self.created_at,
text)
else:
return u"Status(ID={0}, Created={1}, Text={2!r})".format(
self.id,
self.created_at,
text)
[docs] @classmethod
def NewFromJsonDict(cls, data, **kwargs):
""" Create a new instance based on a JSON dict.
Args:
data: A JSON dict, as converted from the JSON in the twitter API
Returns:
A twitter.Status instance
"""
current_user_retweet = None
hashtags = None
media = None
quoted_status = None
retweeted_status = None
urls = None
user = None
user_mentions = None
# for loading extended tweets from the streaming API.
if 'extended_tweet' in data:
for k, v in data['extended_tweet'].items():
data[k] = v
if 'user' in data:
user = User.NewFromJsonDict(data['user'])
if 'retweeted_status' in data:
retweeted_status = Status.NewFromJsonDict(data['retweeted_status'])
if 'current_user_retweet' in data:
current_user_retweet = data['current_user_retweet']['id']
if 'quoted_status' in data:
quoted_status = Status.NewFromJsonDict(data.get('quoted_status'))
if 'entities' in data:
if 'urls' in data['entities']:
urls = [Url.NewFromJsonDict(u) for u in data['entities']['urls']]
if 'user_mentions' in data['entities']:
user_mentions = [User.NewFromJsonDict(u) for u in data['entities']['user_mentions']]
if 'hashtags' in data['entities']:
hashtags = [Hashtag.NewFromJsonDict(h) for h in data['entities']['hashtags']]
if 'media' in data['entities']:
media = [Media.NewFromJsonDict(m) for m in data['entities']['media']]
# the new extended entities
if 'extended_entities' in data:
if 'media' in data['extended_entities']:
media = [Media.NewFromJsonDict(m) for m in data['extended_entities']['media']]
return super(cls, cls).NewFromJsonDict(data=data,
current_user_retweet=current_user_retweet,
hashtags=hashtags,
media=media,
quoted_status=quoted_status,
retweeted_status=retweeted_status,
urls=urls,
user=user,
user_mentions=user_mentions)