misc: add DateString class
This commit is contained in:
parent
eb359ed53b
commit
9be0387eed
321
izzylib/activitypub.py
Normal file
321
izzylib/activitypub.py
Normal file
|
|
@ -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
|
||||
|
|
@ -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
|
||||
|
|
|
|||
Loading…
Reference in a new issue