# -*- coding: utf-8 -*-
import base64
import json
import logging
import re
import urllib
import requests
from gtts.lang import _fallback_deprecated_lang, tts_langs
from gtts.tokenizer import Tokenizer, pre_processors, tokenizer_cases
from gtts.utils import _clean_tokens, _minimize, _translate_url
__all__ = ["gTTS", "gTTSError"]
# Logger
log = logging.getLogger(__name__)
log.addHandler(logging.NullHandler())
class Speed:
"""Read Speed
The Google TTS Translate API supports two speeds:
Slow: True
Normal: None
"""
SLOW = True
NORMAL = None
[docs]
class gTTS:
"""gTTS -- Google Text-to-Speech.
An interface to Google Translate's Text-to-Speech API.
Args:
text (string): The text to be read.
tld (string): Top-level domain for the Google Translate host,
i.e `https://translate.google.<tld>`. Different Google domains
can produce different localized 'accents' for a given
language. This is also useful when ``google.com`` might be blocked
within a network but a local or different Google host
(e.g. ``google.com.hk``) is not. Default is ``com``.
lang (string, optional): The language (IETF language tag) to
read the text in. Default is ``en``.
slow (bool, optional): Reads text more slowly. Defaults to ``False``.
lang_check (bool, optional): Strictly enforce an existing ``lang``,
to catch a language error early. If set to ``True``,
a ``ValueError`` is raised if ``lang`` doesn't exist.
Setting ``lang_check`` to ``False`` skips Web requests
(to validate language) and therefore speeds up instantiation.
Default is ``True``.
pre_processor_funcs (list): A list of zero or more functions that are
called to transform (pre-process) text before tokenizing. Those
functions must take a string and return a string. Defaults to::
[
pre_processors.tone_marks,
pre_processors.end_of_line,
pre_processors.abbreviations,
pre_processors.word_sub
]
tokenizer_func (callable): A function that takes in a string and
returns a list of string (tokens). Defaults to::
Tokenizer([
tokenizer_cases.tone_marks,
tokenizer_cases.period_comma,
tokenizer_cases.colon,
tokenizer_cases.other_punctuation
]).run
timeout (float or tuple, optional): Seconds to wait for the server to
send data before giving up, as a float, or a ``(connect timeout,
read timeout)`` tuple. ``None`` will wait forever (default).
See Also:
:doc:`Pre-processing and tokenizing <tokenizer>`
Raises:
AssertionError: When ``text`` is ``None`` or empty; when there's nothing
left to speak after pre-processing, tokenizing and cleaning.
ValueError: When ``lang_check`` is ``True`` and ``lang`` is not supported.
RuntimeError: When ``lang_check`` is ``True`` but there's an error loading
the languages dictionary.
"""
GOOGLE_TTS_MAX_CHARS = 100 # Max characters the Google TTS API takes at a time
GOOGLE_TTS_HEADERS = {
"Referer": "http://translate.google.com/",
"User-Agent": "Mozilla/5.0 (Windows NT 10.0; WOW64) "
"AppleWebKit/537.36 (KHTML, like Gecko) "
"Chrome/47.0.2526.106 Safari/537.36",
"Content-Type": "application/x-www-form-urlencoded;charset=utf-8",
}
GOOGLE_TTS_RPC = "jQ1olc"
def __init__(
self,
text,
tld="com",
lang="en",
slow=False,
lang_check=True,
pre_processor_funcs=[
pre_processors.tone_marks,
pre_processors.end_of_line,
pre_processors.abbreviations,
pre_processors.word_sub,
],
tokenizer_func=Tokenizer(
[
tokenizer_cases.tone_marks,
tokenizer_cases.period_comma,
tokenizer_cases.colon,
tokenizer_cases.other_punctuation,
]
).run,
timeout=None,
):
# Debug
for k, v in dict(locals()).items():
if k == "self":
continue
log.debug("%s: %s", k, v)
# Text
assert text, "No text to speak"
self.text = text
# Translate URL top-level domain
self.tld = tld
# Language
self.lang_check = lang_check
self.lang = lang
if self.lang_check:
# Fallback lang in case it is deprecated
self.lang = _fallback_deprecated_lang(lang)
try:
langs = tts_langs()
if self.lang not in langs:
raise ValueError("Language not supported: %s" % lang)
except RuntimeError as e:
log.debug(str(e), exc_info=True)
log.warning(str(e))
# Read speed
if slow:
self.speed = Speed.SLOW
else:
self.speed = Speed.NORMAL
# Pre-processors and tokenizer
self.pre_processor_funcs = pre_processor_funcs
self.tokenizer_func = tokenizer_func
self.timeout = timeout
def _tokenize(self, text):
# Pre-clean
text = text.strip()
# Apply pre-processors
for pp in self.pre_processor_funcs:
log.debug("pre-processing: %s", pp)
text = pp(text)
if len(text) <= self.GOOGLE_TTS_MAX_CHARS:
return _clean_tokens([text])
# Tokenize
log.debug("tokenizing: %s", self.tokenizer_func)
tokens = self.tokenizer_func(text)
# Clean
tokens = _clean_tokens(tokens)
# Minimize
min_tokens = []
for t in tokens:
min_tokens += _minimize(t, " ", self.GOOGLE_TTS_MAX_CHARS)
# Filter empty tokens, post-minimize
tokens = [t for t in min_tokens if t]
return tokens
def _prepare_requests(self):
"""Created the TTS API the request(s) without sending them.
Returns:
list: ``requests.PreparedRequests_``. <https://2.python-requests.org/en/master/api/#requests.PreparedRequest>`_``.
"""
# TTS API URL
translate_url = _translate_url(
tld=self.tld, path="_/TranslateWebserverUi/data/batchexecute"
)
text_parts = self._tokenize(self.text)
log.debug("text_parts: %s", str(text_parts))
log.debug("text_parts: %i", len(text_parts))
assert text_parts, "No text to send to TTS API"
prepared_requests = []
for idx, part in enumerate(text_parts):
data = self._package_rpc(part)
log.debug("data-%i: %s", idx, data)
# Request
r = requests.Request(
method="POST",
url=translate_url,
data=data,
headers=self.GOOGLE_TTS_HEADERS,
)
# Prepare request
prepared_requests.append(r.prepare())
return prepared_requests
def _package_rpc(self, text):
parameter = [text, self.lang, self.speed, "null"]
escaped_parameter = json.dumps(parameter, separators=(",", ":"))
rpc = [[[self.GOOGLE_TTS_RPC, escaped_parameter, None, "generic"]]]
espaced_rpc = json.dumps(rpc, separators=(",", ":"))
return "f.req={}&".format(urllib.parse.quote(espaced_rpc))
[docs]
def get_bodies(self):
"""Get TTS API request bodies(s) that would be sent to the TTS API.
Returns:
list: A list of TTS API request bodies to make.
"""
return [pr.body for pr in self._prepare_requests()]
[docs]
def stream(self):
"""Do the TTS API request(s) and stream bytes
Raises:
:class:`gTTSError`: When there's an error with the API request.
"""
# When disabling ssl verify in requests (for proxies and firewalls),
# urllib3 prints an insecure warning on stdout. We disable that.
try:
requests.packages.urllib3.disable_warnings(
requests.packages.urllib3.exceptions.InsecureRequestWarning
)
except:
pass
prepared_requests = self._prepare_requests()
for idx, pr in enumerate(prepared_requests):
try:
with requests.Session() as s:
# Send request
r = s.send(
request=pr,
verify=False,
proxies=urllib.request.getproxies(),
timeout=self.timeout,
)
log.debug("headers-%i: %s", idx, r.request.headers)
log.debug("url-%i: %s", idx, r.request.url)
log.debug("status-%i: %s", idx, r.status_code)
r.raise_for_status()
except requests.exceptions.HTTPError as e: # pragma: no cover
# Request successful, bad response
log.debug(str(e))
raise gTTSError(tts=self, response=r)
except requests.exceptions.RequestException as e: # pragma: no cover
# Request failed
log.debug(str(e))
raise gTTSError(tts=self)
# Write
for line in r.iter_lines(chunk_size=1024):
decoded_line = line.decode("utf-8")
if "jQ1olc" in decoded_line:
audio_search = re.search(r'jQ1olc","\[\\"(.*)\\"]', decoded_line)
if audio_search:
as_bytes = audio_search.group(1).encode("ascii")
yield base64.b64decode(as_bytes)
else:
# Request successful, good response,
# no audio stream in response
raise gTTSError(tts=self, response=r)
log.debug("part-%i created", idx)
[docs]
def write_to_fp(self, fp):
"""Do the TTS API request(s) and write bytes to a file-like object.
Args:
fp (file object): Any file-like object to write the ``mp3`` to.
Raises:
:class:`gTTSError`: When there's an error with the API request.
TypeError: When ``fp`` is not a file-like object that takes bytes.
"""
try:
for idx, decoded in enumerate(self.stream()):
fp.write(decoded)
log.debug("part-%i written to %s", idx, fp)
except (AttributeError, TypeError) as e:
raise TypeError(
"'fp' is not a file-like object or it does not take bytes: %s" % str(e)
)
[docs]
def save(self, savefile):
"""Do the TTS API request and write result to file.
Args:
savefile (string): The path and file name to save the ``mp3`` to.
Raises:
:class:`gTTSError`: When there's an error with the API request.
"""
with open(str(savefile), "wb") as f:
self.write_to_fp(f)
f.flush()
log.debug("Saved to %s", savefile)
[docs]
class gTTSError(Exception):
"""Exception that uses context to present a meaningful error message"""
def __init__(self, msg=None, **kwargs):
self.tts = kwargs.pop("tts", None)
self.rsp = kwargs.pop("response", None)
if msg:
self.msg = msg
elif self.tts is not None:
self.msg = self.infer_msg(self.tts, self.rsp)
else:
self.msg = None
super(gTTSError, self).__init__(self.msg)
[docs]
def infer_msg(self, tts, rsp=None):
"""Attempt to guess what went wrong by using known
information (e.g. http response) and observed behaviour
"""
cause = "Unknown"
if rsp is None:
premise = "Failed to connect"
if tts.tld != "com":
host = _translate_url(tld=tts.tld)
cause = "Host '{}' is not reachable".format(host)
else:
# rsp should be <requests.Response>
# http://docs.python-requests.org/en/master/api/
status = rsp.status_code
reason = rsp.reason
premise = "{:d} ({}) from TTS API".format(status, reason)
if status == 403:
cause = "Bad token or upstream API changes"
elif status == 404 and tts.tld != "com":
cause = "Unsupported tld '{}'".format(tts.tld)
elif status == 200 and not tts.lang_check:
cause = (
"No audio stream in response. Unsupported language '%s'"
% self.tts.lang
)
elif status >= 500:
cause = "Upstream API error. Try again later."
return "{}. Probable cause: {}".format(premise, cause)