'
if markdown.endswith('**'):
markdown = markdown[:len(markdown) - 2] + '
'
elif markdown.endswith('*'):
markdown = markdown[:len(markdown) - 1] + ''
elif markdown.endswith('_'):
markdown = markdown[:len(markdown) - 1] + ''
return markdown
def _markdownReplaceQuotes(markdown: str) -> str:
"""Replaces > quotes with html blockquote
"""
if '> ' not in markdown:
return markdown
lines = markdown.split('\n')
result = ''
prevQuoteLine = None
for line in lines:
if '> ' not in line:
result += line + '\n'
prevQuoteLine = None
continue
lineStr = line.strip()
if not lineStr.startswith('> '):
result += line + '\n'
prevQuoteLine = None
continue
lineStr = lineStr.replace('> ', '', 1).strip()
if prevQuoteLine:
newPrevLine = prevQuoteLine.replace('\n', '')
result = result.replace(prevQuoteLine, newPrevLine) + ' '
lineStr += '\n'
else:
lineStr = '
' + lineStr + '
\n'
result += lineStr
prevQuoteLine = lineStr
if '\n' in result:
result = result.replace('\n', '')
if result.endswith('\n') and \
not markdown.endswith('\n'):
result = result[:len(result) - 1]
return result
def _markdownReplaceLinks(markdown: str, images=False) -> str:
"""Replaces markdown links with html
Optionally replace image links
"""
replaceLinks = {}
text = markdown
startChars = '['
if images:
startChars = '!['
while startChars in text:
if ')' not in text:
break
text = text.split(startChars, 1)[1]
markdownLink = startChars + text.split(')')[0] + ')'
if ']' not in markdownLink or \
'(' not in markdownLink:
text = text.split(')', 1)[1]
continue
if not images:
replaceLinks[markdownLink] = \
'' + \
markdownLink.split(startChars)[1].split(']')[0] + \
''
else:
replaceLinks[markdownLink] = \
''
text = text.split(')', 1)[1]
for mdLink, htmlLink in replaceLinks.items():
markdown = markdown.replace(mdLink, htmlLink)
return markdown
def markdownToHtml(markdown: str) -> str:
"""Converts markdown formatted text to html
"""
markdown = _markdownReplaceQuotes(markdown)
markdown = _markdownEmphasisHtml(markdown)
markdown = _markdownReplaceLinks(markdown, True)
markdown = _markdownReplaceLinks(markdown)
# replace headers
linesList = markdown.split('\n')
htmlStr = ''
ctr = 0
for line in linesList:
if ctr > 0:
htmlStr += ' '
if line.startswith('#####'):
line = line.replace('#####', '').strip()
line = '
' + line + '
'
ctr = -1
elif line.startswith('####'):
line = line.replace('####', '').strip()
line = '
' + line + '
'
ctr = -1
elif line.startswith('###'):
line = line.replace('###', '').strip()
line = '
' + line + '
'
ctr = -1
elif line.startswith('##'):
line = line.replace('##', '').strip()
line = '
' + line + '
'
ctr = -1
elif line.startswith('#'):
line = line.replace('#', '').strip()
line = '
' + line + '
'
ctr = -1
htmlStr += line
ctr += 1
return htmlStr
def getBrokenLinkSubstitute() -> str:
"""Returns html used to show a default image if the link to
an image is broken
"""
return " onerror=\"this.onerror=null; this.src='" + \
"/icons/avatar_default.png'\""
def htmlFollowingList(cssCache: {}, baseDir: str,
followingFilename: str) -> str:
"""Returns a list of handles being followed
"""
with open(followingFilename, 'r') as followingFile:
msg = followingFile.read()
followingList = msg.split('\n')
followingList.sort()
if followingList:
cssFilename = baseDir + '/epicyon-profile.css'
if os.path.isfile(baseDir + '/epicyon.css'):
cssFilename = baseDir + '/epicyon.css'
instanceTitle = \
getConfigParam(baseDir, 'instanceTitle')
followingListHtml = htmlHeaderWithExternalStyle(cssFilename,
instanceTitle)
for followingAddress in followingList:
if followingAddress:
followingListHtml += \
'
\n'
return headerStr
def getAltPath(actor: str, domainFull: str, callingDomain: str) -> str:
"""Returns alternate path from the actor
eg. https://clearnetdomain/path becomes http://oniondomain/path
"""
postActor = actor
if callingDomain not in actor and domainFull in actor:
if callingDomain.endswith('.onion') or \
callingDomain.endswith('.i2p'):
postActor = \
'http://' + callingDomain + actor.split(domainFull)[1]
print('Changed POST domain from ' + actor + ' to ' + postActor)
return postActor
def getContentWarningButton(postID: str, translate: {},
content: str) -> str:
"""Returns the markup for a content warning button
"""
return ' ' + \
translate['SHOW MORE'] + '' + \
'
' + content + \
'
\n'
def _getActorPropertyUrl(actorJson: {}, propertyName: str) -> str:
"""Returns a url property from an actor
"""
if not actorJson.get('attachment'):
return ''
propertyName = propertyName.lower()
for propertyValue in actorJson['attachment']:
if not propertyValue.get('name'):
continue
if not propertyValue['name'].lower().startswith(propertyName):
continue
if not propertyValue.get('type'):
continue
if not propertyValue.get('value'):
continue
if propertyValue['type'] != 'PropertyValue':
continue
propertyValue['value'] = propertyValue['value'].strip()
prefixes = getProtocolPrefixes()
prefixFound = False
for prefix in prefixes:
if propertyValue['value'].startswith(prefix):
prefixFound = True
break
if not prefixFound:
continue
if '.' not in propertyValue['value']:
continue
if ' ' in propertyValue['value']:
continue
if ',' in propertyValue['value']:
continue
return propertyValue['value']
return ''
def getBlogAddress(actorJson: {}) -> str:
"""Returns blog address for the given actor
"""
return _getActorPropertyUrl(actorJson, 'Blog')
def _setActorPropertyUrl(actorJson: {}, propertyName: str, url: str) -> None:
"""Sets a url for the given actor property
"""
if not actorJson.get('attachment'):
actorJson['attachment'] = []
propertyNameLower = propertyName.lower()
# remove any existing value
propertyFound = None
for propertyValue in actorJson['attachment']:
if not propertyValue.get('name'):
continue
if not propertyValue.get('type'):
continue
if not propertyValue['name'].lower().startswith(propertyNameLower):
continue
propertyFound = propertyValue
break
if propertyFound:
actorJson['attachment'].remove(propertyFound)
prefixes = getProtocolPrefixes()
prefixFound = False
for prefix in prefixes:
if url.startswith(prefix):
prefixFound = True
break
if not prefixFound:
return
if '.' not in url:
return
if ' ' in url:
return
if ',' in url:
return
for propertyValue in actorJson['attachment']:
if not propertyValue.get('name'):
continue
if not propertyValue.get('type'):
continue
if not propertyValue['name'].lower().startswith(propertyNameLower):
continue
if propertyValue['type'] != 'PropertyValue':
continue
propertyValue['value'] = url
return
newAddress = {
"name": propertyName,
"type": "PropertyValue",
"value": url
}
actorJson['attachment'].append(newAddress)
def setBlogAddress(actorJson: {}, blogAddress: str) -> None:
"""Sets an blog address for the given actor
"""
_setActorPropertyUrl(actorJson, 'Blog', removeHtml(blogAddress))
def updateAvatarImageCache(session, baseDir: str, httpPrefix: str,
actor: str, avatarUrl: str,
personCache: {}, allowDownloads: bool,
force=False, debug=False) -> str:
"""Updates the cached avatar for the given actor
"""
if not avatarUrl:
return None
actorStr = actor.replace('/', '-')
avatarImagePath = baseDir + '/cache/avatars/' + actorStr
# try different image types
imageFormats = {
'png': 'png',
'jpg': 'jpeg',
'jpeg': 'jpeg',
'gif': 'gif',
'svg': 'svg+xml',
'webp': 'webp',
'avif': 'avif'
}
avatarImageFilename = None
for imFormat, mimeType in imageFormats.items():
if avatarUrl.endswith('.' + imFormat) or \
'.' + imFormat + '?' in avatarUrl:
sessionHeaders = {
'Accept': 'image/' + mimeType
}
avatarImageFilename = avatarImagePath + '.' + imFormat
if not avatarImageFilename:
return None
if (not os.path.isfile(avatarImageFilename) or force) and allowDownloads:
try:
if debug:
print('avatar image url: ' + avatarUrl)
result = session.get(avatarUrl,
headers=sessionHeaders,
params=None)
if result.status_code < 200 or \
result.status_code > 202:
if debug:
print('Avatar image download failed with status ' +
str(result.status_code))
# remove partial download
if os.path.isfile(avatarImageFilename):
os.remove(avatarImageFilename)
else:
with open(avatarImageFilename, 'wb') as f:
f.write(result.content)
if debug:
print('avatar image downloaded for ' + actor)
return avatarImageFilename.replace(baseDir + '/cache', '')
except Exception as e:
if debug:
print('Failed to download avatar image: ' + str(avatarUrl))
print(e)
prof = 'https://www.w3.org/ns/activitystreams'
if '/channel/' not in actor or '/accounts/' not in actor:
sessionHeaders = {
'Accept': 'application/activity+json; profile="' + prof + '"'
}
else:
sessionHeaders = {
'Accept': 'application/ld+json; profile="' + prof + '"'
}
personJson = \
getJson(session, actor, sessionHeaders, None,
debug, __version__, httpPrefix, None)
if personJson:
if not personJson.get('id'):
return None
if not personJson.get('publicKey'):
return None
if not personJson['publicKey'].get('publicKeyPem'):
return None
if personJson['id'] != actor:
return None
if not personCache.get(actor):
return None
if personCache[actor]['actor']['publicKey']['publicKeyPem'] != \
personJson['publicKey']['publicKeyPem']:
print("ERROR: " +
"public keys don't match when downloading actor for " +
actor)
return None
storePersonInCache(baseDir, actor, personJson, personCache,
allowDownloads)
return getPersonAvatarUrl(baseDir, actor, personCache,
allowDownloads)
return None
return avatarImageFilename.replace(baseDir + '/cache', '')
def getPersonAvatarUrl(baseDir: str, personUrl: str, personCache: {},
allowDownloads: bool) -> str:
"""Returns the avatar url for the person
"""
personJson = \
getPersonFromCache(baseDir, personUrl, personCache, allowDownloads)
if not personJson:
return None
# get from locally stored image
if not personJson.get('id'):
return None
actorStr = personJson['id'].replace('/', '-')
avatarImagePath = baseDir + '/cache/avatars/' + actorStr
imageExtension = getImageExtensions()
for ext in imageExtension:
if os.path.isfile(avatarImagePath + '.' + ext):
return '/avatars/' + actorStr + '.' + ext
elif os.path.isfile(avatarImagePath.lower() + '.' + ext):
return '/avatars/' + actorStr.lower() + '.' + ext
if personJson.get('icon'):
if personJson['icon'].get('url'):
return personJson['icon']['url']
return None
def scheduledPostsExist(baseDir: str, nickname: str, domain: str) -> bool:
"""Returns true if there are posts scheduled to be delivered
"""
scheduleIndexFilename = \
baseDir + '/accounts/' + nickname + '@' + domain + '/schedule.index'
if not os.path.isfile(scheduleIndexFilename):
return False
if '#users#' in open(scheduleIndexFilename).read():
return True
return False
def sharesTimelineJson(actor: str, pageNumber: int, itemsPerPage: int,
baseDir: str, maxSharesPerAccount: int) -> ({}, bool):
"""Get a page on the shared items timeline as json
maxSharesPerAccount helps to avoid one person dominating the timeline
by sharing a large number of things
"""
allSharesJson = {}
for subdir, dirs, files in os.walk(baseDir + '/accounts'):
for handle in dirs:
if '@' not in handle:
continue
accountDir = baseDir + '/accounts/' + handle
sharesFilename = accountDir + '/shares.json'
if not os.path.isfile(sharesFilename):
continue
sharesJson = loadJson(sharesFilename)
if not sharesJson:
continue
nickname = handle.split('@')[0]
# actor who owns this share
owner = actor.split('/users/')[0] + '/users/' + nickname
ctr = 0
for itemID, item in sharesJson.items():
# assign owner to the item
item['actor'] = owner
allSharesJson[str(item['published'])] = item
ctr += 1
if ctr >= maxSharesPerAccount:
break
break
# sort the shared items in descending order of publication date
sharesJson = OrderedDict(sorted(allSharesJson.items(), reverse=True))
lastPage = False
startIndex = itemsPerPage * pageNumber
maxIndex = len(sharesJson.items())
if maxIndex < itemsPerPage:
lastPage = True
if startIndex >= maxIndex - itemsPerPage:
lastPage = True
startIndex = maxIndex - itemsPerPage
if startIndex < 0:
startIndex = 0
ctr = 0
resultJson = {}
for published, item in sharesJson.items():
if ctr >= startIndex + itemsPerPage:
break
if ctr < startIndex:
ctr += 1
continue
resultJson[published] = item
ctr += 1
return resultJson, lastPage
def postContainsPublic(postJsonObject: {}) -> bool:
"""Does the given post contain #Public
"""
containsPublic = False
if not postJsonObject['object'].get('to'):
return containsPublic
for toAddress in postJsonObject['object']['to']:
if toAddress.endswith('#Public'):
containsPublic = True
break
if not containsPublic:
if postJsonObject['object'].get('cc'):
for toAddress in postJsonObject['object']['cc']:
if toAddress.endswith('#Public'):
containsPublic = True
break
return containsPublic
def _getImageFile(baseDir: str, name: str, directory: str,
nickname: str, domain: str, theme: str) -> (str, str):
"""
returns the filenames for an image with the given name
"""
bannerExtensions = getImageExtensions()
bannerFile = ''
bannerFilename = ''
for ext in bannerExtensions:
bannerFileTest = name + '.' + ext
bannerFilenameTest = directory + '/' + bannerFileTest
if os.path.isfile(bannerFilenameTest):
bannerFile = name + '_' + theme + '.' + ext
bannerFilename = bannerFilenameTest
break
return bannerFile, bannerFilename
def getBannerFile(baseDir: str,
nickname: str, domain: str, theme: str) -> (str, str):
return _getImageFile(baseDir, 'banner',
baseDir + '/accounts/' + nickname + '@' + domain,
nickname, domain, theme)
def getSearchBannerFile(baseDir: str,
nickname: str, domain: str, theme: str) -> (str, str):
return _getImageFile(baseDir, 'search_banner',
baseDir + '/accounts/' + nickname + '@' + domain,
nickname, domain, theme)
def getLeftImageFile(baseDir: str,
nickname: str, domain: str, theme: str) -> (str, str):
return _getImageFile(baseDir, 'left_col_image',
baseDir + '/accounts/' + nickname + '@' + domain,
nickname, domain, theme)
def getRightImageFile(baseDir: str,
nickname: str, domain: str, theme: str) -> (str, str):
return _getImageFile(baseDir, 'right_col_image',
baseDir + '/accounts/' + nickname + '@' + domain,
nickname, domain, theme)
def htmlHeaderWithExternalStyle(cssFilename: str, instanceTitle: str,
lang='en') -> str:
htmlStr = '\n'
htmlStr += '\n'
htmlStr += ' \n'
htmlStr += ' \n'
cssFile = '/' + cssFilename.split('/')[-1]
htmlStr += ' \n'
htmlStr += ' \n'
htmlStr += ' \n'
htmlStr += ' ' + instanceTitle + '\n'
htmlStr += ' \n'
htmlStr += ' \n'
return htmlStr
def htmlHeaderWithPersonMarkup(cssFilename: str, instanceTitle: str,
actorJson: {}, city: str,
lang='en') -> str:
"""html header which includes person markup
https://schema.org/Person
"""
htmlStr = htmlHeaderWithExternalStyle(cssFilename, instanceTitle, lang)
if not actorJson:
return htmlStr
cityMarkup = ''
if city:
city = city.lower().title()
addComma = ''
countryMarkup = ''
if ',' in city:
country = city.split(',', 1)[1].strip().title()
city = city.split(',', 1)[0]
countryMarkup = \
' "addressCountry": "' + country + '"\n'
addComma = ','
cityMarkup = \
' "address": {\n' + \
' "@type": "PostalAddress",\n' + \
' "addressLocality": "' + city + '"' + addComma + '\n' + \
countryMarkup + \
' },\n'
skillsMarkup = ''
if actorJson.get('hasOccupation'):
if isinstance(actorJson['hasOccupation'], list):
skillsMarkup = ' "hasOccupation": [\n'
firstEntry = True
for skillDict in actorJson['hasOccupation']:
if skillDict['@type'] == 'Role':
if not firstEntry:
skillsMarkup += ',\n'
sk = skillDict['hasOccupation']
roleName = sk['name']
if not roleName:
roleName = 'member'
category = \
sk['occupationalCategory']['codeValue']
categoryUrl = \
'https://www.onetonline.org/link/summary/' + category
skillsMarkup += ' {\n'
skillsMarkup += ' "@type": "Role",\n'
skillsMarkup += ' "hasOccupation": {\n'
skillsMarkup += ' "@type": "Occupation",\n'
skillsMarkup += ' "name": "' + roleName + '",\n'
skillsMarkup += ' "description": ' + \
'"Fediverse instance role",\n'
skillsMarkup += ' "occupationLocation": {\n'
skillsMarkup += \
' "@type": "City",\n'
skillsMarkup += \
' "name": "' + city + '"\n'
skillsMarkup += ' },\n'
skillsMarkup += ' "occupationalCategory": {\n'
skillsMarkup += ' "@type": "CategoryCode",\n'
skillsMarkup += ' "inCodeSet": {\n'
skillsMarkup += \
' "@type": "CategoryCodeSet",\n'
skillsMarkup += ' "name": "O*Net-SOC",\n'
skillsMarkup += ' "dateModified": "2019",\n'
skillsMarkup += \
' ' + \
'"url": "https://www.onetonline.org/"\n'
skillsMarkup += ' },\n'
skillsMarkup += \
' "codeValue": "' + category + '",\n'
skillsMarkup += \
' "url": "' + categoryUrl + '"\n'
skillsMarkup += ' }\n'
skillsMarkup += ' }\n'
skillsMarkup += ' }'
elif skillDict['@type'] == 'Occupation':
if not firstEntry:
skillsMarkup += ',\n'
ocName = skillDict['name']
if not ocName:
ocName = 'member'
skillsList = skillDict['skills']
skillsListStr = '['
for skillStr in skillsList:
if skillsListStr != '[':
skillsListStr += ', '
skillsListStr += '"' + skillStr + '"'
skillsListStr += ']'
skillsMarkup += ' {\n'
skillsMarkup += ' "@type": "Occupation",\n'
skillsMarkup += ' "name": "' + ocName + '",\n'
skillsMarkup += ' "description": ' + \
'"Fediverse instance occupation",\n'
skillsMarkup += ' "occupationLocation": {\n'
skillsMarkup += ' "@type": "City",\n'
skillsMarkup += \
' "name": "' + city + '"\n'
skillsMarkup += ' },\n'
skillsMarkup += \
' "skills": ' + skillsListStr + '\n'
skillsMarkup += ' }'
firstEntry = False
skillsMarkup += '\n ],\n'
description = removeHtml(actorJson['summary'])
nameStr = removeHtml(actorJson['name'])
personMarkup = \
' \n'
htmlStr = htmlStr.replace('\n', '\n' + personMarkup)
return htmlStr
def htmlHeaderWithWebsiteMarkup(cssFilename: str, instanceTitle: str,
httpPrefix: str, domain: str,
systemLanguage: str) -> str:
"""html header which includes website markup
https://schema.org/WebSite
"""
htmlStr = htmlHeaderWithExternalStyle(cssFilename, instanceTitle,
systemLanguage)
licenseUrl = 'https://www.gnu.org/licenses/agpl-3.0.rdf'
# social networking category
genreUrl = 'http://vocab.getty.edu/aat/300312270'
websiteMarkup = \
' \n'
htmlStr = htmlStr.replace('\n', '\n' + websiteMarkup)
return htmlStr
def htmlHeaderWithBlogMarkup(cssFilename: str, instanceTitle: str,
httpPrefix: str, domain: str, nickname: str,
systemLanguage: str, published: str,
title: str, snippet: str) -> str:
"""html header which includes blog post markup
https://schema.org/BlogPosting
"""
htmlStr = htmlHeaderWithExternalStyle(cssFilename, instanceTitle,
systemLanguage)
authorUrl = httpPrefix + '://' + domain + '/users/' + nickname
aboutUrl = httpPrefix + '://' + domain + '/about.html'
# license for content on the site may be different from
# the software license
contentLicenseUrl = 'https://creativecommons.org/licenses/by/3.0'
blogMarkup = \
' \n'
htmlStr = htmlStr.replace('\n', '\n' + blogMarkup)
return htmlStr
def htmlFooter() -> str:
htmlStr = ' \n'
htmlStr += '\n'
return htmlStr
def loadIndividualPostAsHtmlFromCache(baseDir: str,
nickname: str, domain: str,
postJsonObject: {}) -> str:
"""If a cached html version of the given post exists then load it and
return the html text
This is much quicker than generating the html from the json object
"""
cachedPostFilename = \
getCachedPostFilename(baseDir, nickname, domain, postJsonObject)
postHtml = ''
if not cachedPostFilename:
return postHtml
if not os.path.isfile(cachedPostFilename):
return postHtml
tries = 0
while tries < 3:
try:
with open(cachedPostFilename, 'r') as file:
postHtml = file.read()
break
except Exception as e:
print(e)
# no sleep
tries += 1
if postHtml:
return postHtml
def addEmojiToDisplayName(baseDir: str, httpPrefix: str,
nickname: str, domain: str,
displayName: str, inProfileName: bool) -> str:
"""Adds emoji icons to display names or CW on individual posts
"""
if ':' not in displayName:
return displayName
displayName = displayName.replace('
\n'
attachmentCtr += 1
if mediaStyleAdded:
attachmentStr += '
'
return attachmentStr, galleryStr
def htmlPostSeparator(baseDir: str, column: str) -> str:
"""Returns the html for a timeline post separator image
"""
theme = getConfigParam(baseDir, 'theme')
filename = 'separator.png'
separatorClass = "postSeparatorImage"
if column:
separatorClass = "postSeparatorImage" + column.title()
filename = 'separator_' + column + '.png'
separatorImageFilename = baseDir + '/theme/' + theme + '/icons/' + filename
separatorStr = ''
if os.path.isfile(separatorImageFilename):
separatorStr = \
'
' + \
'
\n'
return separatorStr
def htmlHighlightLabel(label: str, highlight: bool) -> str:
"""If the given text should be highlighted then return
the appropriate markup.
This is so that in shell browsers, like lynx, it's possible
to see if the replies or DM button are highlighted.
"""
if not highlight:
return label
return '*' + str(label) + '*'
def getAvatarImageUrl(session,
baseDir: str, httpPrefix: str,
postActor: str, personCache: {},
avatarUrl: str, allowDownloads: bool) -> str:
"""Returns the avatar image url
"""
# get the avatar image url for the post actor
if not avatarUrl:
avatarUrl = \
getPersonAvatarUrl(baseDir, postActor, personCache,
allowDownloads)
avatarUrl = \
updateAvatarImageCache(session, baseDir, httpPrefix,
postActor, avatarUrl, personCache,
allowDownloads)
else:
updateAvatarImageCache(session, baseDir, httpPrefix,
postActor, avatarUrl, personCache,
allowDownloads)
if not avatarUrl:
avatarUrl = postActor + '/avatar.png'
return avatarUrl
def htmlHideFromScreenReader(htmlStr: str) -> str:
"""Returns html which is hidden from screen readers
"""
return '' + htmlStr + ''
def htmlKeyboardNavigation(banner: str, links: {}, accessKeys: {},
subHeading=None,
usersPath=None, translate=None,
followApprovals=False) -> str:
"""Given a set of links return the html for keyboard navigation
"""
htmlStr = '
\n'
if banner:
htmlStr += '
\n' + banner + '\n
\n'
if subHeading:
htmlStr += ' \n'
# show new follower approvals
if usersPath and translate and followApprovals:
htmlStr += '
\n'
# show the list of links
for title, url in links.items():
accessKeyStr = ''
if accessKeys.get(title):
accessKeyStr = 'accesskey="' + accessKeys[title] + '"'
htmlStr += '\n'
htmlStr += '