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

Snake case

main
Bob Mottram 2022-01-03 10:27:55 +00:00
parent 551f24706b
commit d778105091
9 changed files with 259 additions and 250 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

12
matrix.py
View File

@ -43,7 +43,7 @@ def set_matrix_address(actor_json: {}, matrix_address: str) -> None:
actor_json['attachment'] = [] actor_json['attachment'] = []
# remove any existing value # remove any existing value
propertyFound = None property_found = None
for property_value in actor_json['attachment']: for property_value in actor_json['attachment']:
if not property_value.get('name'): if not property_value.get('name'):
continue continue
@ -51,10 +51,10 @@ def set_matrix_address(actor_json: {}, matrix_address: str) -> None:
continue continue
if not property_value['name'].lower().startswith('matrix'): if not property_value['name'].lower().startswith('matrix'):
continue continue
propertyFound = property_value property_found = property_value
break break
if propertyFound: if property_found:
actor_json['attachment'].remove(propertyFound) actor_json['attachment'].remove(property_found)
if '@' not in matrix_address: if '@' not in matrix_address:
return return
@ -81,9 +81,9 @@ def set_matrix_address(actor_json: {}, matrix_address: str) -> None:
property_value['value'] = matrix_address property_value['value'] = matrix_address
return return
newMatrixAddress = { new_matrix_address = {
"name": "Matrix", "name": "Matrix",
"type": "PropertyValue", "type": "PropertyValue",
"value": matrix_address "value": matrix_address
} }
actor_json['attachment'].append(newMatrixAddress) actor_json['attachment'].append(new_matrix_address)

281
media.py
View File

