Source code for granary.as2

"""Convert between ActivityStreams 1 and 2.


import copy
import datetime
import logging

from oauth_dropins.webutil import util

from . import as1

logger = logging.getLogger(__name__)

# ActivityPub Content-Type details:
CONTENT_TYPE = 'application/activity+json'
CONTENT_TYPE_LD = 'application/ld+json; profile=""'
    'Accept': f'{CONTENT_TYPE}; q=0.9, {CONTENT_TYPE_LD}; q=0.8',

# All known Public values, cargo culted from:
PUBLICS = frozenset((

def _invert(d):
  return {v: k for k, v in d.items()}

  'article': 'Article',
  'audio': 'Audio',
  'collection': 'Collection',
  'comment': 'Note',
  'event': 'Event',
  'hashtag': 'Tag',  # not in AS2 spec; needed for correct round trip conversion
  'image': 'Image',
  # not in AS1 spec; needed to identify mentions in eg Bridgy Fed
  'mention': 'Mention',
  'note': 'Note',
  'person': 'Person',
  'place': 'Place',
  'video': 'Video',
TYPE_TO_OBJECT_TYPE['Note'] = 'note'  # disambiguate

  'favorite': 'Like',
  'follow': 'Follow',
  'invite': 'Invite',
  'like': 'Like',
  'post': 'Create',
  'rsvp-maybe': 'TentativeAccept',
  'rsvp-no': 'Reject',
  'rsvp-yes': 'Accept',
  'share': 'Announce',
  'tag': 'Add',
  'update': 'Update',
TYPE_TO_VERB['Like'] = 'like'  # disambiguate

[docs]def from_as1(obj, type=None, context=CONTEXT, top_level=True): """Converts an ActivityStreams 1 activity or object to ActivityStreams 2. Args: obj: dict, AS1 activity or object type: string, default type if type inference can't determine a type. context: string, included as @context Returns: dict, AS2 activity or object """ if not obj: return {} elif not isinstance(obj, dict): raise ValueError(f'Expected dict, got {obj!r}') obj = copy.deepcopy(obj) verb = obj.pop('verb', None) obj_type = obj.pop('objectType', None) type = (OBJECT_TYPE_TO_TYPE.get(verb or obj_type) or VERB_TO_TYPE.get(verb or obj_type) or type) if context: obj['@context'] = context def all_from_as1(field, type=None): return [from_as1(elem, type=type, context=None, top_level=False) for elem in util.pop_list(obj, field)] inner_objs = all_from_as1('object') if len(inner_objs) == 1: inner_objs = inner_objs[0] obj.update({ 'type': type, 'name': obj.pop('displayName', None), 'actor': from_as1(obj.get('actor'), context=None, top_level=False), 'attachment': all_from_as1('attachments'), 'attributedTo': all_from_as1('author', type='Person'), 'inReplyTo': util.trim_nulls([orig.get('id') or orig.get('url') for orig in obj.get('inReplyTo', [])]), 'object': inner_objs, 'tag': all_from_as1('tags'), 'preferredUsername': obj.pop('username', None), 'url': as1.object_urls(obj), }) obj.pop('urls', None) # images; separate featured (aka header) and non-featured. images = util.get_list(obj, 'image') featured = [] non_featured = [] if images: for img in images: # objectType featured is non-standard; granary uses it for u-featured # microformats2 images if img.get('objectType') == 'featured': featured.append(img) else: non_featured.append(img) # prefer non-featured first for icon, featured for image. ActivityPub/Mastodon # use icon for profile picture, image for header. if obj_type == 'person': obj['icon'] = from_as1((non_featured or featured)[0], type='Image', context=None, top_level=False) obj['image'] = [from_as1(img, type='Image', context=None) for img in featured + non_featured] if len(obj['image']) == 1: obj['image'] = obj['image'][0] # other type-specific fields if obj_type == 'mention': obj['href'] = util.get_first(obj, 'url') obj.pop('url', None) elif obj_type in ('audio', 'video'): stream = util.pop_list(obj, 'stream') if stream: obj.update({ 'url': stream[0].get('url'), # file size in bytes. nonstandard, not in AS2 proper 'size': stream[0].get('size'), }) duration = stream[0].get('duration') if duration: try: # ISO 8601 duration # # obj['duration'] = util.to_iso8601_duration( datetime.timedelta(seconds=duration)) except TypeError: logger.warning(f'Dropping unexpected duration {duration!r}; expected int, is {duration.__class__}') elif obj_type == 'person' and top_level: # Mastodon-specific metadata property fields # obj['attachment'].extend({ 'type': 'PropertyValue', 'name': 'Link', 'value': util.pretty_link(url, attrs={'rel': 'me'}), } for url in obj['url'] if util.is_web(url)) urls = util.get_list(obj, 'url') if len(urls) == 1: obj['url'] = urls[0] loc = obj.get('location') if loc: obj['location'] = from_as1(loc, type='Place', context=None) obj = util.trim_nulls(obj) if list(obj.keys()) == ['url']: return obj['url'] return obj
[docs]def to_as1(obj, use_type=True): """Converts an ActivityStreams 2 activity or object to ActivityStreams 1. Args: obj: dict, AS2 activity or object use_type: boolean, whether to include objectType and verb Returns: dict, AS1 activity or object """ if not obj: return {} elif isinstance(obj, str): return {'url': obj} elif not isinstance(obj, dict): raise ValueError(f'Expected dict, got {obj!r}') obj = copy.deepcopy(obj) obj.pop('@context', None) type = obj.pop('type', None) if use_type: if type and not isinstance(type, str): raise ValueError(f'Expected type to be string, got {type!r}') obj['objectType'] = TYPE_TO_OBJECT_TYPE.get(type) obj['verb'] = TYPE_TO_VERB.get(type) if obj.get('inReplyTo') and obj['objectType'] in ('note', 'article'): obj['objectType'] = 'comment' elif obj['verb'] and not obj['objectType']: obj['objectType'] = 'activity' def url_or_as1(val): return {'url': val} if isinstance(val, str) else to_as1(val) def all_to_as1(field): return [to_as1(elem) for elem in util.pop_list(obj, field) if not (type == 'Person' and elem.get('type') == 'PropertyValue')] images = [] # ActivityPub/Mastodon uses icon for profile picture, image for header. for as2_img in util.pop_list(obj, 'icon') + util.pop_list(obj, 'image'): as1_img = to_as1(as2_img, use_type=False) if as1_img not in images: images.append(as1_img) # inner objects inner_objs = all_to_as1('object') actor = to_as1(obj.get('actor', {})) if type == 'Create': for inner_obj in inner_objs: inner_obj.setdefault('author', {}).update(actor) if len(inner_objs) == 1: inner_objs = inner_objs[0] # attachments. if mediaType is image/..., override type attachments = all_to_as1('attachment') for att in attachments: if att.get('mediaType', '').split('/')[0] == 'image': att['objectType'] = 'image' obj.update({ 'displayName': obj.pop('name', None), 'username': obj.pop('preferredUsername', None), 'actor': actor, 'attachments': attachments, 'image': images, 'inReplyTo': [url_or_as1(orig) for orig in util.get_list(obj, 'inReplyTo')], 'location': url_or_as1(obj.get('location')), 'object': inner_objs, 'tags': all_to_as1('tag'), }) # media if type in ('Audio', 'Video'): duration = util.parse_iso8601_duration(obj.pop('duration', None)) if duration: duration = duration.total_seconds() obj['stream'] = { 'url': obj.pop('url', None), # file size in bytes. nonstandard, not in AS1 proper 'size': obj.pop('size', None), 'duration': duration or None, } elif type == 'Mention': obj['url'] = obj.pop('href', None) # object author attrib = util.pop_list(obj, 'attributedTo') if attrib: if len(attrib) > 1: logger.warning(f'ActivityStreams 1 only supports single author; dropping extra attributedTo values: {attrib[1:]}') obj.setdefault('author', {}).update(to_as1(attrib[0])) return util.trim_nulls(obj)
[docs]def is_public(activity): """Returns True if the given AS2 object or activity is public, False otherwise. Args: activity: dict, AS2 activity or object """ if not isinstance(activity, dict): return False audience = util.get_list(activity, 'to') + util.get_list(activity, 'cc') obj = activity.get('object') if isinstance(obj, dict): audience.extend(util.get_list(obj, 'to') + util.get_list(obj, 'cc')) return bool(PUBLICS.intersection(audience))