# Tweepy # Copyright 2009-2010 Joshua Roesslein # See LICENSE for details. from tweepy.error import TweepError from tweepy.utils import parse_datetime, parse_html_value, parse_a_href, \ parse_search_datetime, unescape_html class ResultSet(list): """A list like object that holds results from a Twitter API query.""" def __init__(self, max_id=None, since_id=None): super(ResultSet, self).__init__() self._max_id = max_id self._since_id = since_id @property def max_id(self): if self._max_id: return self._max_id ids = self.ids() return max(ids) if ids else None @property def since_id(self): if self._since_id: return self._since_id ids = self.ids() return min(ids) if ids else None def ids(self): return [item.id for item in self if hasattr(item, 'id')] class Model(object): def __init__(self, api=None): self._api = api def __getstate__(self): # pickle pickle = dict(self.__dict__) try: del pickle['_api'] # do not pickle the API reference except KeyError: pass return pickle @classmethod def parse(cls, api, json): """Parse a JSON object into a model instance.""" raise NotImplementedError @classmethod def parse_list(cls, api, json_list): """Parse a list of JSON objects into a result set of model instances.""" results = ResultSet() for obj in json_list: if obj: results.append(cls.parse(api, obj)) return results class Status(Model): @classmethod def parse(cls, api, json): status = cls(api) for k, v in json.items(): if k == 'user': user_model = getattr(api.parser.model_factory, 'user') user = user_model.parse(api, v) setattr(status, 'author', user) setattr(status, 'user', user) # DEPRECIATED elif k == 'created_at': setattr(status, k, parse_datetime(v)) elif k == 'source': if '<' in v: setattr(status, k, parse_html_value(v)) setattr(status, 'source_url', parse_a_href(v)) else: setattr(status, k, v) setattr(status, 'source_url', None) elif k == 'retweeted_status': setattr(status, k, Status.parse(api, v)) elif k == 'place': if v is not None: setattr(status, k, Place.parse(api, v)) else: setattr(status, k, None) else: setattr(status, k, v) return status def destroy(self): return self._api.destroy_status(self.id) def retweet(self): return self._api.retweet(self.id) def retweets(self): return self._api.retweets(self.id) def favorite(self): return self._api.create_favorite(self.id) class User(Model): @classmethod def parse(cls, api, json): user = cls(api) for k, v in json.items(): if k == 'created_at': setattr(user, k, parse_datetime(v)) elif k == 'status': setattr(user, k, Status.parse(api, v)) elif k == 'following': # twitter sets this to null if it is false if v is True: setattr(user, k, True) else: setattr(user, k, False) else: setattr(user, k, v) return user @classmethod def parse_list(cls, api, json_list): if isinstance(json_list, list): item_list = json_list else: item_list = json_list['users'] results = ResultSet() for obj in item_list: results.append(cls.parse(api, obj)) return results def timeline(self, **kargs): return self._api.user_timeline(user_id=self.id, **kargs) def friends(self, **kargs): return self._api.friends(user_id=self.id, **kargs) def followers(self, **kargs): return self._api.followers(user_id=self.id, **kargs) def follow(self): self._api.create_friendship(user_id=self.id) self.following = True def unfollow(self): self._api.destroy_friendship(user_id=self.id) self.following = False def lists_memberships(self, *args, **kargs): return self._api.lists_memberships(user=self.screen_name, *args, **kargs) def lists_subscriptions(self, *args, **kargs): return self._api.lists_subscriptions(user=self.screen_name, *args, **kargs) def lists(self, *args, **kargs): return self._api.lists(user=self.screen_name, *args, **kargs) def followers_ids(self, *args, **kargs): return self._api.followers_ids(user_id=self.id, *args, **kargs) class DirectMessage(Model): @classmethod def parse(cls, api, json): dm = cls(api) for k, v in json.items(): if k == 'sender' or k == 'recipient': setattr(dm, k, User.parse(api, v)) elif k == 'created_at': setattr(dm, k, parse_datetime(v)) else: setattr(dm, k, v) return dm def destroy(self): return self._api.destroy_direct_message(self.id) class Friendship(Model): @classmethod def parse(cls, api, json): relationship = json['relationship'] # parse source source = cls(api) for k, v in relationship['source'].items(): setattr(source, k, v) # parse target target = cls(api) for k, v in relationship['target'].items(): setattr(target, k, v) return source, target class Category(Model): @classmethod def parse(cls, api, json): category = cls(api) for k, v in json.items(): setattr(category, k, v) return category class SavedSearch(Model): @classmethod def parse(cls, api, json): ss = cls(api) for k, v in json.items(): if k == 'created_at': setattr(ss, k, parse_datetime(v)) else: setattr(ss, k, v) return ss def destroy(self): return self._api.destroy_saved_search(self.id) class SearchResults(ResultSet): @classmethod def parse(cls, api, json): metadata = json['search_metadata'] results = SearchResults(metadata.get('max_id'), metadata.get('since_id')) results.refresh_url = metadata.get('refresh_url') results.completed_in = metadata.get('completed_in') results.query = metadata.get('query') for status in json['statuses']: results.append(Status.parse(api, status)) return results class List(Model): @classmethod def parse(cls, api, json): lst = List(api) for k,v in json.items(): if k == 'user': setattr(lst, k, User.parse(api, v)) elif k == 'created_at': setattr(lst, k, parse_datetime(v)) else: setattr(lst, k, v) return lst @classmethod def parse_list(cls, api, json_list, result_set=None): results = ResultSet() if isinstance(json_list, dict): json_list = json_list['lists'] for obj in json_list: results.append(cls.parse(api, obj)) return results def update(self, **kargs): return self._api.update_list(self.slug, **kargs) def destroy(self): return self._api.destroy_list(self.slug) def timeline(self, **kargs): return self._api.list_timeline(self.user.screen_name, self.slug, **kargs) def add_member(self, id): return self._api.add_list_member(self.slug, id) def remove_member(self, id): return self._api.remove_list_member(self.slug, id) def members(self, **kargs): return self._api.list_members(self.user.screen_name, self.slug, **kargs) def is_member(self, id): return self._api.is_list_member(self.user.screen_name, self.slug, id) def subscribe(self): return self._api.subscribe_list(self.user.screen_name, self.slug) def unsubscribe(self): return self._api.unsubscribe_list(self.user.screen_name, self.slug) def subscribers(self, **kargs): return self._api.list_subscribers(self.user.screen_name, self.slug, **kargs) def is_subscribed(self, id): return self._api.is_subscribed_list(self.user.screen_name, self.slug, id) class Relation(Model): @classmethod def parse(cls, api, json): result = cls(api) for k,v in json.items(): if k == 'value' and json['kind'] in ['Tweet', 'LookedupStatus']: setattr(result, k, Status.parse(api, v)) elif k == 'results': setattr(result, k, Relation.parse_list(api, v)) else: setattr(result, k, v) return result class Relationship(Model): @classmethod def parse(cls, api, json): result = cls(api) for k,v in json.items(): if k == 'connections': setattr(result, 'is_following', 'following' in v) setattr(result, 'is_followed_by', 'followed_by' in v) else: setattr(result, k, v) return result class JSONModel(Model): @classmethod def parse(cls, api, json): return json class IDModel(Model): @classmethod def parse(cls, api, json): if isinstance(json, list): return json else: return json['ids'] class BoundingBox(Model): @classmethod def parse(cls, api, json): result = cls(api) if json is not None: for k, v in json.items(): setattr(result, k, v) return result def origin(self): """ Return longitude, latitude of southwest (bottom, left) corner of bounding box, as a tuple. This assumes that bounding box is always a rectangle, which appears to be the case at present. """ return tuple(self.coordinates[0][0]) def corner(self): """ Return longitude, latitude of northeast (top, right) corner of bounding box, as a tuple. This assumes that bounding box is always a rectangle, which appears to be the case at present. """ return tuple(self.coordinates[0][2]) class Place(Model): @classmethod def parse(cls, api, json): place = cls(api) for k, v in json.items(): if k == 'bounding_box': # bounding_box value may be null (None.) # Example: "United States" (id=96683cc9126741d1) if v is not None: t = BoundingBox.parse(api, v) else: t = v setattr(place, k, t) elif k == 'contained_within': # contained_within is a list of Places. setattr(place, k, Place.parse_list(api, v)) else: setattr(place, k, v) return place @classmethod def parse_list(cls, api, json_list): if isinstance(json_list, list): item_list = json_list else: item_list = json_list['result']['places'] results = ResultSet() for obj in item_list: results.append(cls.parse(api, obj)) return results class ModelFactory(object): """ Used by parsers for creating instances of models. You may subclass this factory to add your own extended models. """ status = Status user = User direct_message = DirectMessage friendship = Friendship saved_search = SavedSearch search_results = SearchResults category = Category list = List relation = Relation relationship = Relationship json = JSONModel ids = IDModel place = Place bounding_box = BoundingBox