Drugsosos
3 years ago
parent
c617af98ce
commit
c3c90cfd90
@ -1,101 +1,101 @@
|
||||
import base64
|
||||
from aiohttp import ClientSession
|
||||
|
||||
from utils import settings
|
||||
import random
|
||||
import requests
|
||||
from requests.adapters import HTTPAdapter, Retry
|
||||
from random import choice
|
||||
|
||||
from attr import attrs, attrib
|
||||
from attr.validators import instance_of
|
||||
|
||||
from TTS.common import BaseApiTTS, get_random_voice
|
||||
|
||||
# TTS examples: https://twitter.com/scanlime/status/1512598559769702406
|
||||
|
||||
# from profanity_filter import ProfanityFilter
|
||||
# pf = ProfanityFilter()
|
||||
# Code by @JasonLovesDoggo
|
||||
# https://twitter.com/scanlime/status/1512598559769702406
|
||||
voices = dict()
|
||||
|
||||
nonhuman = [ # DISNEY VOICES
|
||||
"en_us_ghostface", # Ghost Face
|
||||
"en_us_chewbacca", # Chewbacca
|
||||
"en_us_c3po", # C3PO
|
||||
"en_us_stitch", # Stitch
|
||||
"en_us_stormtrooper", # Stormtrooper
|
||||
"en_us_rocket", # Rocket
|
||||
voices['nonhuman'] = [ # DISNEY VOICES
|
||||
'en_us_ghostface', # Ghost Face
|
||||
'en_us_chewbacca', # Chewbacca
|
||||
'en_us_c3po', # C3PO
|
||||
'en_us_stitch', # Stitch
|
||||
'en_us_stormtrooper', # Stormtrooper
|
||||
'en_us_rocket', # Rocket
|
||||
# ENGLISH VOICES
|
||||
]
|
||||
human = [
|
||||
"en_au_001", # English AU - Female
|
||||
"en_au_002", # English AU - Male
|
||||
"en_uk_001", # English UK - Male 1
|
||||
"en_uk_003", # English UK - Male 2
|
||||
"en_us_001", # English US - Female (Int. 1)
|
||||
"en_us_002", # English US - Female (Int. 2)
|
||||
"en_us_006", # English US - Male 1
|
||||
"en_us_007", # English US - Male 2
|
||||
"en_us_009", # English US - Male 3
|
||||
"en_us_010",
|
||||
voices['human'] = [
|
||||
'en_au_001', # English AU - Female
|
||||
'en_au_002', # English AU - Male
|
||||
'en_uk_001', # English UK - Male 1
|
||||
'en_uk_003', # English UK - Male 2
|
||||
'en_us_001', # English US - Female (Int. 1)
|
||||
'en_us_002', # English US - Female (Int. 2)
|
||||
'en_us_006', # English US - Male 1
|
||||
'en_us_007', # English US - Male 2
|
||||
'en_us_009', # English US - Male 3
|
||||
'en_us_010',
|
||||
]
|
||||
voices = nonhuman + human
|
||||
|
||||
noneng = [
|
||||
"fr_001", # French - Male 1
|
||||
"fr_002", # French - Male 2
|
||||
"de_001", # German - Female
|
||||
"de_002", # German - Male
|
||||
"es_002", # Spanish - Male
|
||||
voices['non_eng'] = [
|
||||
'fr_001', # French - Male 1
|
||||
'fr_002', # French - Male 2
|
||||
'de_001', # German - Female
|
||||
'de_002', # German - Male
|
||||
'es_002', # Spanish - Male
|
||||
# AMERICA VOICES
|
||||
"es_mx_002", # Spanish MX - Male
|
||||
"br_001", # Portuguese BR - Female 1
|
||||
"br_003", # Portuguese BR - Female 2
|
||||
"br_004", # Portuguese BR - Female 3
|
||||
"br_005", # Portuguese BR - Male
|
||||
'es_mx_002', # Spanish MX - Male
|
||||
'br_001', # Portuguese BR - Female 1
|
||||
'br_003', # Portuguese BR - Female 2
|
||||
'br_004', # Portuguese BR - Female 3
|
||||
'br_005', # Portuguese BR - Male
|
||||
# ASIA VOICES
|
||||
"id_001", # Indonesian - Female
|
||||
"jp_001", # Japanese - Female 1
|
||||
"jp_003", # Japanese - Female 2
|
||||
"jp_005", # Japanese - Female 3
|
||||
"jp_006", # Japanese - Male
|
||||
"kr_002", # Korean - Male 1
|
||||
"kr_003", # Korean - Female
|
||||
"kr_004", # Korean - Male 2
|
||||
'id_001', # Indonesian - Female
|
||||
'jp_001', # Japanese - Female 1
|
||||
'jp_003', # Japanese - Female 2
|
||||
'jp_005', # Japanese - Female 3
|
||||
'jp_006', # Japanese - Male
|
||||
'kr_002', # Korean - Male 1
|
||||
'kr_003', # Korean - Female
|
||||
'kr_004', # Korean - Male 2
|
||||
]
|
||||
|
||||
|
||||
# good_voices = {'good': ['en_us_002', 'en_us_006'],
|
||||
# 'ok': ['en_au_002', 'en_uk_001']} # less en_us_stormtrooper more less en_us_rocket en_us_ghostface
|
||||
# good_voices: 'en_us_002', 'en_us_006'
|
||||
# ok: 'en_au_002', 'en_uk_001'
|
||||
# less: en_us_stormtrooper
|
||||
# more or less: en_us_rocket, en_us_ghostface
|
||||
|
||||
|
||||
class TikTok: # TikTok Text-to-Speech Wrapper
|
||||
def __init__(self):
|
||||
self.URI_BASE = (
|
||||
"https://api16-normal-useast5.us.tiktokv.com/media/api/text/speech/invoke/?text_speaker="
|
||||
)
|
||||
self.max_chars = 300
|
||||
self.voices = {"human": human, "nonhuman": nonhuman, "noneng": noneng}
|
||||
|
||||
def run(self, text, filepath, random_voice: bool = False):
|
||||
# if censor:
|
||||
# req_text = pf.censor(req_text)
|
||||
# pass
|
||||
voice = (
|
||||
self.randomvoice()
|
||||
if random_voice
|
||||
else (
|
||||
settings.config["settings"]["tts"]["tiktok_voice"]
|
||||
or random.choice(self.voices["human"])
|
||||
@attrs(auto_attribs=True)
|
||||
class TikTok(BaseApiTTS): # TikTok Text-to-Speech Wrapper
|
||||
client: ClientSession = attrib(
|
||||
validator=instance_of(ClientSession),
|
||||
)
|
||||
random_voice: bool = False
|
||||
uri_base: str = attrib(
|
||||
default='https://api16-normal-useast5.us.tiktokv.com/media/api/text/speech/invoke',
|
||||
kw_only=True,
|
||||
)
|
||||
try:
|
||||
r = requests.post(f"{self.URI_BASE}{voice}&req_text={text}&speaker_map_type=0")
|
||||
except requests.exceptions.SSLError:
|
||||
# https://stackoverflow.com/a/47475019/18516611
|
||||
session = requests.Session()
|
||||
retry = Retry(connect=3, backoff_factor=0.5)
|
||||
adapter = HTTPAdapter(max_retries=retry)
|
||||
session.mount("http://", adapter)
|
||||
session.mount("https://", adapter)
|
||||
r = session.post(f"{self.URI_BASE}{voice}&req_text={text}&speaker_map_type=0")
|
||||
# print(r.text)
|
||||
vstr = [r.json()["data"]["v_str"]][0]
|
||||
b64d = base64.b64decode(vstr)
|
||||
max_chars = 300
|
||||
decode_base64 = True
|
||||
|
||||
with open(filepath, "wb") as out:
|
||||
out.write(b64d)
|
||||
def __attrs_post_init__(self):
|
||||
self.voice = (
|
||||
get_random_voice(voices, 'human')
|
||||
if self.random_voice
|
||||
else str(settings.config['settings']['tts']['tiktok_voice']).lower()
|
||||
if str(settings.config['settings']['tts']['tiktok_voice']).lower() in [
|
||||
voice.lower() for dict_title in voices for voice in voices[dict_title]]
|
||||
else get_random_voice(voices, 'human')
|
||||
)
|
||||
|
||||
def randomvoice(self):
|
||||
return random.choice(self.voices["human"])
|
||||
async def make_request(
|
||||
self,
|
||||
text: str,
|
||||
):
|
||||
return await self.client.post(
|
||||
f'{self.uri_base}',
|
||||
params={
|
||||
'text_speaker': self.voice,
|
||||
'req_text': text,
|
||||
'speaker_map_type': 0,
|
||||
}
|
||||
)
|
||||
|
||||
@ -0,0 +1,70 @@
|
||||
from aiofiles import open
|
||||
|
||||
import base64
|
||||
from random import choice
|
||||
from typing import Union, Optional
|
||||
|
||||
|
||||
class BaseApiTTS:
|
||||
max_chars: int
|
||||
decode_base64: bool = False
|
||||
|
||||
@staticmethod
|
||||
def text_len_sanitize(
|
||||
text: str,
|
||||
max_length: int,
|
||||
) -> list:
|
||||
# Split by comma or dot (else you can lose intonations), if there is non, split by groups of 299 chars
|
||||
if '.' in text and all([split_text.__len__() < max_length for split_text in text.split('.')]):
|
||||
return text.split('.')
|
||||
|
||||
if ',' in text and all([split_text.__len__() < max_length for split_text in text.split(',')]):
|
||||
return text.split(',')
|
||||
|
||||
return [text[i:i + max_length] for i in range(0, len(text), max_length)]
|
||||
|
||||
async def write_file(
|
||||
self,
|
||||
output_text: str,
|
||||
filename: str,
|
||||
) -> None:
|
||||
decoded_text = base64.b64decode(output_text) if self.decode_base64 else output_text
|
||||
|
||||
async with open(filename, 'wb') as out:
|
||||
await out.write(decoded_text)
|
||||
|
||||
async def run(
|
||||
self,
|
||||
req_text: str,
|
||||
filename: str,
|
||||
) -> None:
|
||||
output_text = ''
|
||||
if len(req_text) > self.max_chars:
|
||||
for part in self.text_len_sanitize(req_text, self.max_chars):
|
||||
if part:
|
||||
output_text += await self.make_request(part)
|
||||
else:
|
||||
output_text = await self.make_request(req_text)
|
||||
await self.write_file(output_text, filename)
|
||||
|
||||
|
||||
def get_random_voice(
|
||||
voices: Union[list, dict],
|
||||
key: Optional[str] = None,
|
||||
) -> str:
|
||||
if isinstance(voices, list):
|
||||
return choice(voices)
|
||||
else:
|
||||
return choice(voices[key])
|
||||
|
||||
|
||||
def audio_length(
|
||||
path: str,
|
||||
) -> float | int:
|
||||
from mutagen.mp3 import MP3
|
||||
|
||||
try:
|
||||
audio = MP3(path)
|
||||
return audio.info.length
|
||||
except Exception as e: # TODO add logging
|
||||
return 0
|
||||
@ -1,62 +1,71 @@
|
||||
import random
|
||||
import requests
|
||||
from requests.exceptions import JSONDecodeError
|
||||
from aiohttp import ClientSession
|
||||
|
||||
from random import choice
|
||||
from utils import settings
|
||||
from utils.voice import check_ratelimit
|
||||
from attr import attrs, attrib
|
||||
from attr.validators import instance_of
|
||||
|
||||
from TTS.common import BaseApiTTS, get_random_voice
|
||||
|
||||
|
||||
voices = [
|
||||
"Brian",
|
||||
"Emma",
|
||||
"Russell",
|
||||
"Joey",
|
||||
"Matthew",
|
||||
"Joanna",
|
||||
"Kimberly",
|
||||
"Amy",
|
||||
"Geraint",
|
||||
"Nicole",
|
||||
"Justin",
|
||||
"Ivy",
|
||||
"Kendra",
|
||||
"Salli",
|
||||
"Raveena",
|
||||
'Brian',
|
||||
'Emma',
|
||||
'Russell',
|
||||
'Joey',
|
||||
'Matthew',
|
||||
'Joanna',
|
||||
'Kimberly',
|
||||
'Amy',
|
||||
'Geraint',
|
||||
'Nicole',
|
||||
'Justin',
|
||||
'Ivy',
|
||||
'Kendra',
|
||||
'Salli',
|
||||
'Raveena',
|
||||
]
|
||||
|
||||
|
||||
# valid voices https://lazypy.ro/tts/
|
||||
|
||||
|
||||
class StreamlabsPolly:
|
||||
def __init__(self):
|
||||
self.url = "https://streamlabs.com/polly/speak"
|
||||
self.max_chars = 550
|
||||
self.voices = voices
|
||||
@attrs(auto_attribs=True)
|
||||
class StreamlabsPolly(BaseApiTTS):
|
||||
client: ClientSession = attrib(
|
||||
validator=instance_of(ClientSession),
|
||||
)
|
||||
random_voice: bool = False
|
||||
url: str = attrib(
|
||||
default='https://streamlabs.com/polly/speak',
|
||||
kw_only=True,
|
||||
)
|
||||
|
||||
def run(self, text, filepath, random_voice: bool = False):
|
||||
if random_voice:
|
||||
voice = self.randomvoice()
|
||||
else:
|
||||
if not settings.config["settings"]["tts"]["streamlabs_polly_voice"]:
|
||||
return ValueError(
|
||||
f"Please set the config variable STREAMLABS_VOICE to a valid voice. options are: {voices}"
|
||||
max_chars = 550
|
||||
|
||||
async def make_request(
|
||||
self,
|
||||
text: str,
|
||||
):
|
||||
voice = (
|
||||
get_random_voice(voices)
|
||||
if self.random_voice
|
||||
else str(settings.config['settings']['tts']['streamlabs_polly_voice']).capitalize()
|
||||
if str(settings.config['settings']['tts']['streamlabs_polly_voice']).lower() in [
|
||||
voice.lower() for voice in voices]
|
||||
else get_random_voice(voices)
|
||||
)
|
||||
|
||||
response = await self.client.post(
|
||||
self.url,
|
||||
data={
|
||||
'voice': voice,
|
||||
'text': text,
|
||||
'service': 'polly',
|
||||
}
|
||||
)
|
||||
voice = str(settings.config["settings"]["tts"]["streamlabs_polly_voice"]).capitalize()
|
||||
body = {"voice": voice, "text": text, "service": "polly"}
|
||||
response = requests.post(self.url, data=body)
|
||||
if not check_ratelimit(response):
|
||||
self.run(text, filepath, random_voice)
|
||||
|
||||
else:
|
||||
try:
|
||||
voice_data = requests.get(response.json()["speak_url"])
|
||||
with open(filepath, "wb") as f:
|
||||
f.write(voice_data.content)
|
||||
except (KeyError, JSONDecodeError):
|
||||
try:
|
||||
if response.json()["error"] == "No text specified!":
|
||||
raise ValueError("Please specify a text to convert to speech.")
|
||||
except (KeyError, JSONDecodeError):
|
||||
print("Error occurred calling Streamlabs Polly")
|
||||
|
||||
def randomvoice(self):
|
||||
return random.choice(self.voices)
|
||||
speak_url = await(
|
||||
await response.json()
|
||||
)['speak_url']
|
||||
|
||||
return await self.client.get(speak_url)
|
||||
|
||||
Loading…
Reference in new issue