diff --git a/izzylib/activitypub.py b/izzylib/activitypub.py new file mode 100644 index 0000000..1ced8bc --- /dev/null +++ b/izzylib/activitypub.py @@ -0,0 +1,321 @@ +import json, mimetypes + +from datetime import datetime, timezone +from functools import partial +from typing import Union + +from .dotdict import DotDict +from .misc import DateString, Url + + +pubstr = 'https://www.w3.org/ns/activitystreams#Public' +actor_types = ['Application', 'Group', 'Organization', 'Person', 'Service'] +activity_types = [ + 'Accept', 'Add', 'Announce', 'Arrive', 'Block', 'Create', 'Delete', 'Dislike', + 'Flag', 'Follow', 'Ignore', 'Invite', 'Join', 'Leave', 'Like', 'Listen', + 'Move', 'Offer', 'Question', 'Reject', 'Read', 'Remove', 'TentativeAccept', + 'TentativeReject', 'Travel', 'Undo', 'Update', 'View' + +] + +link_types = ['Mention'] +object_types = [ + 'Article', 'Audio', 'Document', 'Event', 'Image', 'Note', 'Page', 'Place', + 'Profile', 'Relationship', 'Tombstone', 'Video' +] + + +def parse_privacy_level(to: list=[], cc: list=[]): + if to == [pubstr] and len(cc) == 1: + return 'public' + + elif to and self.actor in to[0] and not cc: + return 'unlisted' + + elif pubstr not in to: + return 'private' + + else: + logging.warning('Not sure what this privacy level is') + logging.debug(f'to: {json.dumps(to)}') + logging.debug(f'cc: {json.dumps(cc)}') + + +def generate_privacy_fields(privacy='public'): + if privacy == 'public': + return ([pubstr]) + + +class Object(DotDict): + @property + def privacy_level(self): + return parse_privacy_level(self.get('to', []), self.get('cc', [])) + + + @property + def shared_inbox(self): + try: return self.endpoints.shared_inbox + except KeyError: pass + + + @property + def pubkey(self): + try: return self.publicKey.publicKeyPem + except KeyError: pass + + + @property + def handle(self): + return self.get('preferredUsername') + + + @property + def display_name(self): + return self.get('name') + + + @classmethod + def new_note(cls, id, url, actor, content, **kwargs): + assert False not in map(isinstance, [id, actor, url], [Url]) + + date = kwargs.get('date', DateString.now('activitypub')) + + return cls({ + "@context": [ + "https://www.w3.org/ns/activitystreams", + { + "ostatus": "http://ostatus.org#", + "atomUri": "ostatus:atomUri", + "inReplyToAtomUri": "ostatus:inReplyToAtomUri", + "conversation": "ostatus:conversation", + "sensitive": "as:sensitive", + "toot": "http://joinmastodon.org/ns#", + "votersCount": "toot:votersCount", + "litepub": "http://litepub.social/ns#", + "directMessage": "litepub:directMessage" + } + ], + "id": id, + "type": "Note", + "summary": kwargs.get('summary'), + "inReplyTo": kwargs.get('replyto'), + "published": date, + "url": url, + "attributedTo": actor, + "to": [ + "https://www.w3.org/ns/activitystreams#Public" + ], + "cc": [ + f'{actor}/followers' + ], + "sensitive": kwargs.get('sensitive', False), + "atomUri": id, + "inReplyToAtomUri": kwargs.get('replyto_id'), + "conversation": f'tag:{actor.host},{date.dump_to_string("activitypub-date")}:objectId=490:objectType=Conversation', + "content": content, + "contentMap": { + "en": content + }, + "attachment": [], + "tag": [], + "replies": { + "id": f"{id}/replies", + "type": "Collection", + "first": { + "type": "CollectionPage", + "next": f"{id}/replies?only_other_accounts=true&page=true", + "partOf": f"{id}/replies", + "items": [] + } + } + }) + + + # not complete + @classmethod + def new_actor(cls, actor, handle, pubkey, published=None, table={}, full=True, **kwargs): + actor_type = kwargs.get('type', 'Person').title() + + assert actor_type in actor_types + + actor = Url(actor) + data = cls({ + '@context': [ + 'https://www.w3.org/ns/activitystreams', + 'https://w3id.org/security/v1', + { + 'schema': 'http://schema.org', + 'toot': 'https://joinmastodon.org/ns#', + 'manuallyApprovesFollowers': 'as:manuallyApprovesFollowers', + 'PropertyValue': 'schema:PropertyValue', + 'value': 'schema:value', + 'IdentityProof': 'toot:IdentityProof', + 'discoverable': 'toot:discoverable', + 'Device': 'toot:Device', + 'Ed25519Signature': 'toot:Ed25519Signature', + 'Ed25519Key': 'toot:Ed25519Key', + 'Curve25519Key': 'toot:Curve25519Key', + 'EncryptedMessage': 'toot:EncryptedMessage', + 'publicKeyBase64': 'toot:publicKeyBase64', + 'deviceId': 'toot:deviceId', + 'messageFranking': 'toot:messageFranking', + 'messageType': 'toot:messageType', + 'cipherText': 'toot:cipherText', + 'suspended': 'toot:suspended', + 'Emoji': 'toot:Emoji', + "featured": { + "@id": "toot:featured", + "@type": "@id" + }, + "featuredTags": { + "@id": "toot:featuredTags", + "@type": "@id" + }, + "alsoKnownAs": { + "@id": "as:alsoKnownAs", + "@type": "@id" + }, + "movedTo": { + "@id": "as:movedTo", + "@type": "@id" + }, + "claim": { + "@type": "@id", + "@id": "toot:claim" + }, + "fingerprintKey": { + "@type": "@id", + "@id": "toot:fingerprintKey" + }, + "identityKey": { + "@type": "@id", + "@id": "toot:identityKey" + }, + "devices": { + "@type": "@id", + "@id": "toot:devices" + }, + "focalPoint": { + "@container": "@list", + "@id": "toot:focalPoint" + } + } + ], + 'id': actor, + 'type': actor_type, + 'following': f'{actor}/following', + 'followers': f'{actor}/followers', + 'inbox': kwargs.get('inbox', f'{actor}'), + 'outbox': f'{actor}/outbox', + 'featured': f'{actor}/collections/featured', + 'featuredTags': f'{actor}/collections/tags', + 'preferredUsername': handle, + 'name': kwargs.get('display_name', handle), + 'summary': kwargs.get('bio'), + 'url': kwargs.get('url', actor), + 'manuallyApprovesFollowers': kwargs.get('locked', False), + 'discoverable': kwargs.get('discoverable', False), + 'published': published or DateString.now('activitypub'), + 'devices': f'{actor}/collections/devices', + 'publicKey': { + 'id': f'{actor}#main-key', + 'owner': actor, + 'publicKeyPem': pubkey + }, + 'tag': [], + 'attachment': [], + 'endpoints': { + 'sharedInbox': kwargs.get('shared_inbox', f'https://{actor.host}/inbox') + } + }) + + for key, value in table.items(): + data.attachment.append(PropertyValue(key, value)) + + if kwargs.get('avatar_url'): + data.icon = Object.new_icon(kwargs.get('avatar_url'), kwargs.get('avatar_type')) + + # need to add data when "full" is true + if not full: + del data.featured + del data.featuredTags + del data.devices + del data.following + del data.followers + del data.outbox + + return data + + + + @classmethod + def new_activity(cls, id: str, type: str, actor_src: Union[str, dict], object: Union[str, dict], to: list=[pubstr], cc: list=[]): + assert type in activity_types + + cls({ + '@context': 'https://www.w3.org/ns/activitystreams', + 'type': type, + 'to': to, + 'cc': cc, + 'object': object, + 'id': id, + 'actor': actor_src + }) + + + @classmethod + def new_emoji(cls, id, name, url, image): + return cls({ + 'id': id, + 'type': 'Emoji', + 'name': name, + 'updated': updated or DateTime.now('activitypub'), + 'icon': image + }) + + + @classmethod + def new_image(cls, url, type=None): + return cls({ + 'type': 'Image', + 'mediaType': type or mimetypes.guess_type(url)[0] or 'image/png', + 'url': url + }) + + +class Collection(Object): + @classmethod + def new_replies(cls, id): + { + "id": f"{id}/replies", + "type": "Collection", + "first": { + "type": "CollectionPage", + "next": f"{id}/replies?only_other_accounts=true&page=true", + "partOf": f"{id}/replies", + "items": [] + } + } + + +class PropertyValue(DotDict): + def __init__(self, key, value): + super().__init__({ + 'type': 'PropertyValue', + 'name': key, + 'value': value + }) + + + def __setitem__(self, key, value): + key = key.lower() + + assert key in ['type', 'name', 'value'] + assert type(value) == str + + super().__setitem__(key, value) + + + def set_pair(self, key, value): + self.name = key + self.value = value diff --git a/izzylib/misc.py b/izzylib/misc.py index 83dbc5b..27f6427 100644 --- a/izzylib/misc.py +++ b/izzylib/misc.py @@ -1,6 +1,6 @@ import grp, hashlib, os, platform, random, shlex, signal, socket, statistics, string, time, timeit -from datetime import datetime +from datetime import datetime, timezone from getpass import getpass, getuser from importlib import util from subprocess import Popen, PIPE @@ -12,7 +12,6 @@ from .path import Path __all__ = [ - 'ap_date', 'boolean', 'catch_kb_interrupt', 'get_current_user_info', @@ -31,32 +30,16 @@ __all__ = [ 'time_function_pprint', 'timestamp', 'var_name', + 'DateString', 'Url' ] -def ap_date(date=None, alt=False): - ''' - Takes a datetime object and returns it as an ActivityPub-friendly string - - Arguments: - date (datetime): The datetime object to be converted. It not set, will create a new datetime object with the current date and time - alt (bool): If True, the returned string will be in the Mastodon API format - - Return: - str: The date in an ActivityPub-friendly format - ''' - - if not date: - date = datetime.utcnow() - - elif type(date) == int: - date = datetime.fromtimestamp(date) - - elif type(date) != datetime: - raise TypeError(f'Unsupported object type for ApDate: {type(date)}') - - return date.strftime('%a, %d %b %Y %H:%M:%S GMT' if alt else '%Y-%m-%dT%H:%M:%SZ') +datetime_formats = { + 'http': '%a, %d %b %Y %H:%M:%S GMT', + 'activitypub': '%Y-%m-%dT%H:%M:%SZ', + 'activitypub-date': '%Y-%m-%d' +} def boolean(v, return_value=False): @@ -486,6 +469,68 @@ def var_name(single=True, **kwargs): return key[0] if single else keys +class DateString(str): + tz_utc = timezone.utc + tz_local = datetime.now(tz_utc).astimezone().tzinfo + dt = None + + + def __init__(self, string, format): + assert format in datetime_formats + + self.dt = datetime.strptime(string, datetime_formats[format]).replace(tzinfo=self.tz_utc) + + + def __new__(cls, string, format): + date = str.__new__(cls, string) + return date + + + def __getattr__(self, key): + return getattr(self.dt, key) + + + @classmethod + def from_datetime(cls, date, format): + assert format in datetime_formats + return cls(date.astimezone(cls.tz_utc).strftime(datetime_formats[format]), format) + + + @classmethod + def from_timestamp(cls, timestamp, format): + return cls.from_datetime(datetime.fromtimestamp(timestamp), format) + + + @classmethod + def now(cls, format): + return cls.from_datetime(datetime.now(cls.tz_utc), format) + + + @property + def http(self): + return DateString(self.dump_to_string('http'), 'http') + + + @property + def activitypub(self): + return DateString(self.dump_to_string('activitypub'), 'activitypub') + + + @property + def utc(self): + return self.dt.astimezone(self.tz_utc) + + + @property + def local(self): + return self.dt.astimezone(self.tz_local) + + + def dump_to_string(self, format): + assert format in datetime_formats + return self.dt.strftime(datetime_formats[format]) + + class Url(str): protocols = { 'http': 80, @@ -495,8 +540,6 @@ class Url(str): } def __init__(self, url): - str.__new__(Url, url) - parsed = urlparse(url) self.__parsed = parsed