#!/usr/bin/env python3 """This module is only imported in other diaspy modules and MUST NOT import anything. """ import json import copy BS4_SUPPORT=False try: from bs4 import BeautifulSoup except ImportError: import re print("[diaspy] BeautifulSoup not found, falling back on regex.") else: BS4_SUPPORT=True from diaspy import errors class Aspect(): """This class represents an aspect. Class can be initialized by passing either an id and/or name as parameters. If both are missing, an exception will be raised. """ def __init__(self, connection, id, name=None): self._connection = connection self.id, self.name = id, name self._cached = [] def getUsers(self, fetch = True): """Returns list of GUIDs of users who are listed in this aspect. """ if fetch: request = self._connection.get('contacts.json?a_id={}'.format(self.id)) self._cached = request.json() return self._cached def removeAspect(self): """ --> POST /aspects/{id} HTTP/1.1 --> _method=delete&authenticity_token={token} <-- HTTP/1.1 302 Found Removes whole aspect. :returns: None """ request = self._connection.tokenFrom('contacts').delete('aspects/{}'.format(self.id)) if request.status_code != 302: raise errors.AspectError('wrong status code: {0}'.format(request.status_code)) def addUser(self, user_id): """Add user to current aspect. :param user_id: user to add to aspect :type user_id: int :returns: JSON from request --> POST /aspect_memberships HTTP/1.1 --> Accept: application/json, text/javascript, */*; q=0.01 --> Content-Type: application/json; charset=UTF-8 --> {"aspect_id":123,"person_id":123} <-- HTTP/1.1 200 OK """ data = {'aspect_id': self.id, 'person_id': user_id} headers = {'content-type': 'application/json', 'accept': 'application/json'} request = self._connection.tokenFrom('contacts').post('aspect_memberships', data=json.dumps(data), headers=headers) if request.status_code == 400: raise errors.AspectError('duplicate record, user already exists in aspect: {0}'.format(request.status_code)) elif request.status_code == 404: raise errors.AspectError('user not found from this pod: {0}'.format(request.status_code)) elif request.status_code != 200: raise errors.AspectError('wrong status code: {0}'.format(request.status_code)) response = None try: response = request.json() except json.decoder.JSONDecodeError: """ Should be OK now, but I'll leave this commentary here at first to see if anything comes up """ # FIXME For some (?) reason removing users from aspects works, but # adding them is a no-go and Diaspora* kicks us out with CSRF errors. # Weird. pass if response is None: raise errors.CSRFProtectionKickedIn() # Now you should fetchguid(fetch_stream=False) on User to update aspect membership_id's # Or update it locally with the response return response def removeUser(self, user): """Remove user from current aspect. :param user: user to remove from aspect :type user: diaspy.people.User object """ membership_id = None to_remove = None for each in user.aspectMemberships(): if each.get('aspect', {}).get('id') == self.id: membership_id = each.get('id') to_remove = each break # no need to continue if membership_id is None: raise errors.UserIsNotMemberOfAspect(user, self) request = self._connection.delete('aspect_memberships/{0}'.format(membership_id)) if request.status_code == 404: raise errors.AspectError('cannot remove user from aspect, probably tried too fast after adding: {0}'.format(request.status_code)) elif request.status_code != 200: raise errors.AspectError('cannot remove user from aspect: {0}'.format(request.status_code)) if 'contact' in user.data: # User object if to_remove: user.data['contact']['aspect_memberships'].remove( to_remove ) # remove local aspect membership_id else: # User object from Contacts() if to_remove: user.data['aspect_memberships'].remove( to_remove ) # remove local aspect membership_id return request.json() class Notification(): """This class represents single notification. """ if not BS4_SUPPORT: _who_regexp = re.compile(r'/people/([0-9a-f]+)["\']{1} class=["\']{1}hovercardable') _aboutid_regexp = re.compile(r'/posts/[0-9a-f]+') _htmltag_regexp = re.compile('?[a-z]+( *[a-z_-]+=["\'].*?["\'])* */?>') def __init__(self, connection, data): self._connection = connection self.type = data['type'] self._data = data[self.type] self.id = self._data['id'] self.unread = self._data['unread'] def __getitem__(self, key): """Returns a key from notification data. """ return self._data[key] def __str__(self): """Returns notification note. """ if BS4_SUPPORT: soup = BeautifulSoup(self._data['note_html'], 'lxml') media_body = soup.find('div', {"class": "media-body"}) div = media_body.find('div') if div: div.decompose() return media_body.getText().strip() else: string = re.sub(self._htmltag_regexp, '', self._data['note_html']) string = string.strip().split('\n')[0] while ' ' in string: string = string.replace(' ', ' ') return string def __repr__(self): """Returns notification note with more details. """ return '{0}: {1}'.format(self.when(), str(self)) def about(self): """Returns id of post about which the notification is informing OR: If the id is None it means that it's about user so .who() is called. """ if BS4_SUPPORT: soup = BeautifulSoup(self._data['note_html'], 'lxml') id = soup.find('a', {"data-ref": True}) if id: return id['data-ref'] else: return self.who()[0] else: about = self._aboutid_regexp.search(self._data['note_html']) if about is None: about = self.who()[0] else: about = int(about.group(0)[7:]) return about def who(self): """Returns list of guids of the users who caused you to get the notification. """ if BS4_SUPPORT: # Parse the HTML with BS4 soup = BeautifulSoup(self._data['note_html'], 'lxml') hovercardable_soup = soup.findAll('a', {"class": "hovercardable"}) return list(set([soup['href'][8:] for soup in hovercardable_soup])) else: return list(set([who for who in self._who_regexp.findall(self._data['note_html'])])) def when(self): """Returns UTC time as found in note_html. """ return self._data['created_at'] def mark(self, unread=False): """Marks notification to read/unread. Marks notification to read if `unread` is False. Marks notification to unread if `unread` is True. :param unread: which state set for notification :type unread: bool """ headers = {'x-csrf-token': repr(self._connection)} params = {'set_unread': json.dumps(unread)} response = self._connection.put('notifications/{0}'.format(self['id']), params=params, headers=headers) if response.status_code != 200: raise errors.NotificationError('Cannot mark notification: {0}'.format(response.status_code)) self._data['unread'] = unread self.unread = unread class Conversation(): """This class represents a conversation. .. note:: Remember that you need to have access to the conversation. """ if not BS4_SUPPORT: _message_stream_regexp = re.compile(r'