diaspy/streams.py

   1 """Docstrings for this module are taken from:
   2 https://gist.github.com/MrZYX/01c93096c30dc44caf71
   3
   4 Documentation for D* JSON API taken from:
   5 http://pad.spored.de/ro/r.qWmvhSZg7rk4OQam
   6 """
   7
   8
   9 import json
  10 import time
  11 from diaspy.models import Post, Aspect
  12 from diaspy import errors
  13
  14
  15 class Generic():
  16     """Object representing generic stream.
  17     """
  18     _location = 'stream.json'
  19
  20     def __init__(self, connection, location='', fetch=True):
  21         """
  22         :param connection: Connection() object
  23         :type connection: diaspy.connection.Connection
  24         :param location: location of json (optional)
  25         :type location: str
  26         :param fetch: will call .fill() if true
  27         :type fetch: bool
  28         """
  29         self._connection = connection
  30         if location: self._location = location
  31         self._stream = []
  32         #   since epoch
  33         self.max_time = int(time.mktime(time.gmtime()))
  34         if fetch: self.fill()
  35
  36     def __contains__(self, post):
  37         """Returns True if stream contains given post.
  38         """
  39         return post in self._stream
  40
  41     def __iter__(self):
  42         """Provides iterable interface for stream.
  43         """
  44         return iter(self._stream)
  45
  46     def __getitem__(self, n):
  47         """Returns n-th item in Stream.
  48         """
  49         return self._stream[n]
  50
  51     def __len__(self):
  52         """Returns length of the Stream.
  53         """
  54         return len(self._stream)
  55
  56     def _obtain(self, max_time=0, suppress=True):
  57         """Obtains stream from pod.
  58
  59         suppress:bool - suppress post-fetching errors (e.g. 404)
  60         """
  61         params = {}
  62         if max_time:
  63             params['max_time'] = max_time
  64             params['_'] = int(time.time() * 1000)
  65         request = self._connection.get(self._location, params=params)
  66         if request.status_code != 200:
  67             raise errors.StreamError('wrong status code: {0}'.format(request.status_code))
  68         posts = []
  69         for post in request.json():
  70             try:
  71                 posts.append(Post(self._connection, guid=post['guid']))
  72             except errors.PostError:
  73                 if not suppress:
  74                     raise
  75         return posts
  76
  77     def _expand(self, new_stream):
  78         """Appends older posts to stream.
  79         """
  80         ids = [post.id for post in self._stream]
  81         stream = self._stream
  82         for post in new_stream:
  83             if post.id not in ids:
  84                 stream.append(post)
  85                 ids.append(post.id)
  86         self._stream = stream
  87
  88     def _update(self, new_stream):
  89         """Updates stream with new posts.
  90         """
  91         ids = [post.id for post in self._stream]
  92
  93         stream = self._stream
  94         for i in range(len(new_stream)):
  95             if new_stream[-i].id not in ids:
  96                 stream = [new_stream[-i]] + stream
  97                 ids.append(new_stream[-i].id)
  98         self._stream = stream
  99
 100     def clear(self):
 101         """Set stream to empty.
 102         """
 103         self._stream = []
 104
 105     def purge(self):
 106         """Removes all unexistent posts from stream.
 107         """
 108         stream = []
 109         for post in self._stream:
 110             deleted = False
 111             try:
 112                 # error will tell us that the post has been deleted
 113                 post.update()
 114             except Exception:
 115                 deleted = True
 116             finally:
 117                 if not deleted: stream.append(post)
 118         self._stream = stream
 119
 120     def update(self):
 121         """Updates stream with new posts.
 122         """
 123         self._update(self._obtain())
 124
 125     def fill(self):
 126         """Fills the stream with posts.
 127
 128         **Notice:** this will create entirely new list of posts.
 129         If you want to preseve posts already present in stream use update().
 130         """
 131         self._stream = self._obtain()
 132
 133     def more(self, max_time=0, backtime=86400):
 134         """Tries to download more (older posts) posts from Stream.
 135
 136         :param backtime: how many seconds substract each time (defaults to one day)
 137         :type backtime: int
 138         :param max_time: seconds since epoch (optional, diaspy'll figure everything on its own)
 139         :type max_time: int
 140         """
 141         if not max_time: max_time = self.max_time - backtime
 142         self.max_time = max_time
 143         new_stream = self._obtain(max_time=max_time)
 144         self._expand(new_stream)
 145
 146     def full(self, backtime=86400, retry=42, callback=None):
 147         """Fetches full stream - containing all posts.
 148         WARNING: this is a **VERY** long running function.
 149         Use callback parameter to access information about the stream during its
 150         run.
 151
 152         Default backtime is one day. But sometimes user might not have any activity for longer
 153         period (in the beginning of my D* activity I was posting once a month or so).
 154         The role of retry is to hadle such situations by trying to go further back in time.
 155         If a post is found the counter is restored.
 156
 157         Default retry is 42. If you don't know why go to the nearest library (or to the nearest
 158         Piratebay mirror) and grab a copy of "A Hitchhiker's Guide to the Galaxy" and read the
 159         book to find out. This will also increase your level of geekiness and you'll have a
 160         great time reading the book.
 161
 162         :param backtime: how many seconds to substract each time
 163         :type backtime: int
 164         :param retry: how many times the functin should look deeper than your last post
 165         :type retry: int
 166         :param callback: callable taking diaspy.streams.Generic as an argument
 167         :returns: integer, lenght of the stream
 168         """
 169         oldstream = self.copy()
 170         self.more()
 171         while len(oldstream) < len(self):
 172             oldstream = self.copy()
 173             if callback is not None: callback(self)
 174             self.more(backtime=backtime)
 175             if len(oldstream) < len(self): continue
 176             # but if no posts were found start retrying...
 177             print('retrying... {0}'.format(retry))
 178             n = retry
 179             while n > 0:
 180                 print('\t', n, self.max_time)
 181                 # try to get even more posts...
 182                 self.more(backtime=backtime)
 183                 print('\t', len(oldstream), len(self))
 184                 # check if it was a success...
 185                 if len(oldstream) < len(self):
 186                     # and if so restore normal order of execution by
 187                     # going one loop higher
 188                     break
 189                 oldstream = self.copy()
 190                 # if it was not a success substract one backtime, keep calm and
 191                 # try going further back in time...
 192                 n -= 1
 193             # check the comment below
 194             # no commented code should be present in good software
 195             #if len(oldstream) == len(self): break
 196         return len(self)
 197
 198     def copy(self):
 199         """Returns copy (list of posts) of current stream.
 200         """
 201         return [p for p in self._stream]
 202
 203     def json(self, comments=False, **kwargs):
 204         """Returns JSON encoded string containing stream's data.
 205
 206         :param comments: to include comments or not to include 'em, that is the question this param holds answer to
 207         :type comments: bool
 208         """
 209         stream = [post for post in self._stream]
 210         if comments:
 211             for i, post in enumerate(stream):
 212                 post._fetchcomments()
 213                 comments = [c.data for c in post.comments]
 214                 post['interactions']['comments'] = comments
 215                 stream[i] = post
 216         stream = [post._data for post in stream]
 217         return json.dumps(stream, **kwargs)
 218
 219
 220 class Outer(Generic):
 221     """Object used by diaspy.models.User to represent
 222     stream of other user.
 223     """
 224     def __init__(self, connection, guid, fetch=True):
 225         location = 'people/{}/stream.json'.format(guid)
 226         super().__init__(connection, location, fetch)
 227
 228     def _obtain(self, max_time=0):
 229         """Obtains stream from pod.
 230         """
 231         params = {}
 232         if max_time: params['max_time'] = max_time
 233         request = self._connection.get(self._location, params=params)
 234         if request.status_code != 200:
 235             raise errors.StreamError('wrong status code: {0}'.format(request.status_code))
 236         return [Post(self._connection, post['id']) for post in request.json()]
 237
 238
 239 class Stream(Generic):
 240     """The main stream containing the combined posts of the
 241     followed users and tags and the community spotlights posts
 242     if the user enabled those.
 243     """
 244     location = 'stream.json'
 245
 246     def post(self, text='', aspect_ids='public', photos=None, photo='', provider_display_name=''):
 247         """This function sends a post to an aspect.
 248         If both `photo` and `photos` are specified `photos` takes precedence.
 249
 250         :param text: Text to post.
 251         :type text: str
 252         :param aspect_ids: Aspect ids to send post to.
 253         :type aspect_ids: str
 254         :param photo: filename of photo to post
 255         :type photo: str
 256         :param photos: id of photo to post (obtained from _photoupload())
 257         :type photos: int
 258         :param provider_display_name: name of provider displayed under the post
 259         :type provider_display_name: str
 260
 261         :returns: diaspy.models.Post -- the Post which has been created
 262         """
 263         data = {}
 264         data['aspect_ids'] = aspect_ids
 265         data['status_message'] = {'text': text, 'provider_display_name': provider_display_name}
 266         if photo: data['photos'] = self._photoupload(photo)
 267         if photos: data['photos'] = photos
 268
 269         request = self._connection.post('status_messages',
 270                                         data=json.dumps(data),
 271                                         headers={'content-type': 'application/json',
 272                                                  'accept': 'application/json',
 273                                                  'x-csrf-token': repr(self._connection)})
 274         if request.status_code != 201:
 275             raise Exception('{0}: Post could not be posted.'.format(request.status_code))
 276         post = Post(self._connection, request.json()['id'])
 277         return post
 278
 279     def _photoupload(self, filename, aspects=[]):
 280         """Uploads picture to the pod.
 281
 282         :param filename: path to picture file
 283         :type filename: str
 284         :param aspect_ids: list of ids of aspects to which you want to upload this photo
 285         :type aspect_ids: list of integers
 286
 287         :returns: id of the photo being uploaded
 288         """
 289         data = open(filename, 'rb')
 290         image = data.read()
 291         data.close()
 292
 293         params = {}
 294         params['photo[pending]'] = 'true'
 295         params['set_profile_image'] = ''
 296         params['qqfile'] = filename
 297         if not aspects: aspects = self._connection.getUserData()['aspects']
 298         for i, aspect in enumerate(aspects):
 299             params['photo[aspect_ids][{0}]'.format(i)] = aspect['id']
 300
 301         headers = {'content-type': 'application/octet-stream',
 302                    'x-csrf-token': repr(self._connection),
 303                    'x-file-name': filename}
 304
 305         request = self._connection.post('photos', data=image, params=params, headers=headers)
 306         if request.status_code != 200:
 307             raise errors.StreamError('photo cannot be uploaded: {0}'.format(request.status_code))
 308         return request.json()['data']['photo']['id']
 309
 310
 311 class Activity(Stream):
 312     """Stream representing user's activity.
 313     """
 314     _location = 'activity.json'
 315
 316     def _delid(self, id):
 317         """Deletes post with given id.
 318         """
 319         post = None
 320         for p in self._stream:
 321             if p['id'] == id:
 322                 post = p
 323                 break
 324         if post is not None: post.delete()
 325
 326     def delete(self, post):
 327         """Deletes post from users activity.
 328         `post` can be either post id or Post()
 329         object which will be identified and deleted.
 330         After deleting post the stream will be purged.
 331
 332         :param post: post identifier
 333         :type post: str, diaspy.models.Post
 334         """
 335         if type(post) == str: self._delid(post)
 336         elif type(post) == Post: post.delete()
 337         else: raise TypeError('this method accepts str or Post types: {0} given')
 338         self.purge()
 339
 340
 341 class Aspects(Generic):
 342     """This stream contains the posts filtered by
 343     the specified aspect IDs. You can choose the aspect IDs with
 344     the parameter `aspect_ids` which value should be
 345     a comma seperated list of aspect IDs.
 346     If the parameter is ommitted all aspects are assumed.
 347     An example call would be `aspects.json?aspect_ids=23,5,42`
 348     """
 349     _location = 'aspects.json'
 350
 351     def getAspectID(self, aspect_name):
 352         """Returns id of an aspect of given name.
 353         Returns -1 if aspect is not found.
 354
 355         :param aspect_name: aspect name (must be spelled exactly as when created)
 356         :type aspect_name: str
 357         :returns: int
 358         """
 359         id = -1
 360         aspects = self._connection.getUserData()['aspects']
 361         for aspect in aspects:
 362             if aspect['name'] == aspect_name: id = aspect['id']
 363         return id
 364
 365     def filter(self, ids):
 366         """Filters posts by given aspect ids.
 367
 368         :parameter ids: list of apsect ids
 369         :type ids: list of integers
 370         """
 371         self._location = 'aspects.json' + '?{0}'.format(','.join(ids))
 372         self.fill()
 373
 374     def add(self, aspect_name, visible=0):
 375         """This function adds a new aspect.
 376         Status code 422 is accepted because it is returned by D* when
 377         you try to add aspect already present on your aspect list.
 378
 379         :param aspect_name: name of aspect to create
 380         :param visible: whether the contacts in this aspect are visible to each other or not
 381
 382         :returns: Aspect() object of just created aspect
 383         """
 384         data = {'authenticity_token': repr(self._connection),
 385                 'aspect[name]': aspect_name,
 386                 'aspect[contacts_visible]': visible}
 387
 388         request = self._connection.post('aspects', data=data)
 389         if request.status_code not in [200, 422]:
 390             raise Exception('wrong status code: {0}'.format(request.status_code))
 391
 392         id = self.getAspectID(aspect_name)
 393         return Aspect(self._connection, id)
 394
 395     def remove(self, id=-1, name=''):
 396         """This method removes an aspect.
 397         You can give it either id or name of the aspect.
 398         When both are specified, id takes precedence over name.
 399
 400         Status code 500 is accepted because although the D* will
 401         go nuts it will remove the aspect anyway.
 402
 403         :param aspect_id: id fo aspect to remove
 404         :type aspect_id: int
 405         :param name: name of aspect to remove
 406         :type name: str
 407         """
 408         if id == -1 and name: id = self.getAspectID(name)
 409         data = {'_method': 'delete',
 410                 'authenticity_token': repr(self._connection)}
 411         request = self._connection.post('aspects/{0}'.format(id), data=data)
 412         if request.status_code not in [200, 302, 500]:
 413             raise Exception('wrong status code: {0}: cannot remove aspect'.format(request.status_code))
 414
 415
 416 class Commented(Generic):
 417     """This stream contains all posts
 418     the user has made a comment on.
 419     """
 420     _location = 'commented.json'
 421
 422
 423 class Liked(Generic):
 424     """This stream contains all posts the user liked.
 425     """
 426     _location = 'liked.json'
 427
 428
 429 class Mentions(Generic):
 430     """This stream contains all posts
 431     the user is mentioned in.
 432     """
 433     _location = 'mentions.json'
 434
 435
 436 class FollowedTags(Generic):
 437     """This stream contains all posts
 438     containing tags the user is following.
 439     """
 440     _location = 'followed_tags.json'
 441
 442     def get(self):
 443         """Returns list of followed tags.
 444         """
 445         return []
 446
 447     def remove(self, tag_id):
 448         """Stop following a tag.
 449
 450         :param tag_id: tag id
 451         :type tag_id: int
 452         """
 453         data = {'authenticity_token': self._connection.get_token()}
 454         request = self._connection.delete('tag_followings/{0}'.format(tag_id), data=data)
 455         if request.status_code != 404:
 456             raise Exception('wrong status code: {0}'.format(request.status_code))
 457
 458     def add(self, tag_name):
 459         """Follow new tag.
 460         Error code 403 is accepted because pods respod with it when request
 461         is sent to follow a tag that a user already follows.
 462
 463         :param tag_name: tag name
 464         :type tag_name: str
 465         :returns: int (response code)
 466         """
 467         data = {'name': tag_name,
 468                 'authenticity_token': repr(self._connection),
 469                 }
 470         headers = {'content-type': 'application/json',
 471                    'x-csrf-token': repr(self._connection),
 472                    'accept': 'application/json'
 473                    }
 474
 475         request = self._connection.post('tag_followings', data=json.dumps(data), headers=headers)
 476
 477         if request.status_code not in [201, 403]:
 478             raise Exception('wrong error code: {0}'.format(request.status_code))
 479         return request.status_code
 480
 481
 482 class Tag(Generic):
 483     """This stream contains all posts containing a tag.
 484     """
 485     def __init__(self, connection, tag, fetch=True):
 486         """
 487         :param connection: Connection() object
 488         :type connection: diaspy.connection.Connection
 489         :param tag: tag name
 490         :type tag: str
 491         """
 492         self._connection = connection
 493         self._location = 'tags/{0}.json'.format(tag)
 494         if fetch: self.fill()