indymedia
/
epicyon
mirror of https://gitlab.com/bashrc2/epicyon
7
4
Fork 1
Code Issues 14 Releases Wiki Activity

Snake case

merge-requests/30/head
Bob Mottram 2022-01-02 21:45:26 +00:00
parent c10f83bb21
commit 019f9b3415
2 changed files with 131 additions and 128 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
languages.py
View File

@ -119,7 +119,7 @@ def understood_post_language(base_dir: str, nickname: str, domain: str,
if msg_object['contentMap'].get(lang): if msg_object['contentMap'].get(lang):
return True return True
# is the language for this post supported by libretranslate? # is the language for this post supported by libretranslate?
libretranslate_url = get_config_param(base_dir, "libretranslate_url") libretranslate_url = get_config_param(base_dir, "libretranslateUrl")
if libretranslate_url: if libretranslate_url:
libretranslate_api_key = \ libretranslate_api_key = \
get_config_param(base_dir, "libretranslateApiKey") get_config_param(base_dir, "libretranslateApiKey")

247
like.py
View File

@ -56,14 +56,14 @@ def liked_by_person(post_json_object: {}, nickname: str, domain: str) -> bool:
""" """
if no_of_likes(post_json_object) == 0: if no_of_likes(post_json_object) == 0:
return False return False
actorMatch = domain + '/users/' + nickname actor_match = domain + '/users/' + nickname
obj = post_json_object obj = post_json_object
if has_object_dict(post_json_object): if has_object_dict(post_json_object):
obj = post_json_object['object'] obj = post_json_object['object']
for item in obj['likes']['items']: for item in obj['likes']['items']:
if item['actor'].endswith(actorMatch): if item['actor'].endswith(actor_match):
return True return True
return False return False
@ -71,8 +71,8 @@ def liked_by_person(post_json_object: {}, nickname: str, domain: str) -> bool:
def _create_like(recent_posts_cache: {}, def _create_like(recent_posts_cache: {},
session, base_dir: str, federation_list: [], session, base_dir: str, federation_list: [],
nickname: str, domain: str, port: int, nickname: str, domain: str, port: int,
ccList: [], http_prefix: str, cc_list: [], http_prefix: str,
objectUrl: str, actorLiked: str, object_url: str, actor_liked: str,
client_to_server: bool, client_to_server: bool,
send_threads: [], postLog: [], send_threads: [], postLog: [],
person_cache: {}, cached_webfingers: {}, person_cache: {}, cached_webfingers: {},
@ -83,58 +83,60 @@ def _create_like(recent_posts_cache: {},
'to' might be a specific person (actor) whose post was liked 'to' might be a specific person (actor) whose post was liked
object is typically the url of the message which was liked object is typically the url of the message which was liked
""" """
if not url_permitted(objectUrl, federation_list): if not url_permitted(object_url, federation_list):
return None return None
fullDomain = get_full_domain(domain, port) full_domain = get_full_domain(domain, port)
newLikeJson = { new_like_json = {
"@context": "https://www.w3.org/ns/activitystreams", "@context": "https://www.w3.org/ns/activitystreams",
'type': 'Like', 'type': 'Like',
'actor': local_actor_url(http_prefix, nickname, fullDomain), 'actor': local_actor_url(http_prefix, nickname, full_domain),
'object': objectUrl 'object': object_url
} }
if ccList: if cc_list:
if len(ccList) > 0: if len(cc_list) > 0:
newLikeJson['cc'] = ccList new_like_json['cc'] = cc_list
# Extract the domain and nickname from a statuses link # Extract the domain and nickname from a statuses link
likedPostNickname = None liked_post_nickname = None
likedPostDomain = None liked_post_domain = None
likedPostPort = None liked_post_port = None
group_account = False group_account = False
if actorLiked: if actor_liked:
likedPostNickname = get_nickname_from_actor(actorLiked) liked_post_nickname = get_nickname_from_actor(actor_liked)
likedPostDomain, likedPostPort = get_domain_from_actor(actorLiked) liked_post_domain, liked_post_port = get_domain_from_actor(actor_liked)
group_account = has_group_type(base_dir, actorLiked, person_cache) group_account = has_group_type(base_dir, actor_liked, person_cache)
else: else:
if has_users_path(objectUrl): if has_users_path(object_url):
likedPostNickname = get_nickname_from_actor(objectUrl) liked_post_nickname = get_nickname_from_actor(object_url)
likedPostDomain, likedPostPort = get_domain_from_actor(objectUrl) liked_post_domain, liked_post_port = \
if '/' + str(likedPostNickname) + '/' in objectUrl: get_domain_from_actor(object_url)
actorLiked = \ if '/' + str(liked_post_nickname) + '/' in object_url:
objectUrl.split('/' + likedPostNickname + '/')[0] + \ actor_liked = \
'/' + likedPostNickname object_url.split('/' + liked_post_nickname + '/')[0] + \
'/' + liked_post_nickname
group_account = \ group_account = \
has_group_type(base_dir, actorLiked, person_cache) has_group_type(base_dir, actor_liked, person_cache)
if likedPostNickname: if liked_post_nickname:
post_filename = locate_post(base_dir, nickname, domain, objectUrl) post_filename = locate_post(base_dir, nickname, domain, object_url)
if not post_filename: if not post_filename:
print('DEBUG: like base_dir: ' + base_dir) print('DEBUG: like base_dir: ' + base_dir)
print('DEBUG: like nickname: ' + nickname) print('DEBUG: like nickname: ' + nickname)
print('DEBUG: like domain: ' + domain) print('DEBUG: like domain: ' + domain)
print('DEBUG: like objectUrl: ' + objectUrl) print('DEBUG: like object_url: ' + object_url)
return None return None
update_likes_collection(recent_posts_cache, update_likes_collection(recent_posts_cache,
base_dir, post_filename, objectUrl, base_dir, post_filename, object_url,
newLikeJson['actor'], new_like_json['actor'],
nickname, domain, debug, None) nickname, domain, debug, None)
send_signed_json(newLikeJson, session, base_dir, send_signed_json(new_like_json, session, base_dir,
nickname, domain, port, nickname, domain, port,
likedPostNickname, likedPostDomain, likedPostPort, liked_post_nickname, liked_post_domain,
liked_post_port,
'https://www.w3.org/ns/activitystreams#Public', 'https://www.w3.org/ns/activitystreams#Public',
http_prefix, True, client_to_server, federation_list, http_prefix, True, client_to_server, federation_list,
send_threads, postLog, cached_webfingers, send_threads, postLog, cached_webfingers,
@ -142,39 +144,39 @@ def _create_like(recent_posts_cache: {},
debug, project_version, None, group_account, debug, project_version, None, group_account,
signing_priv_key_pem, 7367374) signing_priv_key_pem, 7367374)
return newLikeJson return new_like_json
def like_post(recent_posts_cache: {}, def like_post(recent_posts_cache: {},
session, base_dir: str, federation_list: [], session, base_dir: str, federation_list: [],
nickname: str, domain: str, port: int, http_prefix: str, nickname: str, domain: str, port: int, http_prefix: str,
likeNickname: str, likeDomain: str, likePort: int, like_nickname: str, like_domain: str, like_port: int,
ccList: [], cc_list: [],
likeStatusNumber: int, client_to_server: bool, like_status_number: int, client_to_server: bool,
send_threads: [], postLog: [], send_threads: [], postLog: [],
person_cache: {}, cached_webfingers: {}, person_cache: {}, cached_webfingers: {},
debug: bool, project_version: str, debug: bool, project_version: str,
signing_priv_key_pem: str) -> {}: signing_priv_key_pem: str) -> {}:
"""Likes a given status post. This is only used by unit tests """Likes a given status post. This is only used by unit tests
""" """
likeDomain = get_full_domain(likeDomain, likePort) like_domain = get_full_domain(like_domain, like_port)
actorLiked = local_actor_url(http_prefix, likeNickname, likeDomain) actor_liked = local_actor_url(http_prefix, like_nickname, like_domain)
objectUrl = actorLiked + '/statuses/' + str(likeStatusNumber) object_url = actor_liked + '/statuses/' + str(like_status_number)
return _create_like(recent_posts_cache, return _create_like(recent_posts_cache,
session, base_dir, federation_list, session, base_dir, federation_list,
nickname, domain, port, nickname, domain, port,
ccList, http_prefix, objectUrl, actorLiked, cc_list, http_prefix, object_url, actor_liked,
client_to_server, client_to_server,
send_threads, postLog, person_cache, cached_webfingers, send_threads, postLog, person_cache, cached_webfingers,
debug, project_version, signing_priv_key_pem) debug, project_version, signing_priv_key_pem)
def send_like_via_server(base_dir: str, session, def send_like_via_server(base_dir: str, session,
fromNickname: str, password: str, from_nickname: str, password: str,
fromDomain: str, fromPort: int, from_domain: str, from_port: int,
http_prefix: str, likeUrl: str, http_prefix: str, like_url: str,
cached_webfingers: {}, person_cache: {}, cached_webfingers: {}, person_cache: {},
debug: bool, project_version: str, debug: bool, project_version: str,
signing_priv_key_pem: str) -> {}: signing_priv_key_pem: str) -> {}:
@ -184,23 +186,23 @@ def send_like_via_server(base_dir: str, session,
print('WARN: No session for send_like_via_server') print('WARN: No session for send_like_via_server')
return 6 return 6
fromDomainFull = get_full_domain(fromDomain, fromPort) from_domain_full = get_full_domain(from_domain, from_port)
actor = local_actor_url(http_prefix, fromNickname, fromDomainFull) actor = local_actor_url(http_prefix, from_nickname, from_domain_full)
newLikeJson = { new_like_json = {
"@context": "https://www.w3.org/ns/activitystreams", "@context": "https://www.w3.org/ns/activitystreams",
'type': 'Like', 'type': 'Like',
'actor': actor, 'actor': actor,
'object': likeUrl 'object': like_url
} }
handle = http_prefix + '://' + fromDomainFull + '/@' + fromNickname handle = http_prefix + '://' + from_domain_full + '/@' + from_nickname
# lookup the inbox for the To handle # lookup the inbox for the To handle
wf_request = webfinger_handle(session, handle, http_prefix, wf_request = webfinger_handle(session, handle, http_prefix,
cached_webfingers, cached_webfingers,
fromDomain, project_version, debug, False, from_domain, project_version, debug, False,
signing_priv_key_pem) signing_priv_key_pem)
if not wf_request: if not wf_request:
if debug: if debug:
@ -211,53 +213,53 @@ def send_like_via_server(base_dir: str, session,
' did not return a dict. ' + str(wf_request)) ' did not return a dict. ' + str(wf_request))
return 1 return 1
postToBox = 'outbox' post_to_box = 'outbox'
# get the actor inbox for the To handle # get the actor inbox for the To handle
originDomain = fromDomain origin_domain = from_domain
(inboxUrl, _, _, fromPersonId, _, _, (inbox_url, _, _, from_person_id, _, _,
_, _) = get_person_box(signing_priv_key_pem, _, _) = get_person_box(signing_priv_key_pem,
originDomain, origin_domain,
base_dir, session, wf_request, base_dir, session, wf_request,
person_cache, person_cache,
project_version, http_prefix, project_version, http_prefix,
fromNickname, fromDomain, from_nickname, from_domain,
postToBox, 72873) post_to_box, 72873)
if not inboxUrl: if not inbox_url:
if debug: if debug:
print('DEBUG: like no ' + postToBox + ' was found for ' + handle) print('DEBUG: like no ' + post_to_box + ' was found for ' + handle)
return 3 return 3
if not fromPersonId: if not from_person_id:
if debug: if debug:
print('DEBUG: like no actor was found for ' + handle) print('DEBUG: like no actor was found for ' + handle)
return 4 return 4
authHeader = create_basic_auth_header(fromNickname, password) auth_header = create_basic_auth_header(from_nickname, password)
headers = { headers = {
'host': fromDomain, 'host': from_domain,
'Content-type': 'application/json', 'Content-type': 'application/json',
'Authorization': authHeader 'Authorization': auth_header
} }
postResult = post_json(http_prefix, fromDomainFull, post_result = post_json(http_prefix, from_domain_full,
session, newLikeJson, [], inboxUrl, session, new_like_json, [], inbox_url,
headers, 3, True) headers, 3, True)
if not postResult: if not post_result:
if debug: if debug:
print('WARN: POST like failed for c2s to ' + inboxUrl) print('WARN: POST like failed for c2s to ' + inbox_url)
return 5 return 5
if debug: if debug:
print('DEBUG: c2s POST like success') print('DEBUG: c2s POST like success')
return newLikeJson return new_like_json
def send_undo_like_via_server(base_dir: str, session, def send_undo_like_via_server(base_dir: str, session,
fromNickname: str, password: str, from_nickname: str, password: str,
fromDomain: str, fromPort: int, from_domain: str, from_port: int,
http_prefix: str, likeUrl: str, http_prefix: str, like_url: str,
cached_webfingers: {}, person_cache: {}, cached_webfingers: {}, person_cache: {},
debug: bool, project_version: str, debug: bool, project_version: str,
signing_priv_key_pem: str) -> {}: signing_priv_key_pem: str) -> {}:
@ -267,27 +269,27 @@ def send_undo_like_via_server(base_dir: str, session,
print('WARN: No session for send_undo_like_via_server') print('WARN: No session for send_undo_like_via_server')
return 6 return 6
fromDomainFull = get_full_domain(fromDomain, fromPort) from_domain_full = get_full_domain(from_domain, from_port)
actor = local_actor_url(http_prefix, fromNickname, fromDomainFull) actor = local_actor_url(http_prefix, from_nickname, from_domain_full)
newUndoLikeJson = { new_undo_like_json = {
"@context": "https://www.w3.org/ns/activitystreams", "@context": "https://www.w3.org/ns/activitystreams",
'type': 'Undo', 'type': 'Undo',
'actor': actor, 'actor': actor,
'object': { 'object': {
'type': 'Like', 'type': 'Like',
'actor': actor, 'actor': actor,
'object': likeUrl 'object': like_url
} }
} }
handle = http_prefix + '://' + fromDomainFull + '/@' + fromNickname handle = http_prefix + '://' + from_domain_full + '/@' + from_nickname
# lookup the inbox for the To handle # lookup the inbox for the To handle
wf_request = webfinger_handle(session, handle, http_prefix, wf_request = webfinger_handle(session, handle, http_prefix,
cached_webfingers, cached_webfingers,
fromDomain, project_version, debug, False, from_domain, project_version, debug, False,
signing_priv_key_pem) signing_priv_key_pem)
if not wf_request: if not wf_request:
if debug: if debug:
@ -299,47 +301,48 @@ def send_undo_like_via_server(base_dir: str, session,
' did not return a dict. ' + str(wf_request)) ' did not return a dict. ' + str(wf_request))
return 1 return 1
postToBox = 'outbox' post_to_box = 'outbox'
# get the actor inbox for the To handle # get the actor inbox for the To handle
originDomain = fromDomain origin_domain = from_domain
(inboxUrl, pubKeyId, pubKey, fromPersonId, sharedInbox, avatarUrl, (inbox_url, _, _, from_person_id, _, _,
displayName, _) = get_person_box(signing_priv_key_pem, _, _) = get_person_box(signing_priv_key_pem,
originDomain, origin_domain,
base_dir, session, wf_request, base_dir, session, wf_request,
person_cache, project_version, person_cache, project_version,
http_prefix, fromNickname, http_prefix, from_nickname,
fromDomain, postToBox, from_domain, post_to_box,
72625) 72625)
if not inboxUrl: if not inbox_url:
if debug: if debug:
print('DEBUG: unlike no ' + postToBox + ' was found for ' + handle) print('DEBUG: unlike no ' + post_to_box + ' was found for ' +
handle)
return 3 return 3
if not fromPersonId: if not from_person_id:
if debug: if debug:
print('DEBUG: unlike no actor was found for ' + handle) print('DEBUG: unlike no actor was found for ' + handle)
return 4 return 4
authHeader = create_basic_auth_header(fromNickname, password) auth_header = create_basic_auth_header(from_nickname, password)
headers = { headers = {
'host': fromDomain, 'host': from_domain,
'Content-type': 'application/json', 'Content-type': 'application/json',
'Authorization': authHeader 'Authorization': auth_header
} }
postResult = post_json(http_prefix, fromDomainFull, post_result = post_json(http_prefix, from_domain_full,
session, newUndoLikeJson, [], inboxUrl, session, new_undo_like_json, [], inbox_url,
headers, 3, True) headers, 3, True)
if not postResult: if not post_result:
if debug: if debug:
print('WARN: POST unlike failed for c2s to ' + inboxUrl) print('WARN: POST unlike failed for c2s to ' + inbox_url)
return 5 return 5
if debug: if debug:
print('DEBUG: c2s POST unlike success') print('DEBUG: c2s POST unlike success')
return newUndoLikeJson return new_undo_like_json
def outbox_like(recent_posts_cache: {}, def outbox_like(recent_posts_cache: {},
@ -361,16 +364,16 @@ def outbox_like(recent_posts_cache: {},
if debug: if debug:
print('DEBUG: c2s like request arrived in outbox') print('DEBUG: c2s like request arrived in outbox')
messageId = remove_id_ending(message_json['object']) message_id = remove_id_ending(message_json['object'])
domain = remove_domain_port(domain) domain = remove_domain_port(domain)
post_filename = locate_post(base_dir, nickname, domain, messageId) post_filename = locate_post(base_dir, nickname, domain, message_id)
if not post_filename: if not post_filename:
if debug: if debug:
print('DEBUG: c2s like post not found in inbox or outbox') print('DEBUG: c2s like post not found in inbox or outbox')
print(messageId) print(message_id)
return True return True
update_likes_collection(recent_posts_cache, update_likes_collection(recent_posts_cache,
base_dir, post_filename, messageId, base_dir, post_filename, message_id,
message_json['actor'], message_json['actor'],
nickname, domain, debug, None) nickname, domain, debug, None)
if debug: if debug:
@ -398,16 +401,16 @@ def outbox_undo_like(recent_posts_cache: {},
if debug: if debug:
print('DEBUG: c2s undo like request arrived in outbox') print('DEBUG: c2s undo like request arrived in outbox')
messageId = remove_id_ending(message_json['object']['object']) message_id = remove_id_ending(message_json['object']['object'])
domain = remove_domain_port(domain) domain = remove_domain_port(domain)
post_filename = locate_post(base_dir, nickname, domain, messageId) post_filename = locate_post(base_dir, nickname, domain, message_id)
if not post_filename: if not post_filename:
if debug: if debug:
print('DEBUG: c2s undo like post not found in inbox or outbox') print('DEBUG: c2s undo like post not found in inbox or outbox')
print(messageId) print(message_id)
return True return True
undo_likes_collection_entry(recent_posts_cache, base_dir, post_filename, undo_likes_collection_entry(recent_posts_cache, base_dir, post_filename,
messageId, message_json['actor'], message_id, message_json['actor'],
domain, debug, None) domain, debug, None)
if debug: if debug:
print('DEBUG: post undo liked via c2s - ' + post_filename) print('DEBUG: post undo liked via c2s - ' + post_filename)
@ -415,7 +418,7 @@ def outbox_undo_like(recent_posts_cache: {},
def update_likes_collection(recent_posts_cache: {}, def update_likes_collection(recent_posts_cache: {},
base_dir: str, post_filename: str, base_dir: str, post_filename: str,
objectUrl: str, actor: str, object_url: str, actor: str,
nickname: str, domain: str, debug: bool, nickname: str, domain: str, debug: bool,
post_json_object: {}) -> None: post_json_object: {}) -> None:
"""Updates the likes collection within a post """Updates the likes collection within a post
@ -428,29 +431,29 @@ def update_likes_collection(recent_posts_cache: {},
# remove any cached version of this post so that the # remove any cached version of this post so that the
# like icon is changed # like icon is changed
remove_post_from_cache(post_json_object, recent_posts_cache) remove_post_from_cache(post_json_object, recent_posts_cache)
cachedPostFilename = \ cached_post_filename = \
get_cached_post_filename(base_dir, nickname, get_cached_post_filename(base_dir, nickname,
domain, post_json_object) domain, post_json_object)
if cachedPostFilename: if cached_post_filename:
if os.path.isfile(cachedPostFilename): if os.path.isfile(cached_post_filename):
try: try:
os.remove(cachedPostFilename) os.remove(cached_post_filename)
except OSError: except OSError:
print('EX: update_likes_collection unable to delete ' + print('EX: update_likes_collection unable to delete ' +
cachedPostFilename) cached_post_filename)
obj = post_json_object obj = post_json_object
if has_object_dict(post_json_object): if has_object_dict(post_json_object):
obj = post_json_object['object'] obj = post_json_object['object']
if not objectUrl.endswith('/likes'): if not object_url.endswith('/likes'):
objectUrl = objectUrl + '/likes' object_url = object_url + '/likes'
if not obj.get('likes'): if not obj.get('likes'):
if debug: if debug:
print('DEBUG: Adding initial like to ' + objectUrl) print('DEBUG: Adding initial like to ' + object_url)
likesJson = { likes_json = {
"@context": "https://www.w3.org/ns/activitystreams", "@context": "https://www.w3.org/ns/activitystreams",
'id': objectUrl, 'id': object_url,
'type': 'Collection', 'type': 'Collection',
"totalItems": 1, "totalItems": 1,
'items': [{ 'items': [{
@ -458,22 +461,22 @@ def update_likes_collection(recent_posts_cache: {},
'actor': actor 'actor': actor
}] }]
} }
obj['likes'] = likesJson obj['likes'] = likes_json
else: else:
if not obj['likes'].get('items'): if not obj['likes'].get('items'):
obj['likes']['items'] = [] obj['likes']['items'] = []
for likeItem in obj['likes']['items']: for like_item in obj['likes']['items']:
if likeItem.get('actor'): if like_item.get('actor'):
if likeItem['actor'] == actor: if like_item['actor'] == actor:
# already liked # already liked
return return
newLike = { new_like = {
'type': 'Like', 'type': 'Like',
'actor': actor 'actor': actor
} }
obj['likes']['items'].append(newLike) obj['likes']['items'].append(new_like)
itlen = len(obj['likes']['items']) it_len = len(obj['likes']['items'])
obj['likes']['totalItems'] = itlen obj['likes']['totalItems'] = it_len
if debug: if debug:
print('DEBUG: saving post with likes added') print('DEBUG: saving post with likes added')