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

Snake case

main
Bob Mottram 2021-12-29 23:16:38 +00:00
parent ed0ec2b675
commit a994392cd0
1 changed files with 42 additions and 42 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

84
availability.py
View File

@ -30,23 +30,23 @@ def set_availability(base_dir: str, nickname: str, domain: str,
# avoid giant strings # avoid giant strings
if len(status) > 128: if len(status) > 128:
return False return False
actorFilename = acct_dir(base_dir, nickname, domain) + '.json' actor_filename = acct_dir(base_dir, nickname, domain) + '.json'
if not os.path.isfile(actorFilename): if not os.path.isfile(actor_filename):
return False return False
actor_json = load_json(actorFilename) actor_json = load_json(actor_filename)
if actor_json: if actor_json:
actor_json['availability'] = status actor_json['availability'] = status
save_json(actor_json, actorFilename) save_json(actor_json, actor_filename)
return True return True
def get_availability(base_dir: str, nickname: str, domain: str) -> str: def get_availability(base_dir: str, nickname: str, domain: str) -> str:
"""Returns the availability for a given person """Returns the availability for a given person
""" """
actorFilename = acct_dir(base_dir, nickname, domain) + '.json' actor_filename = acct_dir(base_dir, nickname, domain) + '.json'
if not os.path.isfile(actorFilename): if not os.path.isfile(actor_filename):
return False return False
actor_json = load_json(actorFilename) actor_json = load_json(actor_filename)
if actor_json: if actor_json:
if not actor_json.get('availability'): if not actor_json.get('availability'):
return None return None
@ -67,10 +67,10 @@ def outbox_availability(base_dir: str, nickname: str, message_json: {},
if not has_object_string(message_json, debug): if not has_object_string(message_json, debug):
return False return False
actorNickname = get_nickname_from_actor(message_json['actor']) actor_nickname = get_nickname_from_actor(message_json['actor'])
if actorNickname != nickname: if actor_nickname != nickname:
return False return False
domain, port = get_domain_from_actor(message_json['actor']) domain, _ = get_domain_from_actor(message_json['actor'])
status = message_json['object'].replace('"', '') status = message_json['object'].replace('"', '')
return set_availability(base_dir, nickname, domain, status) return set_availability(base_dir, nickname, domain, status)
@ -92,69 +92,69 @@ def send_availability_via_server(base_dir: str, session,
domain_full = get_full_domain(domain, port) domain_full = get_full_domain(domain, port)
toUrl = local_actor_url(http_prefix, nickname, domain_full) to_url = local_actor_url(http_prefix, nickname, domain_full)
ccUrl = toUrl + '/followers' cc_url = to_url + '/followers'
newAvailabilityJson = { new_availability_json = {
'type': 'Availability', 'type': 'Availability',
'actor': toUrl, 'actor': to_url,
'object': '"' + status + '"', 'object': '"' + status + '"',
'to': [toUrl], 'to': [to_url],
'cc': [ccUrl] 'cc': [cc_url]
} }
handle = http_prefix + '://' + domain_full + '/@' + nickname handle = http_prefix + '://' + domain_full + '/@' + nickname
# lookup the inbox for the To handle # lookup the inbox for the To handle
wfRequest = webfinger_handle(session, handle, http_prefix, wf_request = webfinger_handle(session, handle, http_prefix,
cached_webfingers, cached_webfingers,
domain, project_version, debug, False, domain, project_version, debug, False,
signing_priv_key_pem) signing_priv_key_pem)
if not wfRequest: if not wf_request:
if debug: if debug:
print('DEBUG: availability webfinger failed for ' + handle) print('DEBUG: availability webfinger failed for ' + handle)
return 1 return 1
if not isinstance(wfRequest, dict): if not isinstance(wf_request, dict):
print('WARN: availability webfinger for ' + handle + print('WARN: availability webfinger for ' + handle +
' did not return a dict. ' + str(wfRequest)) ' 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 = domain origin_domain = 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, wfRequest, base_dir, session, wf_request,
person_cache, project_version, person_cache, project_version,
http_prefix, nickname, http_prefix, nickname,
domain, postToBox, 57262) domain, post_to_box, 57262)
if not inboxUrl: if not inbox_url:
if debug: if debug:
print('DEBUG: availability no ' + postToBox + print('DEBUG: availability no ' + post_to_box +
' was found for ' + handle) ' was found for ' + handle)
return 3 return 3
if not fromPersonId: if not from_person_id:
if debug: if debug:
print('DEBUG: availability no actor was found for ' + handle) print('DEBUG: availability no actor was found for ' + handle)
return 4 return 4
authHeader = create_basic_auth_header(nickname, password) auth_header = create_basic_auth_header(nickname, password)
headers = { headers = {
'host': domain, 'host': domain,
'Content-type': 'application/json', 'Content-type': 'application/json',
'Authorization': authHeader 'Authorization': auth_header
} }
postResult = post_json(http_prefix, domain_full, post_result = post_json(http_prefix, domain_full,
session, newAvailabilityJson, [], session, new_availability_json, [],
inboxUrl, headers, 30, True) inbox_url, headers, 30, True)
if not postResult: if not post_result:
print('WARN: availability failed to post') print('WARN: availability failed to post')
if debug: if debug:
print('DEBUG: c2s POST availability success') print('DEBUG: c2s POST availability success')
return newAvailabilityJson return new_availability_json