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=84600):
 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=84600, 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 _obtain(self, max_time=0):
 225         """Obtains stream from pod.
 226         """
 227         params = {}
 228         if max_time: params['max_time'] = max_time
 229         request = self._connection.get(self._location, params=params)
 230         if request.status_code != 200:
 231             raise errors.StreamError('wrong status code: {0}'.format(request.status_code))
 232         return [Post(self._connection, post['id']) for post in request.json()]
 233
 234
 235 class Stream(Generic):
 236     """The main stream containing the combined posts of the
 237     followed users and tags and the community spotlights posts
 238     if the user enabled those.
 239     """
 240     location = 'stream.json'
 241
 242     def post(self, text='', aspect_ids='public', photos=None, photo='', provider_display_name=''):
 243         """This function sends a post to an aspect.
 244         If both `photo` and `photos` are specified `photos` takes precedence.
 245
 246         :param text: Text to post.
 247         :type text: str
 248         :param aspect_ids: Aspect ids to send post to.
 249         :type aspect_ids: str
 250         :param photo: filename of photo to post
 251         :type photo: str
 252         :param photos: id of photo to post (obtained from _photoupload())
 253         :type photos: int
 254         :param provider_display_name: name of provider displayed under the post
 255         :type provider_display_name: str
 256
 257         :returns: diaspy.models.Post -- the Post which has been created
 258         """
 259         data = {}
 260         data['aspect_ids'] = aspect_ids
 261         data['status_message'] = {'text': text, 'provider_display_name': provider_display_name}
 262         if photo: data['photos'] = self._photoupload(photo)
 263         if photos: data['photos'] = photos
 264
 265         request = self._connection.post('status_messages',
 266                                         data=json.dumps(data),
 267                                         headers={'content-type': 'application/json',
 268                                                  'accept': 'application/json',
 269                                                  'x-csrf-token': repr(self._connection)})
 270         if request.status_code != 201:
 271             raise Exception('{0}: Post could not be posted.'.format(request.status_code))
 272         post = Post(self._connection, request.json()['id'])
 273         return post
 274
 275     def _photoupload(self, filename, aspects=[]):
 276         """Uploads picture to the pod.
 277
 278         :param filename: path to picture file
 279         :type filename: str
 280         :param aspect_ids: list of ids of aspects to which you want to upload this photo
 281         :type aspect_ids: list of integers
 282
 283         :returns: id of the photo being uploaded
 284         """
 285         data = open(filename, 'rb')
 286         image = data.read()
 287         data.close()
 288
 289         params = {}
 290         params['photo[pending]'] = 'true'
 291         params['set_profile_image'] = ''
 292         params['qqfile'] = filename
 293         if not aspects: aspects = self._connection.getUserData()['aspects']
 294         for i, aspect in enumerate(aspects):
 295             params['photo[aspect_ids][{0}]'.format(i)] = aspect['id']
 296
 297         headers = {'content-type': 'application/octet-stream',
 298                    'x-csrf-token': repr(self._connection),
 299                    'x-file-name': filename}
 300
 301         request = self._connection.post('photos', data=image, params=params, headers=headers)
 302         if request.status_code != 200:
 303             raise errors.StreamError('photo cannot be uploaded: {0}'.format(request.status_code))
 304         return request.json()['data']['photo']['id']
 305
 306
 307 class Activity(Stream):
 308     """Stream representing user's activity.
 309     """
 310     _location = 'activity.json'
 311
 312     def _delid(self, id):
 313         """Deletes post with given id.
 314         """
 315         post = None
 316         for p in self._stream:
 317             if p['id'] == id:
 318                 post = p
 319                 break
 320         if post is not None: post.delete()
 321
 322     def delete(self, post):
 323         """Deletes post from users activity.
 324         `post` can be either post id or Post()
 325         object which will be identified and deleted.
 326         After deleting post the stream will be purged.
 327
 328         :param post: post identifier
 329         :type post: str, diaspy.models.Post
 330         """
 331         if type(post) == str: self._delid(post)
 332         elif type(post) == Post: post.delete()
 333         else: raise TypeError('this method accepts str or Post types: {0} given')
 334         self.purge()
 335
 336
 337 class Aspects(Generic):
 338     """This stream contains the posts filtered by
 339     the specified aspect IDs. You can choose the aspect IDs with
 340     the parameter `aspect_ids` which value should be
 341     a comma seperated list of aspect IDs.
 342     If the parameter is ommitted all aspects are assumed.
 343     An example call would be `aspects.json?aspect_ids=23,5,42`
 344     """
 345     _location = 'aspects.json'
 346
 347     def getAspectID(self, aspect_name):
 348         """Returns id of an aspect of given name.
 349         Returns -1 if aspect is not found.
 350
 351         :param aspect_name: aspect name (must be spelled exactly as when created)
 352         :type aspect_name: str
 353         :returns: int
 354         """
 355         id = -1
 356         aspects = self._connection.getUserData()['aspects']
 357         for aspect in aspects:
 358             if aspect['name'] == aspect_name: id = aspect['id']
 359         return id
 360
 361     def filter(self, ids):
 362         """Filters posts by given aspect ids.
 363
 364         :parameter ids: list of apsect ids
 365         :type ids: list of integers
 366         """
 367         self._location = 'aspects.json' + '?{0}'.format(','.join(ids))
 368         self.fill()
 369
 370     def add(self, aspect_name, visible=0):
 371         """This function adds a new aspect.
 372         Status code 422 is accepted because it is returned by D* when
 373         you try to add aspect already present on your aspect list.
 374
 375         :param aspect_name: name of aspect to create
 376         :param visible: whether the contacts in this aspect are visible to each other or not
 377
 378         :returns: Aspect() object of just created aspect
 379         """
 380         data = {'authenticity_token': repr(self._connection),
 381                 'aspect[name]': aspect_name,
 382                 'aspect[contacts_visible]': visible}
 383
 384         request = self._connection.post('aspects', data=data)
 385         if request.status_code not in [200, 422]:
 386             raise Exception('wrong status code: {0}'.format(request.status_code))
 387
 388         id = self.getAspectID(aspect_name)
 389         return Aspect(self._connection, id)
 390
 391     def remove(self, id=-1, name=''):
 392         """This method removes an aspect.
 393         You can give it either id or name of the aspect.
 394         When both are specified, id takes precedence over name.
 395
 396         Status code 500 is accepted because although the D* will
 397         go nuts it will remove the aspect anyway.
 398
 399         :param aspect_id: id fo aspect to remove
 400         :type aspect_id: int
 401         :param name: name of aspect to remove
 402         :type name: str
 403         """
 404         if id == -1 and name: id = self.getAspectID(name)
 405         data = {'_method': 'delete',
 406                 'authenticity_token': repr(self._connection)}
 407         request = self._connection.post('aspects/{0}'.format(id), data=data)
 408         if request.status_code not in [200, 302, 500]:
 409             raise Exception('wrong status code: {0}: cannot remove aspect'.format(request.status_code))
 410
 411
 412 class Commented(Generic):
 413     """This stream contains all posts
 414     the user has made a comment on.
 415     """
 416     _location = 'commented.json'
 417
 418
 419 class Liked(Generic):
 420     """This stream contains all posts the user liked.
 421     """
 422     _location = 'liked.json'
 423
 424
 425 class Mentions(Generic):
 426     """This stream contains all posts
 427     the user is mentioned in.
 428     """
 429     _location = 'mentions.json'
 430
 431
 432 class FollowedTags(Generic):
 433     """This stream contains all posts
 434     containing tags the user is following.
 435     """
 436     _location = 'followed_tags.json'
 437
 438     def get(self):
 439         """Returns list of followed tags.
 440         """
 441         return []
 442
 443     def remove(self, tag_id):
 444         """Stop following a tag.
 445
 446         :param tag_id: tag id
 447         :type tag_id: int
 448         """
 449         data = {'authenticity_token': self._connection.get_token()}
 450         request = self._connection.delete('tag_followings/{0}'.format(tag_id), data=data)
 451         if request.status_code != 404:
 452             raise Exception('wrong status code: {0}'.format(request.status_code))
 453
 454     def add(self, tag_name):
 455         """Follow new tag.
 456         Error code 403 is accepted because pods respod with it when request
 457         is sent to follow a tag that a user already follows.
 458
 459         :param tag_name: tag name
 460         :type tag_name: str
 461         :returns: int (response code)
 462         """
 463         data = {'name': tag_name,
 464                 'authenticity_token': repr(self._connection),
 465                 }
 466         headers = {'content-type': 'application/json',
 467                    'x-csrf-token': repr(self._connection),
 468                    'accept': 'application/json'
 469                    }
 470
 471         request = self._connection.post('tag_followings', data=json.dumps(data), headers=headers)
 472
 473         if request.status_code not in [201, 403]:
 474             raise Exception('wrong error code: {0}'.format(request.status_code))
 475         return request.status_code
 476
 477
 478 class Tag(Generic):
 479     """This stream contains all posts containing a tag.
 480     """
 481     def __init__(self, connection, tag, fetch=True):
 482         """
 483         :param connection: Connection() object
 484         :type connection: diaspy.connection.Connection
 485         :param tag: tag name
 486         :type tag: str
 487         """
 488         self._connection = connection
 489         self._location = 'tags/{0}.json'.format(tag)
 490         if fetch: self.fill()