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

631 lines
23 KiB
Python
Raw Blame History

__filename__ = "speaker.py"
__author__ = "Bob Mottram"
__license__ = "AGPL3+"
__version__ = "1.5.0"
__maintainer__ = "Bob Mottram"
__email__ = "bob@libreserver.org"
__status__ = "Production"
__module_group__ = "Accessibility"
import os
import html
import random
import urllib.parse
from utils import data_dir
from utils import get_post_attachments
from utils import get_cached_post_filename
from utils import remove_id_ending
from utils import is_dm
from utils import is_reply
from utils import camel_case_split
from utils import get_domain_from_actor
from utils import get_nickname_from_actor
from utils import get_gender_from_bio
from utils import get_display_name
from utils import remove_html
from utils import load_json
from utils import save_json
from utils import is_pgp_encrypted
from utils import has_object_dict
from utils import acct_dir
from utils import local_actor_url
from utils import get_actor_from_post
from content import html_replace_quote_marks
SPEAKER_REMOVE_CHARS = ('.\n', '. ', ',', ';', '?', '!')
def get_speaker_pitch(display_name: str, screenreader: str,
gender: str) -> int:
"""Returns the speech synthesis pitch for the given name
"""
random.seed(display_name)
range_min = 1
range_max = 100
if 'She' in gender:
range_min = 50
elif 'Him' in gender:
range_max = 50
if screenreader == 'picospeaker':
range_min = -6
range_max = 3
if 'She' in gender:
range_min = -1
elif 'Him' in gender:
range_max = -1
return random.randint(range_min, range_max)
def get_speaker_rate(display_name: str, screenreader: str) -> int:
"""Returns the speech synthesis rate for the given name
"""
random.seed(display_name)
if screenreader == 'picospeaker':
return random.randint(-40, -20)
return random.randint(50, 120)
def get_speaker_range(display_name: str) -> int:
"""Returns the speech synthesis range for the given name
"""
random.seed(display_name)
return random.randint(300, 800)
def _speaker_pronounce(base_dir: str, say_text: str, translate: {}) -> str:
"""Screen readers may not always pronounce correctly, so you
can have a file which specifies conversions. File should contain
line items such as:
Epicyon -> Epi-cyon
"""
pronounce_filename = data_dir(base_dir) + '/speaker_pronounce.txt'
convert_dict = {}
if translate:
convert_dict = {
"Epicyon": "Epi-cyon",
"espeak": "e-speak",
"emoji": "emowji",
"clearnet": "clear-net",
"https": "H-T-T-P-S",
"HTTPS": "H-T-T-P-S",
"XMPP": "X-M-P-P",
"xmpp": "X-M-P-P",
"sql": "S-Q-L",
".js": " dot J-S",
"PSQL": "Postgres S-Q-L",
"SQL": "S-Q-L",
"gdpr": "G-D-P-R",
"kde": "K-D-E",
"AGPL": "Affearo G-P-L",
"agpl": "Affearo G-P-L",
"GPL": "G-P-L",
"gpl": "G-P-L",
"coop": "co-op",
"KMail": "K-Mail",
"kmail": "K-Mail",
"gmail": "G-mail",
"Gmail": "G-mail",
"OpenPGP": "Open P-G-P",
"Tor": "Toor",
"memes": "meemes",
"Memes": "Meemes",
"rofl": translate["laughing"],
"ROFL": translate["laughing"],
"lmao": translate["laughing"],
"LMAO": translate["laughing"],
"fwiw": "for what it's worth",
"fyi": "for your information",
"irl": "in real life",
"IRL": "in real life",
"imho": "in my opinion",
"afaik": "as far as I know",
"AFAIK": "as far as I know",
"fediverse": "fediiverse",
"Fediverse": "Fediiverse",
" foss ": " free and open source software ",
" floss ": " free libre and open source software ",
" FOSS ": "free and open source software",
" FLOSS ": "free libre and open source software",
" oss ": " open source software ",
" OSS ": " open source software ",
"🤔": ". " + translate["thinking emoji"],
"RT @": "Re-Tweet ",
"#nowplaying": translate["hashtag"] + " now-playing",
"#NowPlaying": translate["hashtag"] + " now-playing",
"#": translate["hashtag"] + ' ',
"¯\\_(ツ)_/¯": translate["shrug"],
":D": '. ' + translate["laughing"],
":-D": '. ' + translate["laughing"],
":)": '. ' + translate["smile"],
";)": '. ' + translate["wink"],
":(": '. ' + translate["sad face"],
":-)": '. ' + translate["smile"],
":-(": '. ' + translate["sad face"],
";-)": '. ' + translate["wink"],
":O": '. ' + translate['shocked'],
"?": "? ",
'"': "'",
"*": "",
"(": ",",
")": ","
}
if os.path.isfile(pronounce_filename):
pronounce_list = []
try:
with open(pronounce_filename, 'r', encoding='utf-8') as fp_pro:
pronounce_list = fp_pro.readlines()
except OSError:
print('EX: _speaker_pronounce unable to read ' +
pronounce_filename)
if pronounce_list:
for conversion in pronounce_list:
separator = None
if '->' in conversion:
separator = '->'
elif ';' in conversion:
separator = ';'
elif ':' in conversion:
separator = ':'
elif ',' in conversion:
separator = ','
if not separator:
continue
text = conversion.split(separator)[0].strip()
converted = conversion.split(separator)[1].strip()
convert_dict[text] = converted
for text, converted in convert_dict.items():
if text in say_text:
say_text = say_text.replace(text, converted)
return say_text
def speaker_replace_links(http_prefix: str, nickname: str,
orig_domain: str, orig_domain_full: str,
say_text: str, translate: {},
detected_links: []) -> str:
"""Replaces any links in the given text with "link to [domain]".
Instead of reading out potentially very long and meaningless links
"""
text = say_text
text = text.replace('?v=', '__v=')
for char in SPEAKER_REMOVE_CHARS:
text = text.replace(char, ' ')
text = text.replace('__v=', '?v=')
replacements = {}
replacements_hashtags = {}
words_list = text.split(' ')
if translate.get('Linked'):
linked_str = translate['Linked']
else:
linked_str = 'Linked'
prev_word = ''
for word in words_list:
if word.startswith('v='):
replacements[word] = ''
if word.startswith(':'):
if word.endswith(':'):
replacements[word] = ', emowji ' + word.replace(':', '') + ','
continue
if word.startswith('@') and not prev_word.endswith('RT'):
# replace mentions, but not re-tweets
if translate.get('mentioning'):
replacements[word] = \
translate['mentioning'] + ' ' + word[1:] + ', '
prev_word = word
domain = None
domain_full = None
if 'https://' in word:
domain = word.split('https://')[1]
domain_full = 'https://' + domain
elif 'http://' in word:
domain = word.split('http://')[1]
domain_full = 'http://' + domain
if not domain:
continue
if '/' in domain:
domain = domain.split('/')[0]
if domain.startswith('www.'):
domain = domain.replace('www.', '')
replacements[domain_full] = '. ' + linked_str + ' ' + domain + '.'
if '/tags/' in domain_full and domain != orig_domain:
remote_hashtag_link = \
http_prefix + '://' + orig_domain_full + '/users/' + \
nickname + '?remotetag=' + domain_full.replace('/', '--')
detected_links.append(remote_hashtag_link)
else:
detected_links.append(domain_full)
for replace_str, new_str in replacements.items():
say_text = say_text.replace(replace_str, new_str)
for replace_str, new_str in replacements_hashtags.items():
say_text = say_text.replace(replace_str, new_str)
return say_text.replace('..', '.')
def _add_ssml_emphasis(say_text: str) -> str:
"""Adds emphasis to *emphasised* text
"""
if '*' not in say_text:
return say_text
text = say_text
for char in SPEAKER_REMOVE_CHARS:
text = text.replace(char, ' ')
words_list = text.split(' ')
replacements = {}
for word in words_list:
if not word.startswith('*'):
continue
if not word.endswith('*'):
continue
replacements[word] = \
'<emphasis level="strong">' + word.replace('*', '') + \
'</emphasis>'
for replace_str, new_str in replacements.items():
say_text = say_text.replace(replace_str, new_str)
return say_text
def _remove_emoji_from_text(say_text: str) -> str:
"""Removes :emoji: from the given text
"""
if ':' not in say_text:
return say_text
text = say_text
for char in SPEAKER_REMOVE_CHARS:
text = text.replace(char, ' ')
words_list = text.split(' ')
replacements = {}
for word in words_list:
if not word.startswith(':'):
continue
if word.endswith(':'):
replacements[word] = ''
for replace_str, new_str in replacements.items():
say_text = say_text.replace(replace_str, new_str)
return say_text.replace(' ', ' ').strip()
def _speaker_endpoint_json(display_name: str, summary: str,
content: str, say_content: str,
image_description: str,
links: [], gender: str, post_id: str,
post_dm: bool, post_reply: bool,
follow_requests_exist: bool,
follow_requests_list: [],
liked_by: str, published: str, post_cal: bool,
post_share: bool, theme_name: str,
is_direct: bool, reply_to_you: bool) -> {}:
"""Returns a json endpoint for the TTS speaker
"""
speaker_json = {
"name": display_name,
"summary": summary,
"content": content,
"say": say_content,
"published": published,
"imageDescription": image_description,
"detectedLinks": links,
"id": post_id,
"direct": is_direct,
"replyToYou": reply_to_you,
"notify": {
"theme": theme_name,
"dm": post_dm,
"reply": post_reply,
"followRequests": follow_requests_exist,
"followRequestsList": follow_requests_list,
"likedBy": liked_by,
"calendar": post_cal,
"share": post_share
}
}
if gender:
speaker_json['gender'] = gender
return speaker_json
def _ssml_header(system_language: str, box_name: str, summary: str) -> str:
"""Returns a header for an SSML document
"""
if summary:
summary = ': ' + summary
return '<?xml version="1.0"?>\n' + \
'<speak xmlns="http://www.w3.org/2001/10/synthesis"\n' + \
' xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"\n' + \
' xsi:schemaLocation="http://www.w3.org/2001/10/synthesis\n' + \
' http://www.w3.org/TR/speech-synthesis11/synthesis.xsd"\n' + \
' version="1.1">\n' + \
' <metadata>\n' + \
' <dc:title xml:lang="' + system_language + '">' + \
box_name + summary + '</dc:title>\n' + \
' </metadata>\n'
def _speaker_endpoint_ssml(display_name: str, summary: str,
content: str, language: str,
gender: str, box_name: str) -> str:
"""Returns an SSML endpoint for the TTS speaker
https://en.wikipedia.org/wiki/Speech_Synthesis_Markup_Language
https://www.w3.org/TR/speech-synthesis/
"""
lang_short = 'en'
if language:
lang_short = language[:2]
if not gender:
gender = 'neutral'
else:
if lang_short == 'en':
gender = gender.lower()
if 'he/him' in gender:
gender = 'male'
elif 'she/her' in gender:
gender = 'female'
else:
gender = 'neutral'
content = _add_ssml_emphasis(content)
voice_params = 'name="' + display_name + '" gender="' + gender + '"'
if summary is None:
summary = ''
return _ssml_header(lang_short, box_name, summary) + \
' <p>\n' + \
' <s xml:lang="' + language + '">\n' + \
' <voice ' + voice_params + '>\n' + \
' ' + content + '\n' + \
' </voice>\n' + \
' </s>\n' + \
' </p>\n' + \
'</speak>\n'
def get_ssml_box(base_dir: str, path: str,
domain: str,
system_language: str,
box_name: str) -> str:
"""Returns SSML for the given timeline
"""
nickname = path.split('/users/')[1]
if '/' in nickname:
nickname = nickname.split('/')[0]
speaker_filename = \
acct_dir(base_dir, nickname, domain) + '/speaker.json'
if not os.path.isfile(speaker_filename):
return None
speaker_json = load_json(speaker_filename)
if not speaker_json:
return None
gender = None
if speaker_json.get('gender'):
gender = speaker_json['gender']
return _speaker_endpoint_ssml(speaker_json['name'],
speaker_json['summary'],
speaker_json['say'],
system_language,
gender, box_name)
def speakable_text(http_prefix: str,
nickname: str, domain: str, domain_full: str,
base_dir: str, content: str, translate: {}) -> (str, []):
"""Convert the given text to a speakable version
which includes changes for pronunciation
"""
content = str(content)
if is_pgp_encrypted(content):
return content, []
# replace some emoji before removing html
if ' <3' in content:
content = content.replace(' <3', ' ' + translate['heart'])
content = remove_html(html_replace_quote_marks(content))
detected_links = []
content = speaker_replace_links(http_prefix,
nickname, domain, domain_full,
content, translate, detected_links)
# replace all double spaces
while ' ' in content:
content = content.replace(' ', ' ')
content = content.replace(' . ', '. ').strip()
say_content = _speaker_pronounce(base_dir, content, translate)
# replace all double spaces
while ' ' in say_content:
say_content = say_content.replace(' ', ' ')
say_str = say_content.replace(' . ', '. ').strip()
return say_str, detected_links
def _post_to_speaker_json(base_dir: str, http_prefix: str,
nickname: str, domain: str, domain_full: str,
post_json_object: {}, person_cache: {},
translate: {}, announcing_actor: str,
theme_name: str) -> {}:
"""Converts an ActivityPub post into some Json containing
speech synthesis parameters.
NOTE: There currently appears to be no standardized json
format for speech synthesis
"""
if not has_object_dict(post_json_object):
return {}
if not post_json_object['object'].get('content'):
return {}
if not isinstance(post_json_object['object']['content'], str):
return {}
detected_links = []
content = urllib.parse.unquote_plus(post_json_object['object']['content'])
content = html.unescape(content)
content = content.replace('<p>', '').replace('</p>', ' ')
if not is_pgp_encrypted(content):
# replace some emoji before removing html
if ' <3' in content:
content = content.replace(' <3', ' ' + translate['heart'])
content = remove_html(html_replace_quote_marks(content))
content = speaker_replace_links(http_prefix,
nickname, domain, domain_full,
content, translate, detected_links)
# replace all double spaces
while ' ' in content:
content = content.replace(' ', ' ')
content = content.replace(' . ', '. ').strip()
say_content = content
say_content = _speaker_pronounce(base_dir, content, translate)
# replace all double spaces
while ' ' in say_content:
say_content = say_content.replace(' ', ' ')
say_content = say_content.replace(' . ', '. ').strip()
else:
say_content = content
image_description = ''
post_attachments = get_post_attachments(post_json_object)
if post_attachments:
if isinstance(post_attachments, list):
for img in post_attachments:
if not isinstance(img, dict):
continue
if not img.get('name'):
continue
if isinstance(img['name'], str):
image_description += remove_html(img['name']) + '. '
is_direct = is_dm(post_json_object)
actor = local_actor_url(http_prefix, nickname, domain_full)
reply_to_you = is_reply(post_json_object, actor)
published = ''
if post_json_object['object'].get('published'):
published = post_json_object['object']['published']
summary = ''
if post_json_object['object'].get('summary'):
if isinstance(post_json_object['object']['summary'], str):
post_json_object_summary = post_json_object['object']['summary']
summary = \
urllib.parse.unquote_plus(post_json_object_summary)
summary = html.unescape(summary)
actor_url = get_actor_from_post(post_json_object)
speaker_name = \
get_display_name(base_dir, actor_url, person_cache)
if not speaker_name:
return {}
speaker_name = _remove_emoji_from_text(speaker_name)
speaker_name = speaker_name.replace('_', ' ')
speaker_name = camel_case_split(speaker_name)
actor_url = get_actor_from_post(post_json_object)
gender = get_gender_from_bio(base_dir, actor_url,
person_cache, translate)
if announcing_actor:
announced_nickname = get_nickname_from_actor(announcing_actor)
announced_domain, _ = \
get_domain_from_actor(announcing_actor)
if announced_nickname and announced_domain:
announced_handle = announced_nickname + '@' + announced_domain
say_content = \
translate['announces'] + ' ' + \
announced_handle + '. ' + say_content
content = \
translate['announces'] + ' ' + \
announced_handle + '. ' + content
post_id = None
if post_json_object['object'].get('id'):
post_id = remove_id_ending(post_json_object['object']['id'])
follow_requests_exist = False
follow_requests_list = []
accounts_dir = acct_dir(base_dir, nickname, domain_full)
approve_follows_filename = accounts_dir + '/followrequests.txt'
if os.path.isfile(approve_follows_filename):
follows = []
try:
with open(approve_follows_filename, 'r',
encoding='utf-8') as fp_foll:
follows = fp_foll.readlines()
except OSError:
print('EX: _post_to_speaker_json unable to read ' +
approve_follows_filename)
if follows:
if len(follows) > 0:
follow_requests_exist = True
for i, _ in enumerate(follows):
follows[i] = follows[i].strip()
follow_requests_list = follows
post_dm = False
dm_filename = accounts_dir + '/.newDM'
if os.path.isfile(dm_filename):
post_dm = True
post_reply = False
reply_filename = accounts_dir + '/.newReply'
if os.path.isfile(reply_filename):
post_reply = True
liked_by = ''
like_filename = accounts_dir + '/.newLike'
if os.path.isfile(like_filename):
try:
with open(like_filename, 'r', encoding='utf-8') as fp_like:
liked_by = fp_like.read()
except OSError:
print('EX: _post_to_speaker_json unable to read 2 ' +
like_filename)
calendar_filename = accounts_dir + '/.newCalendar'
post_cal = os.path.isfile(calendar_filename)
share_filename = accounts_dir + '/.newShare'
post_share = os.path.isfile(share_filename)
return _speaker_endpoint_json(speaker_name, summary,
content, say_content, image_description,
detected_links, gender, post_id,
post_dm, post_reply,
follow_requests_exist,
follow_requests_list,
liked_by, published,
post_cal, post_share, theme_name,
is_direct, reply_to_you)
def update_speaker(base_dir: str, http_prefix: str,
nickname: str, domain: str, domain_full: str,
post_json_object: {}, person_cache: {},
translate: {}, announcing_actor: str,
theme_name: str,
system_language: str, box_name: str) -> None:
""" Generates a json file which can be used for TTS announcement
of incoming inbox posts
"""
speaker_json = \
_post_to_speaker_json(base_dir, http_prefix,
nickname, domain, domain_full,
post_json_object, person_cache,
translate, announcing_actor,
theme_name)
if not speaker_json:
return
account_dir = acct_dir(base_dir, nickname, domain)
speaker_filename = account_dir + '/speaker.json'
save_json(speaker_json, speaker_filename)
# save the ssml
cached_ssml_filename = \
get_cached_post_filename(base_dir, nickname,
domain, post_json_object)
if not cached_ssml_filename:
return
cached_ssml_filename = cached_ssml_filename.replace('.html', '.ssml')
if box_name == 'outbox':
cached_ssml_filename = \
cached_ssml_filename.replace('/postcache/', '/outbox/')
gender = None
if speaker_json.get('gender'):
gender = speaker_json['gender']
ssml_str = \
_speaker_endpoint_ssml(speaker_json['name'],
speaker_json['summary'],
speaker_json['say'],
system_language,
gender, box_name)
try:
with open(cached_ssml_filename, 'w+', encoding='utf-8') as fp_ssml:
fp_ssml.write(ssml_str)
except OSError:
print('EX: unable to write ssml ' + cached_ssml_filename)
Reference in New Issue View Git Blame Copy Permalink