@ -53,107 +53,107 @@ def _get_blur_hash() -> str:
def _replace_silo_domain(post_json_object: {}, def _replace_silo_domain(post_json_object: {},
siloDomain: str, replacementDomain: str, silo_domain: str, replacement_domain: str,
system_language: str) -> None: system_language: str) -> None:
"""Replace a silo domain with a replacement domain """Replace a silo domain with a replacement domain
""" """
if not replacementDomain: if not replacement_domain:
return return
if not has_object_dict(post_json_object): if not has_object_dict(post_json_object):
return return
if not post_json_object['object'].get('content'): if not post_json_object['object'].get('content'):
return return
contentStr = get_base_content_from_post(post_json_object, system_language) content_str = get_base_content_from_post(post_json_object, system_language)
if siloDomain not in contentStr: if silo_domain not in content_str:
return return
contentStr = contentStr.replace(siloDomain, replacementDomain) content_str = content_str.replace(silo_domain, replacement_domain)
post_json_object['object']['content'] = contentStr post_json_object['object']['content'] = content_str
if post_json_object['object'].get('contentMap'): if post_json_object['object'].get('contentMap'):
post_json_object['object']['contentMap'][system_language] = contentStr post_json_object['object']['contentMap'][system_language] = content_str
def replace_you_tube(post_json_object: {}, replacementDomain: str, def replace_you_tube(post_json_object: {}, replacement_domain: str,
system_language: str) -> None: system_language: str) -> None:
"""Replace YouTube with a replacement domain """Replace YouTube with a replacement domain
This denies Google some, but not all, tracking data This denies Google some, but not all, tracking data
""" """
_replace_silo_domain(post_json_object, 'www.youtube.com', _replace_silo_domain(post_json_object, 'www.youtube.com',
replacementDomain, system_language) replacement_domain, system_language)
def replace_twitter(post_json_object: {}, replacementDomain: str, def replace_twitter(post_json_object: {}, replacement_domain: str,
system_language: str) -> None: system_language: str) -> None:
"""Replace Twitter with a replacement domain """Replace Twitter with a replacement domain
This allows you to view twitter posts without having a twitter account This allows you to view twitter posts without having a twitter account
""" """
_replace_silo_domain(post_json_object, 'twitter.com', _replace_silo_domain(post_json_object, 'twitter.com',
replacementDomain, system_language) replacement_domain, system_language)
def _remove_meta_data(image_filename: str, outputFilename: str) -> None: def _remove_meta_data(image_filename: str, output_filename: str) -> None:
"""Attempts to do this with pure python didn't work well, """Attempts to do this with pure python didn't work well,
so better to use a dedicated tool if one is installed so better to use a dedicated tool if one is installed
""" """
copyfile(image_filename, outputFilename) copyfile(image_filename, output_filename)
if not os.path.isfile(outputFilename): if not os.path.isfile(output_filename):
print('ERROR: unable to remove metadata from ' + image_filename) print('ERROR: unable to remove metadata from ' + image_filename)
return return
if os.path.isfile('/usr/bin/exiftool'): if os.path.isfile('/usr/bin/exiftool'):
print('Removing metadata from ' + outputFilename + ' using exiftool') print('Removing metadata from ' + output_filename + ' using exiftool')
os.system('exiftool -all= ' + outputFilename) # nosec os.system('exiftool -all= ' + output_filename) # nosec
elif os.path.isfile('/usr/bin/mogrify'): elif os.path.isfile('/usr/bin/mogrify'):
print('Removing metadata from ' + outputFilename + ' using mogrify') print('Removing metadata from ' + output_filename + ' using mogrify')
os.system('/usr/bin/mogrify -strip ' + outputFilename) # nosec os.system('/usr/bin/mogrify -strip ' + output_filename) # nosec
def _spoof_meta_data(base_dir: str, nickname: str, domain: str, def _spoof_meta_data(base_dir: str, nickname: str, domain: str,
outputFilename: str, spoofCity: str, output_filename: str, spoof_city: str,
content_license_url: str) -> None: content_license_url: str) -> None:
"""Spoof image metadata using a decoy model for a given city """Spoof image metadata using a decoy model for a given city
""" """
if not os.path.isfile(outputFilename): if not os.path.isfile(output_filename):
print('ERROR: unable to spoof metadata within ' + outputFilename) print('ERROR: unable to spoof metadata within ' + output_filename)
return return
# get the random seed used to generate a unique pattern for this account # get the random seed used to generate a unique pattern for this account
decoySeedFilename = acct_dir(base_dir, nickname, domain) + '/decoyseed' decoy_seed_filename = acct_dir(base_dir, nickname, domain) + '/decoyseed'
decoySeed = 63725 decoy_seed = 63725
if os.path.isfile(decoySeedFilename): if os.path.isfile(decoy_seed_filename):
with open(decoySeedFilename, 'r') as fp: with open(decoy_seed_filename, 'r') as fp_seed:
decoySeed = int(fp.read()) decoy_seed = int(fp_seed.read())
else: else:
decoySeed = randint(10000, 10000000000000000) decoy_seed = randint(10000, 10000000000000000)
try: try:
with open(decoySeedFilename, 'w+') as fp: with open(decoy_seed_filename, 'w+') as fp_seed:
fp.write(str(decoySeed)) fp_seed.write(str(decoy_seed))
except OSError: except OSError:
print('EX: unable to write ' + decoySeedFilename) print('EX: unable to write ' + decoy_seed_filename)
if os.path.isfile('/usr/bin/exiftool'): if os.path.isfile('/usr/bin/exiftool'):
print('Spoofing metadata in ' + outputFilename + ' using exiftool') print('Spoofing metadata in ' + output_filename + ' using exiftool')
curr_timeAdjusted = \ curr_time_adjusted = \
datetime.datetime.utcnow() - \ datetime.datetime.utcnow() - \
datetime.timedelta(minutes=randint(2, 120)) datetime.timedelta(minutes=randint(2, 120))
published = curr_timeAdjusted.strftime("%Y:%m:%d %H:%M:%S+00:00") published = curr_time_adjusted.strftime("%Y:%m:%d %H:%M:%S+00:00")
(latitude, longitude, latitudeRef, longitudeRef, (latitude, longitude, latitude_ref, longitude_ref,
camMake, camModel, camSerialNumber) = \ cam_make, cam_model, cam_serial_number) = \
spoof_geolocation(base_dir, spoofCity, curr_timeAdjusted, spoof_geolocation(base_dir, spoof_city, curr_time_adjusted,
decoySeed, None, None) decoy_seed, None, None)
if os.system('exiftool -artist=@"' + nickname + '@' + domain + '" ' + if os.system('exiftool -artist=@"' + nickname + '@' + domain + '" ' +
'-Make="' + camMake + '" ' + '-Make="' + cam_make + '" ' +
'-Model="' + camModel + '" ' + '-Model="' + cam_model + '" ' +
'-Comment="' + str(camSerialNumber) + '" ' + '-Comment="' + str(cam_serial_number) + '" ' +
'-DateTimeOriginal="' + published + '" ' + '-DateTimeOriginal="' + published + '" ' +
'-FileModifyDate="' + published + '" ' + '-FileModifyDate="' + published + '" ' +
'-CreateDate="' + published + '" ' + '-CreateDate="' + published + '" ' +
'-GPSLongitudeRef=' + longitudeRef + ' ' + '-GPSLongitudeRef=' + longitude_ref + ' ' +
'-GPSAltitude=0 ' + '-GPSAltitude=0 ' +
'-GPSLongitude=' + str(longitude) + ' ' + '-GPSLongitude=' + str(longitude) + ' ' +
'-GPSLatitudeRef=' + latitudeRef + ' ' + '-GPSLatitudeRef=' + latitude_ref + ' ' +
'-GPSLatitude=' + str(latitude) + ' ' + '-GPSLatitude=' + str(latitude) + ' ' +
'-copyright="' + content_license_url + '" ' + '-copyright="' + content_license_url + '" ' +
'-Comment="" ' + '-Comment="" ' +
outputFilename) != 0: # nosec output_filename) != 0: # nosec
print('ERROR: exiftool failed to run') print('ERROR: exiftool failed to run')
else: else:
print('ERROR: exiftool is not installed') print('ERROR: exiftool is not installed')
@ -163,54 +163,54 @@ def _spoof_meta_data(base_dir: str, nickname: str, domain: str,
def convert_image_to_low_bandwidth(image_filename: str) -> None: def convert_image_to_low_bandwidth(image_filename: str) -> None:
"""Converts an image to a low bandwidth version """Converts an image to a low bandwidth version
""" """
low_bandwidthFilename = image_filename + '.low' low_bandwidth_filename = image_filename + '.low'
if os.path.isfile(low_bandwidthFilename): if os.path.isfile(low_bandwidth_filename):
try: try:
os.remove(low_bandwidthFilename) os.remove(low_bandwidth_filename)
except OSError: except OSError:
print('EX: convert_image_to_low_bandwidth unable to delete ' + print('EX: convert_image_to_low_bandwidth unable to delete ' +
low_bandwidthFilename) low_bandwidth_filename)
cmd = \ cmd = \
'/usr/bin/convert +noise Multiplicative ' + \ '/usr/bin/convert +noise Multiplicative ' + \
'-evaluate median 10% -dither Floyd-Steinberg ' + \ '-evaluate median 10% -dither Floyd-Steinberg ' + \
'-monochrome ' + image_filename + ' ' + low_bandwidthFilename '-monochrome ' + image_filename + ' ' + low_bandwidth_filename
print('Low bandwidth image conversion: ' + cmd) print('Low bandwidth image conversion: ' + cmd)
subprocess.call(cmd, shell=True) subprocess.call(cmd, shell=True)
# wait for conversion to happen # wait for conversion to happen
ctr = 0 ctr = 0
while not os.path.isfile(low_bandwidthFilename): while not os.path.isfile(low_bandwidth_filename):
print('Waiting for low bandwidth image conversion ' + str(ctr)) print('Waiting for low bandwidth image conversion ' + str(ctr))
time.sleep(0.2) time.sleep(0.2)
ctr += 1 ctr += 1
if ctr > 100: if ctr > 100:
print('WARN: timed out waiting for low bandwidth image conversion') print('WARN: timed out waiting for low bandwidth image conversion')
break break
if os.path.isfile(low_bandwidthFilename): if os.path.isfile(low_bandwidth_filename):
try: try:
os.remove(image_filename) os.remove(image_filename)
except OSError: except OSError:
print('EX: convert_image_to_low_bandwidth unable to delete ' + print('EX: convert_image_to_low_bandwidth unable to delete ' +
image_filename) image_filename)
os.rename(low_bandwidthFilename, image_filename) os.rename(low_bandwidth_filename, image_filename)
if os.path.isfile(image_filename): if os.path.isfile(image_filename):
print('Image converted to low bandwidth ' + image_filename) print('Image converted to low bandwidth ' + image_filename)
else: else:
print('Low bandwidth converted image not found: ' + print('Low bandwidth converted image not found: ' +
low_bandwidthFilename) low_bandwidth_filename)
def process_meta_data(base_dir: str, nickname: str, domain: str, def process_meta_data(base_dir: str, nickname: str, domain: str,
image_filename: str, outputFilename: str, image_filename: str, output_filename: str,
city: str, content_license_url: str) -> None: city: str, content_license_url: str) -> None:
"""Handles image metadata. This tries to spoof the metadata """Handles image metadata. This tries to spoof the metadata
if possible, but otherwise just removes it if possible, but otherwise just removes it
""" """
# first remove the metadata # first remove the metadata
_remove_meta_data(image_filename, outputFilename) _remove_meta_data(image_filename, output_filename)
# now add some spoofed data to misdirect surveillance capitalists # now add some spoofed data to misdirect surveillance capitalists
_spoof_meta_data(base_dir, nickname, domain, outputFilename, city, _spoof_meta_data(base_dir, nickname, domain, output_filename, city,
content_license_url) content_license_url)
@ -220,65 +220,70 @@ def _is_media(image_filename: str) -> bool:
if not os.path.isfile(image_filename): if not os.path.isfile(image_filename):
print('WARN: Media file does not exist ' + image_filename) print('WARN: Media file does not exist ' + image_filename)
return False return False
permittedMedia = get_media_extensions() permitted_media = get_media_extensions()
for m in permittedMedia: for permit in permitted_media:
if image_filename.endswith('.' + m): if image_filename.endswith('.' + permit):
return True return True
print('WARN: ' + image_filename + ' is not a permitted media type') print('WARN: ' + image_filename + ' is not a permitted media type')
return False return False
def create_media_dirs(base_dir: str, mediaPath: str) -> None: def create_media_dirs(base_dir: str, media_path: str) -> None:
"""Creates stored media directories
"""
if not os.path.isdir(base_dir + '/media'): if not os.path.isdir(base_dir + '/media'):
os.mkdir(base_dir + '/media') os.mkdir(base_dir + '/media')
if not os.path.isdir(base_dir + '/' + mediaPath): if not os.path.isdir(base_dir + '/' + media_path):
os.mkdir(base_dir + '/' + mediaPath) os.mkdir(base_dir + '/' + media_path)
def get_media_path() -> str: def get_media_path() -> str:
"""Returns the path for stored media
"""
curr_time = datetime.datetime.utcnow() curr_time = datetime.datetime.utcnow()
weeksSinceEpoch = int((curr_time - datetime.datetime(1970, 1, 1)).days / 7) weeks_since_epoch = \
return 'media/' + str(weeksSinceEpoch) int((curr_time - datetime.datetime(1970, 1, 1)).days / 7)
return 'media/' + str(weeks_since_epoch)
def get_attachment_media_type(filename: str) -> str: def get_attachment_media_type(filename: str) -> str:
"""Returns the type of media for the given file """Returns the type of media for the given file
image, video or audio image, video or audio
""" """
mediaType = None media_type = None
imageTypes = get_image_extensions() image_types = get_image_extensions()
for mType in imageTypes: for mtype in image_types:
if filename.endswith('.' + mType): if filename.endswith('.' + mtype):
return 'image' return 'image'
videoTypes = get_video_extensions() video_types = get_video_extensions()
for mType in videoTypes: for mtype in video_types:
if filename.endswith('.' + mType): if filename.endswith('.' + mtype):
return 'video' return 'video'
audioTypes = get_audio_extensions() audio_types = get_audio_extensions()
for mType in audioTypes: for mtype in audio_types:
if filename.endswith('.' + mType): if filename.endswith('.' + mtype):
return 'audio' return 'audio'
return mediaType return media_type
def _update_etag(mediaFilename: str) -> None: def _update_etag(media_filename: str) -> None:
""" calculate the etag, which is a sha1 of the data """ calculate the etag, which is a sha1 of the data
""" """
# only create etags for media # only create etags for media
if '/media/' not in mediaFilename: if '/media/' not in media_filename:
return return
# check that the media exists # check that the media exists
if not os.path.isfile(mediaFilename): if not os.path.isfile(media_filename):
return return
# read the binary data # read the binary data
data = None data = None
try: try:
with open(mediaFilename, 'rb') as mediaFile: with open(media_filename, 'rb') as media_file:
data = mediaFile.read() data = media_file.read()
except OSError: except OSError:
print('EX: _update_etag unable to read ' + str(mediaFilename)) print('EX: _update_etag unable to read ' + str(media_filename))
if not data: if not data:
return return
@ -286,17 +291,17 @@ def _update_etag(mediaFilename: str) -> None:
etag = sha1(data).hexdigest() # nosec etag = sha1(data).hexdigest() # nosec
# save the hash # save the hash
try: try:
with open(mediaFilename + '.etag', 'w+') as etagFile: with open(media_filename + '.etag', 'w+') as efile:
etagFile.write(etag) efile.write(etag)
except OSError: except OSError:
print('EX: _update_etag unable to write ' + print('EX: _update_etag unable to write ' +
str(mediaFilename) + '.etag') str(media_filename) + '.etag')
def attach_media(base_dir: str, http_prefix: str, def attach_media(base_dir: str, http_prefix: str,
nickname: str, domain: str, port: int, nickname: str, domain: str, port: int,
post_json: {}, image_filename: str, post_json: {}, image_filename: str,
mediaType: str, description: str, media_type: str, description: str,
city: str, low_bandwidth: bool, city: str, low_bandwidth: bool,
content_license_url: str) -> {}: content_license_url: str) -> {}:
"""Attaches media to a json object post """Attaches media to a json object post
@ -305,76 +310,76 @@ def attach_media(base_dir: str, http_prefix: str,
if not _is_media(image_filename): if not _is_media(image_filename):
return post_json return post_json
fileExtension = None file_extension = None
acceptedTypes = get_media_extensions() accepted_types = get_media_extensions()
for mType in acceptedTypes: for mtype in accepted_types:
if image_filename.endswith('.' + mType): if image_filename.endswith('.' + mtype):
if mType == 'jpg': if mtype == 'jpg':
mType = 'jpeg' mtype = 'jpeg'
if mType == 'mp3': if mtype == 'mp3':
mType = 'mpeg' mtype = 'mpeg'
fileExtension = mType file_extension = mtype
if not fileExtension: if not file_extension:
return post_json return post_json
mediaType = mediaType + '/' + fileExtension media_type = media_type + '/' + file_extension
print('Attached media type: ' + mediaType) print('Attached media type: ' + media_type)
if fileExtension == 'jpeg': if file_extension == 'jpeg':
fileExtension = 'jpg' file_extension = 'jpg'
if mediaType == 'audio/mpeg': if media_type == 'audio/mpeg':
fileExtension = 'mp3' file_extension = 'mp3'
domain = get_full_domain(domain, port) domain = get_full_domain(domain, port)
mPath = get_media_path() mpath = get_media_path()
mediaPath = mPath + '/' + create_password(32) + '.' + fileExtension media_path = mpath + '/' + create_password(32) + '.' + file_extension
if base_dir: if base_dir:
create_media_dirs(base_dir, mPath) create_media_dirs(base_dir, mpath)
mediaFilename = base_dir + '/' + mediaPath media_filename = base_dir + '/' + media_path
mediaPath = \ media_path = \
mediaPath.replace('media/', 'system/media_attachments/files/', 1) media_path.replace('media/', 'system/media_attachments/files/', 1)
attachmentJson = { attachment_json = {
'mediaType': mediaType, 'mediaType': media_type,
'name': description, 'name': description,
'type': 'Document', 'type': 'Document',
'url': http_prefix + '://' + domain + '/' + mediaPath 'url': http_prefix + '://' + domain + '/' + media_path
} }
if mediaType.startswith('image/'): if media_type.startswith('image/'):
attachmentJson['blurhash'] = _get_blur_hash() attachment_json['blurhash'] = _get_blur_hash()
# find the dimensions of the image and add them as metadata # find the dimensions of the image and add them as metadata
attachImageWidth, attachImageHeight = \ attach_image_width, attach_image_height = \
get_image_dimensions(image_filename) get_image_dimensions(image_filename)
if attachImageWidth and attachImageHeight: if attach_image_width and attach_image_height:
attachmentJson['width'] = attachImageWidth attachment_json['width'] = attach_image_width
attachmentJson['height'] = attachImageHeight attachment_json['height'] = attach_image_height
post_json['attachment'] = [attachmentJson] post_json['attachment'] = [attachment_json]
if base_dir: if base_dir:
if mediaType.startswith('image/'): if media_type.startswith('image/'):
if low_bandwidth: if low_bandwidth:
convert_image_to_low_bandwidth(image_filename) convert_image_to_low_bandwidth(image_filename)
process_meta_data(base_dir, nickname, domain, process_meta_data(base_dir, nickname, domain,
image_filename, mediaFilename, city, image_filename, media_filename, city,
content_license_url) content_license_url)
else: else:
copyfile(image_filename, mediaFilename) copyfile(image_filename, media_filename)
_update_etag(mediaFilename) _update_etag(media_filename)
return post_json return post_json
def archive_media(base_dir: str, archive_directory: str, def archive_media(base_dir: str, archive_directory: str,
maxWeeks: int) -> None: max_weeks: int) -> None:
"""Any media older than the given number of weeks gets archived """Any media older than the given number of weeks gets archived
""" """
if maxWeeks == 0: if max_weeks == 0:
return return
curr_time = datetime.datetime.utcnow() curr_time = datetime.datetime.utcnow()
weeksSinceEpoch = int((curr_time - datetime.datetime(1970, 1, 1)).days/7) weeks_since_epoch = int((curr_time - datetime.datetime(1970, 1, 1)).days/7)
minWeek = weeksSinceEpoch - maxWeeks min_week = weeks_since_epoch - max_weeks
if archive_directory: if archive_directory:
if not os.path.isdir(archive_directory): if not os.path.isdir(archive_directory):
@ -382,20 +387,22 @@ def archive_media(base_dir: str, archive_directory: str,
if not os.path.isdir(archive_directory + '/media'): if not os.path.isdir(archive_directory + '/media'):
os.mkdir(archive_directory + '/media') os.mkdir(archive_directory + '/media')
for subdir, dirs, files in os.walk(base_dir + '/media'): for _, dirs, _ in os.walk(base_dir + '/media'):
for weekDir in dirs: for week_dir in dirs:
if int(weekDir) < minWeek: if int(week_dir) < min_week:
if archive_directory: if archive_directory:
move(os.path.join(base_dir + '/media', weekDir), move(os.path.join(base_dir + '/media', week_dir),
archive_directory + '/media') archive_directory + '/media')
else: else:
# archive to /dev/null # archive to /dev/null
rmtree(os.path.join(base_dir + '/media', weekDir), rmtree(os.path.join(base_dir + '/media', week_dir),
ignore_errors=False, onerror=None) ignore_errors=False, onerror=None)
break break
def path_is_video(path: str) -> bool: def path_is_video(path: str) -> bool:
"""Is the given path a video file?
"""
if path.endswith('.ogv') or \ if path.endswith('.ogv') or \
path.endswith('.mp4'): path.endswith('.mp4'):
return True return True
@ -403,6 +410,8 @@ def path_is_video(path: str) -> bool:
def path_is_audio(path: str) -> bool: def path_is_audio(path: str) -> bool:
"""Is the given path an audio file?
"""
if path.endswith('.ogg') or \ if path.endswith('.ogg') or \
path.endswith('.mp3'): path.endswith('.mp3'):
return True return True
@ -420,13 +429,13 @@ def get_image_dimensions(image_filename: str) -> (int, int):
return None, None return None, None
if not result: if not result:
return None, None return None, None
dimensionsStr = result.stdout.decode('utf-8').replace('"', '') dimensions_str = result.stdout.decode('utf-8').replace('"', '')
if 'x' not in dimensionsStr: if 'x' not in dimensions_str:
return None, None return None, None
widthStr = dimensionsStr.split('x')[0] width_str = dimensions_str.split('x')[0]
if not widthStr.isdigit(): if not width_str.isdigit():
return None, None return None, None
heightStr = dimensionsStr.split('x')[1] height_str = dimensions_str.split('x')[1]
if not heightStr.isdigit(): if not height_str.isdigit():
return None, None return None, None
return int(widthStr), int(heightStr) return int(width_str), int(height_str)

158
metadata.py
View File

@ -17,23 +17,23 @@ from utils import no_of_active_accounts_monthly
def _get_status_count(base_dir: str) -> int: def _get_status_count(base_dir: str) -> int:
"""Get the total number of posts """Get the total number of posts
""" """
statusCtr = 0 status_ctr = 0
accountsDir = base_dir + '/accounts' accounts_dir = base_dir + '/accounts'
for subdir, dirs, files in os.walk(accountsDir): for _, dirs, _ in os.walk(accounts_dir):
for acct in dirs: for acct in dirs:
if not is_account_dir(acct): if not is_account_dir(acct):
continue continue
acct_dir = os.path.join(accountsDir, acct + '/outbox') acct_dir = os.path.join(accounts_dir, acct + '/outbox')
for subdir2, dirs2, files2 in os.walk(acct_dir): for _, _, files2 in os.walk(acct_dir):
statusCtr += len(files2) status_ctr += len(files2)
break break
break break
return statusCtr return status_ctr
def meta_data_node_info(base_dir: str, def meta_data_node_info(base_dir: str,
aboutUrl: str, about_url: str,
termsOfServiceUrl: str, terms_of_service_url: str,
registration: bool, version: str, registration: bool, version: str,
showAccounts: bool) -> {}: showAccounts: bool) -> {}:
""" /nodeinfo/2.0 endpoint """ /nodeinfo/2.0 endpoint
@ -47,15 +47,15 @@ def meta_data_node_info(base_dir: str,
sensitive sensitive
""" """
if showAccounts: if showAccounts:
activeAccounts = no_of_accounts(base_dir) active_accounts = no_of_accounts(base_dir)
activeAccountsMonthly = no_of_active_accounts_monthly(base_dir, 1) active_accounts_monthly = no_of_active_accounts_monthly(base_dir, 1)
activeAccountsHalfYear = no_of_active_accounts_monthly(base_dir, 6) active_accounts_half_year = no_of_active_accounts_monthly(base_dir, 6)
localPosts = _get_status_count(base_dir) local_posts = _get_status_count(base_dir)
else: else:
activeAccounts = 1 active_accounts = 1
activeAccountsMonthly = 1 active_accounts_monthly = 1
activeAccountsHalfYear = 1 active_accounts_half_year = 1
localPosts = 1 local_posts = 1
nodeinfo = { nodeinfo = {
'openRegistrations': registration, 'openRegistrations': registration,
@ -65,15 +65,15 @@ def meta_data_node_info(base_dir: str,
'version': version 'version': version
}, },
'documents': { 'documents': {
'about': aboutUrl, 'about': about_url,
'terms': termsOfServiceUrl 'terms': terms_of_service_url
}, },
'usage': { 'usage': {
'localPosts': localPosts, 'localPosts': local_posts,
'users': { 'users': {
'activeHalfyear': activeAccountsHalfYear, 'activeHalfyear': active_accounts_half_year,
'activeMonth': activeAccountsMonthly, 'activeMonth': active_accounts_monthly,
'total': activeAccounts 'total': active_accounts
} }
}, },
'version': '2.0' 'version': '2.0'
@ -82,100 +82,100 @@ def meta_data_node_info(base_dir: str,
def meta_data_instance(showAccounts: bool, def meta_data_instance(showAccounts: bool,
instanceTitle: str, instance_title: str,
instanceDescriptionShort: str, instance_description_short: str,
instanceDescription: str, instance_description: str,
http_prefix: str, base_dir: str, http_prefix: str, base_dir: str,
admin_nickname: str, domain: str, domain_full: str, admin_nickname: str, domain: str, domain_full: str,
registration: bool, system_language: str, registration: bool, system_language: str,
version: str) -> {}: version: str) -> {}:
""" /api/v1/instance endpoint """ /api/v1/instance endpoint
""" """
adminActorFilename = \ admin_actor_filename = \
base_dir + '/accounts/' + admin_nickname + '@' + domain + '.json' base_dir + '/accounts/' + admin_nickname + '@' + domain + '.json'
if not os.path.isfile(adminActorFilename): if not os.path.isfile(admin_actor_filename):
return {} return {}
adminActor = load_json(adminActorFilename, 0) admin_actor = load_json(admin_actor_filename, 0)
if not adminActor: if not admin_actor:
print('WARN: json load exception meta_data_instance') print('WARN: json load exception meta_data_instance')
return {} return {}
rulesList = [] rules_list = []
rulesFilename = \ rules_filename = \
base_dir + '/accounts/tos.md' base_dir + '/accounts/tos.md'
if os.path.isfile(rulesFilename): if os.path.isfile(rules_filename):
with open(rulesFilename, 'r') as fp: with open(rules_filename, 'r') as fp_rules:
rulesLines = fp.readlines() rules_lines = fp_rules.readlines()
ruleCtr = 1 rule_ctr = 1
for line in rulesLines: for line in rules_lines:
line = line.strip() line = line.strip()
if not line: if not line:
continue continue
if line.startswith('#'): if line.startswith('#'):
continue continue
rulesList.append({ rules_list.append({
'id': str(ruleCtr), 'id': str(rule_ctr),
'text': line 'text': line
}) })
ruleCtr += 1 rule_ctr += 1
isBot = False is_bot = False
isGroup = False is_group = False
if adminActor['type'] == 'Group': if admin_actor['type'] == 'Group':
isGroup = True is_group = True
elif adminActor['type'] != 'Person': elif admin_actor['type'] != 'Person':
isBot = True is_bot = True
url = \ url = \
http_prefix + '://' + domain_full + '/@' + \ http_prefix + '://' + domain_full + '/@' + \
adminActor['preferredUsername'] admin_actor['preferredUsername']
if showAccounts: if showAccounts:
activeAccounts = no_of_accounts(base_dir) active_accounts = no_of_accounts(base_dir)
localPosts = _get_status_count(base_dir) local_posts = _get_status_count(base_dir)
else: else:
activeAccounts = 1 active_accounts = 1
localPosts = 1 local_posts = 1
createdAt = '' created_at = ''
if adminActor.get('published'): if admin_actor.get('published'):
createdAt = adminActor['published'] created_at = admin_actor['published']
instance = { instance = {
'approval_required': False, 'approval_required': False,
'invites_enabled': False, 'invites_enabled': False,
'registrations': registration, 'registrations': registration,
'contact_account': { 'contact_account': {
'acct': adminActor['preferredUsername'], 'acct': admin_actor['preferredUsername'],
'created_at': createdAt, 'created_at': created_at,
'avatar': adminActor['icon']['url'], 'avatar': admin_actor['icon']['url'],
'avatar_static': adminActor['icon']['url'], 'avatar_static': admin_actor['icon']['url'],
'header': adminActor['image']['url'], 'header': admin_actor['image']['url'],
'header_static': adminActor['image']['url'], 'header_static': admin_actor['image']['url'],
'bot': isBot, 'bot': is_bot,
'discoverable': True, 'discoverable': True,
'group': isGroup, 'group': is_group,
'display_name': adminActor['name'], 'display_name': admin_actor['name'],
'locked': adminActor['manuallyApprovesFollowers'], 'locked': admin_actor['manuallyApprovesFollowers'],
'note': '<p>Admin of ' + domain + '</p>', 'note': '<p>Admin of ' + domain + '</p>',
'url': url, 'url': url,
'username': adminActor['preferredUsername'] 'username': admin_actor['preferredUsername']
}, },
'description': instanceDescription, 'description': instance_description,
'languages': [system_language], 'languages': [system_language],
'short_description': instanceDescriptionShort, 'short_description': instance_description_short,
'stats': { 'stats': {
'domain_count': 2, 'domain_count': 2,
'status_count': localPosts, 'status_count': local_posts,
'user_count': activeAccounts 'user_count': active_accounts
}, },
'thumbnail': http_prefix + '://' + domain_full + '/login.png', 'thumbnail': http_prefix + '://' + domain_full + '/login.png',
'title': instanceTitle, 'title': instance_title,
'uri': domain_full, 'uri': domain_full,
'urls': {}, 'urls': {},
'version': version, 'version': version,
'rules': rulesList, 'rules': rules_list,
'configuration': { 'configuration': {
'statuses': { 'statuses': {
'max_media_attachments': 1 'max_media_attachments': 1
@ -214,18 +214,18 @@ def metadata_custom_emoji(base_dir: str,
See https://docs.joinmastodon.org/methods/instance/custom_emojis See https://docs.joinmastodon.org/methods/instance/custom_emojis
""" """
result = [] result = []
emojisUrl = http_prefix + '://' + domain_full + '/emoji' emojis_url = http_prefix + '://' + domain_full + '/emoji'
for subdir, dirs, files in os.walk(base_dir + '/emoji'): for _, _, files in os.walk(base_dir + '/emoji'):
for f in files: for fname in files:
if len(f) < 3: if len(fname) < 3:
continue continue
if f[0].isdigit() or f[1].isdigit(): if fname[0].isdigit() or fname[1].isdigit():
continue continue
if not f.endswith('.png'): if not fname.endswith('.png'):
continue continue
url = os.path.join(emojisUrl, f) url = os.path.join(emojis_url, fname)
result.append({ result.append({
"shortcode": f.replace('.png', ''), "shortcode": fname.replace('.png', ''),
"url": url, "url": url,
"static_url": url, "static_url": url,
"visible_in_picker": True "visible_in_picker": True

20
pgp.py
View File

@ -108,7 +108,7 @@ def set_email_address(actor_json: {}, email_address: str) -> None:
actor_json['attachment'] = [] actor_json['attachment'] = []
# remove any existing value # remove any existing value
propertyFound = None property_found = None
for property_value in actor_json['attachment']: for property_value in actor_json['attachment']:
if not property_value.get('name'): if not property_value.get('name'):
continue continue
@ -116,10 +116,10 @@ def set_email_address(actor_json: {}, email_address: str) -> None:
continue continue
if not property_value['name'].lower().startswith('email'): if not property_value['name'].lower().startswith('email'):
continue continue
propertyFound = property_value property_found = property_value
break break
if propertyFound: if property_found:
actor_json['attachment'].remove(propertyFound) actor_json['attachment'].remove(property_found)
if notEmailAddress: if notEmailAddress:
return return
@ -159,7 +159,7 @@ def set_pgp_pub_key(actor_json: {}, pgp_pub_key: str) -> None:
actor_json['attachment'] = [] actor_json['attachment'] = []
# remove any existing value # remove any existing value
propertyFound = None property_found = None
for property_value in actor_json['attachment']: for property_value in actor_json['attachment']:
if not property_value.get('name'): if not property_value.get('name'):
continue continue
@ -167,9 +167,9 @@ def set_pgp_pub_key(actor_json: {}, pgp_pub_key: str) -> None:
continue continue
if not property_value['name'].lower().startswith('pgp'): if not property_value['name'].lower().startswith('pgp'):
continue continue
propertyFound = property_value property_found = property_value
break break
if propertyFound: if property_found:
actor_json['attachment'].remove(property_value) actor_json['attachment'].remove(property_value)
if removeKey: if removeKey:
return return
@ -208,7 +208,7 @@ def set_pgp_fingerprint(actor_json: {}, fingerprint: str) -> None:
actor_json['attachment'] = [] actor_json['attachment'] = []
# remove any existing value # remove any existing value
propertyFound = None property_found = None
for property_value in actor_json['attachment']: for property_value in actor_json['attachment']:
if not property_value.get('name'): if not property_value.get('name'):
continue continue
@ -216,9 +216,9 @@ def set_pgp_fingerprint(actor_json: {}, fingerprint: str) -> None:
continue continue
if not property_value['name'].lower().startswith('openpgp'): if not property_value['name'].lower().startswith('openpgp'):
continue continue
propertyFound = property_value property_found = property_value
break break
if propertyFound: if property_found:
actor_json['attachment'].remove(property_value) actor_json['attachment'].remove(property_value)
if removeFingerprint: if removeFingerprint:
return return

8
ssb.py
View File

@ -60,7 +60,7 @@ def set_ssb_address(actor_json: {}, ssb_address: str) -> None:
actor_json['attachment'] = [] actor_json['attachment'] = []
# remove any existing value # remove any existing value
propertyFound = None property_found = None
for property_value in actor_json['attachment']: for property_value in actor_json['attachment']:
if not property_value.get('name'): if not property_value.get('name'):
continue continue
@ -68,10 +68,10 @@ def set_ssb_address(actor_json: {}, ssb_address: str) -> None:
continue continue
if not property_value['name'].lower().startswith('ssb'): if not property_value['name'].lower().startswith('ssb'):
continue continue
propertyFound = property_value property_found = property_value
break break
if propertyFound: if property_found:
actor_json['attachment'].remove(propertyFound) actor_json['attachment'].remove(property_found)
if notSSBAddress: if notSSBAddress:
return return

6
tests.py
View File

@ -5432,13 +5432,13 @@ def test_update_actor(base_dir: str):
if len(actor_json['attachment']) == 0: if len(actor_json['attachment']) == 0:
print("actor_json['attachment'] has no contents") print("actor_json['attachment'] has no contents")
assert len(actor_json['attachment']) > 0 assert len(actor_json['attachment']) > 0
propertyFound = False property_found = False
for property_value in actor_json['attachment']: for property_value in actor_json['attachment']:
if property_value['name'] == 'PGP': if property_value['name'] == 'PGP':
print('PGP property set within attachment') print('PGP property set within attachment')
assert pubKey in property_value['value'] assert pubKey in property_value['value']
propertyFound = True property_found = True
assert propertyFound assert property_found
# stop the server # stop the server
thrAlice.kill() thrAlice.kill()

8
tox.py
View File

@ -65,7 +65,7 @@ def set_tox_address(actor_json: {}, tox_address: str) -> None:
actor_json['attachment'] = [] actor_json['attachment'] = []
# remove any existing value # remove any existing value
propertyFound = None property_found = None
for property_value in actor_json['attachment']: for property_value in actor_json['attachment']:
if not property_value.get('name'): if not property_value.get('name'):
continue continue
@ -73,10 +73,10 @@ def set_tox_address(actor_json: {}, tox_address: str) -> None:
continue continue
if not property_value['name'].lower().startswith('tox'): if not property_value['name'].lower().startswith('tox'):
continue continue
propertyFound = property_value property_found = property_value
break break
if propertyFound: if property_found:
actor_json['attachment'].remove(propertyFound) actor_json['attachment'].remove(property_found)
if notToxAddress: if notToxAddress:
return return

8
webapp_utils.py
View File

@ -181,7 +181,7 @@ def _set_actor_property_url(actor_json: {},
property_nameLower = property_name.lower() property_nameLower = property_name.lower()
# remove any existing value # remove any existing value
propertyFound = None property_found = None
for property_value in actor_json['attachment']: for property_value in actor_json['attachment']:
if not property_value.get('name'): if not property_value.get('name'):
continue continue
@ -189,10 +189,10 @@ def _set_actor_property_url(actor_json: {},
continue continue
if not property_value['name'].lower().startswith(property_nameLower): if not property_value['name'].lower().startswith(property_nameLower):
continue continue
propertyFound = property_value property_found = property_value
break break
if propertyFound: if property_found:
actor_json['attachment'].remove(propertyFound) actor_json['attachment'].remove(property_found)
prefixes = get_protocol_prefixes() prefixes = get_protocol_prefixes()
prefixFound = False prefixFound = False

8
xmpp.py
View File

@ -51,7 +51,7 @@ def set_xmpp_address(actor_json: {}, xmpp_address: str) -> None:
actor_json['attachment'] = [] actor_json['attachment'] = []
# remove any existing value # remove any existing value
propertyFound = None property_found = None
for property_value in actor_json['attachment']: for property_value in actor_json['attachment']:
if not property_value.get('name'): if not property_value.get('name'):
continue continue
@ -60,10 +60,10 @@ def set_xmpp_address(actor_json: {}, xmpp_address: str) -> None:
if not (property_value['name'].lower().startswith('xmpp') or if not (property_value['name'].lower().startswith('xmpp') or
property_value['name'].lower().startswith('jabber')): property_value['name'].lower().startswith('jabber')):
continue continue
propertyFound = property_value property_found = property_value
break break
if propertyFound: if property_found:
actor_json['attachment'].remove(propertyFound) actor_json['attachment'].remove(property_found)
if notXmppAddress: if notXmppAddress:
return return