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