stream_gTTS.py

import calendar
import math
import re
import time

import gtts
import requests
import urllib3
from gtts_token import gtts_token

from .proxy import proxies


class _Token(gtts_token.Token):
    def _get_token_key(self):
        if self.token_key is not None:
            return self.token_key

        response = requests.get(
            "https://translate.google.com/",
            proxies=proxies('token_google')
        )
        line = response.text.split('\n')[-1]
        tkk_expr = re.search(".*?(TKK=.*?;)W.*?", line).group(1)

        try:
            # Grab the token directly if already generated by function call
            result = re.search("\d{6}\.[0-9]+", tkk_expr).group(0)
        except AttributeError:
            # Generate the token using algorithm
            timestamp = calendar.timegm(time.gmtime())
            hours = int(math.floor(timestamp / 3600))
            a = re.search("a\\\\x3d(-?\d+);", tkk_expr).group(1)
            b = re.search("b\\\\x3d(-?\d+);", tkk_expr).group(1)

            result = str(hours) + "." + str(int(a) + int(b))

        self.token_key = result
        return result


class gTTS(gtts.gTTS):
    def __init__(
            self,
            text,
            lang='en',
            slow=False,
            lang_check=True,
            pre_processor_funcs=None,
            tokenizer_func=gtts.tts.Tokenizer([
                gtts.tts.tokenizer_cases.tone_marks,
                gtts.tts.tokenizer_cases.period_comma,
                gtts.tts.tokenizer_cases.other_punctuation
            ]).run,
            **_
    ):

        if pre_processor_funcs is None:
            pre_processor_funcs = [
                gtts.tts.pre_processors.tone_marks,
                gtts.tts.pre_processors.end_of_line,
                gtts.tts.pre_processors.abbreviations,
                gtts.tts.pre_processors.word_sub
            ]
        # Text
        assert text, 'No text to speak'
        self.text = text

        # Language
        if lang_check:
            try:
                langs = gtts.tts.tts_langs()
                if lang.lower() not in langs:
                    raise ValueError("Language not supported: %s" % lang)
            except RuntimeError:
                pass

        self.lang_check = lang_check
        self.lang = lang.lower()

        # Read speed
        if slow:
            self.speed = gtts.tts.Speed.SLOW
        else:
            self.speed = gtts.tts.Speed.NORMAL

        # Pre-processors and tokenizer
        self.pre_processor_funcs = pre_processor_funcs
        self.tokenizer_func = tokenizer_func

        # Google Translate token
        self.token = _Token()

    def stream_to_fps(self, fps):
        if not isinstance(fps, list):
            fps = [fps]
        urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning)

        text_parts = self._tokenize(self.text)
        assert text_parts, 'No text to send to TTS API'
        for idx, part in enumerate(text_parts):
            try:
                # Calculate token
                part_tk = self.token.calculate_token(part)
            except requests.exceptions.RequestException as e:  # pragma: no cover
                raise gtts.gTTSError(
                    "Connection error during token calculation: %s" %
                    str(e))

            payload = {'ie': 'UTF-8',
                       'q': part,
                       'tl': self.lang,
                       'ttsspeed': self.speed,
                       'total': len(text_parts),
                       'idx': idx,
                       'client': 'tw-ob',
                       'textlen': len(part),
                       'tk': part_tk}
            r = None
            try:
                # Request
                r = requests.get(self.GOOGLE_TTS_URL,
                                 params=payload,
                                 headers=self.GOOGLE_TTS_HEADERS,
                                 proxies=proxies('tts_google'),
                                 verify=False,
                                 stream=True)

                r.raise_for_status()
            except requests.exceptions.HTTPError:
                # Request successful, bad response
                raise gtts.gTTSError(tts=self, response=r)
            except requests.exceptions.RequestException as e:  # pragma: no cover
                # Request failed
                raise gtts.gTTSError(str(e))
            for chunk in r.iter_content(chunk_size=1024):
                for f in fps:
                    f.write(chunk)