Cleanup to prepare for release

Signed-off-by: rany2 <[email protected]>

examples/streaming_with_subtitles.py

@@ -29,7 +29,7 @@ async def amain() -> None:
                 submaker.create_sub((chunk["offset"], chunk["duration"]), chunk["text"])
 
     with open(WEBVTT_FILE, "w", encoding="utf-8") as file:
-        file.write(submaker.generate_subs())
+        file.write(submaker.generate_subs(TEXT))
 
 
 if __name__ == "__main__":

pylintrc

@@ -307,8 +307,7 @@ min-public-methods=2
 [EXCEPTIONS]
 
 # Exceptions that will emit a warning when caught.
-overgeneral-exceptions=BaseException,
-                       Exception
+overgeneral-exceptions=builtins.BaseException,builtins.Exception
 
 
 [FORMAT]

src/edge_tts/communicate.py

@@ -239,7 +239,7 @@ def __init__(
         volume: str = "+0%",
         pitch: str = "+0Hz",
         proxy: Optional[str] = None,
-    ):
+    ):  # pylint: disable=too-many-arguments
         """
         Initializes the Communicate class.
 
@@ -302,7 +302,9 @@ def __init__(
             raise TypeError("proxy must be str")
         self.proxy: Optional[str] = proxy
 
-    async def stream(self) -> AsyncGenerator[Dict[str, Any], None]:
+    async def stream(  # pylint: disable=too-many-statements
+        self,
+    ) -> AsyncGenerator[Dict[str, Any], None]:
         """Streams audio and metadata from the service."""
 
         texts = split_text_by_byte_length(

src/edge_tts/submaker.py

@@ -6,7 +6,8 @@
 """
 
 import math
-from typing import List, Tuple
+import re
+from typing import Callable, List, Tuple, Union
 from xml.sax.saxutils import escape, unescape
 
 
@@ -35,10 +36,60 @@ def mktimestamp(time_unit: float) -> str:
     hour = math.floor(time_unit / 10**7 / 3600)
     minute = math.floor((time_unit / 10**7 / 60) % 60)
     seconds = (time_unit / 10**7) % 60
-    # return f"{hour:02d}:{minute:02d}:{seconds:06.3f}"
     return f"{hour:02d}:{minute:02d}:{seconds:06.3f}".replace(".", ",")
 
 
+def _spinoff_sentence(sentence: str) -> Tuple[str, str, int]:
+    """
+    _spinoff_sentence returns the sentence, the last word of the sentence,
+    and the number of times the last word appears in the sentence.
+
+    Args:
+        sentence (str): The sentence to be processed.
+
+    Returns:
+        Tuple[str, str, int]: The sentence, the last word of the sentence,
+        and the number of times the last word appears in the sentence.
+    """
+    if not isinstance(sentence, str):
+        raise TypeError("sentence must be a string")
+    last_word = sentence[-1]
+    last_word_num = sentence.count(last_word)
+    return (sentence, last_word, last_word_num)
+
+
+def process_text(
+    text: str,
+    *,
+    pattern_chi: str = r"[：“”‘’──{}【】·《》〈〉，、；。？！]",
+    spinoff_sentence: Callable[[str], Tuple[str, str, int]] = _spinoff_sentence,
+) -> List[Tuple[str, str, int]]:
+    """
+    process_text returns the three-dimensional list of the text to be passed
+    to SubMaker's generate_subs method.
+
+    Args:
+        text (str): The text to be processed.
+        pattern_chi (str): The pattern of Chinese characters.
+        spinoff_sentence (function): The function used to process the sentence.
+
+    Returns:
+        List[Tuple[str, str, int]]: The three-dimensional list of the text.
+    """
+    if not isinstance(text, str):
+        raise TypeError("text must be a string")
+    if not isinstance(pattern_chi, str):
+        raise TypeError("pattern_chi must be a string")
+    if not callable(spinoff_sentence):
+        raise TypeError("spinoff_sentence must be a function")
+    sentences = re.split(pattern_chi, text)
+    sentences = [sentence.strip() for sentence in sentences if sentence.strip()]
+    three_dimensional_list = []
+    for sentence in sentences:
+        three_dimensional_list.append(spinoff_sentence(sentence))
+    return three_dimensional_list
+
+
 class SubMaker:
     """
     SubMaker class
@@ -54,7 +105,8 @@ def __init__(self) -> None:
     def create_sub(self, timestamp: Tuple[float, float], text: str) -> None:
         """
         create_sub creates a subtitle with the given timestamp and text
-        and adds it to the list of subtitles
+        and adds it to the list of subtitles, this should be called
+        when receiving the wordboundary event from the service.
 
         Args:
             timestamp (tuple): The offset and duration of the subtitle.
@@ -66,53 +118,58 @@ def create_sub(self, timestamp: Tuple[float, float], text: str) -> None:
         self.offset.append((timestamp[0], timestamp[0] + timestamp[1]))
         self.subs.append(text)
 
-    def generate_subs(self, three_dimensional_list, words_in_cue: int = 10) -> str:
+    def generate_subs(self, text: Union[str, List[Tuple[str, str, int]]]) -> str:
         """
         generate_subs generates the complete subtitle file.
 
         Args:
-            words_in_cue (int): defines the number of words in a given cue
+            text: If the type is List[Tuple[str, str, int]], it is the three-dimensional
+                list of the text already processed. If the type is str, the text will
+                be processed automatically by process_text with the default parameters.
+                It should not use data from WordBoundary events, but the text that was
+                used to generate the audio.
 
         Returns:
             str: The complete subtitle file.
-
-        three_dimensional_list：
-            [(sentence, last_word, last_word_num)， (sentence, last_word, last_word_num)]
         """
         if len(self.subs) != len(self.offset):
             raise ValueError("subs and offset are not of the same length")
 
-        if words_in_cue <= 0:
-            raise ValueError("words_in_cue must be greater than 0")
+        if isinstance(text, str):
+            text = process_text(text)
+        elif isinstance(text, list):
+            for sentence, last_word, last_word_num in text:
+                if not isinstance(sentence, str):
+                    raise TypeError("sentence (first element) must be a string")
+                if not isinstance(last_word, str):
+                    raise TypeError("last_word (second element) must be a string")
+                if not isinstance(last_word_num, int):
+                    raise TypeError("last_word_num (third element) must be an integer")
+        else:
+            raise TypeError("text must be a string or a list")
 
-        # data = "WEBVTT\r\n\r\n"
         data = ""
         sub_state_count = 0
         sub_state_start = -1.0
         sub_state_subs = ""
-        sub_line_count = (
-            0  # new variable used to indicate which line of subtitle this is
-        )
+        sub_line_count = 0
         for idx, (offset, subs) in enumerate(zip(self.offset, self.subs)):
             start_time, end_time = offset
             subs = unescape(subs)
 
             # wordboundary is guaranteed not to contain whitespace
-            # if len(sub_state_subs) > 0:
-            #     sub_state_subs += " "
             sub_state_subs += subs
 
             if sub_state_start == -1.0:
                 sub_state_start = start_time
             sub_state_count += 1
 
-            sentence, last_word, last_word_num = three_dimensional_list[sub_line_count]
+            sentence, last_word, last_word_num = text[sub_line_count]
             if (
                 sub_state_subs.count(last_word) == last_word_num
                 or idx == len(self.offset) - 1
             ):
                 sub_line_count += 1
-                # subs = sub_state_subs
                 subs = sentence
                 split_subs: List[str] = [
                     subs[i : i + 79] for i in range(0, len(subs), 79)

src/edge_tts/util.py

@@ -5,7 +5,6 @@
 
 import argparse
 import asyncio
-import re
 import sys
 from io import TextIOWrapper
 from typing import Any, TextIO, Union
@@ -16,7 +15,7 @@
 async def _print_voices(*, proxy: str) -> None:
     """Print all available voices."""
     voices = await list_voices(proxy=proxy)
-    voices = sorted(voices, key=lambda voice: voice["ShortName"])  # type: ignore
+    voices = sorted(voices, key=lambda voice: voice["ShortName"])
     for idx, voice in enumerate(voices):
         if idx != 0:
             print()
@@ -35,12 +34,6 @@ async def _print_voices(*, proxy: str) -> None:
             print(f"{pretty_key_name}: {voice[key]}")
 
 
-def _spinoff_sentence(sentence):
-    last_word = sentence[-1]
-    last_word_num = sentence.count(last_word)
-    return (sentence, last_word, last_word_num)
-
-
 async def _run_tts(args: Any) -> None:
     """Run TTS after parsing arguments from command line."""
 
@@ -67,14 +60,6 @@ async def _run_tts(args: Any) -> None:
     )
 
     submaker: SubMaker = SubMaker()
-
-    pattern_chi = r"[：“”‘’──{}【】·《》〈〉，、；。？！]"
-    sentences = re.split(pattern_chi, args.text)
-    sentences = [sentence.strip() for sentence in sentences if sentence.strip()]
-    three_dimensional_list = []
-    for sentence in sentences:
-        three_dimensional_list.append(_spinoff_sentence(sentence))
-
     with open(
         args.write_media, "wb"
     ) if args.write_media else sys.stdout.buffer as audio_file:
@@ -90,9 +75,7 @@ async def _run_tts(args: Any) -> None:
         else sys.stderr
     )
     with sub_file:
-        sub_file.write(
-            submaker.generate_subs(three_dimensional_list=three_dimensional_list)
-        )
+        sub_file.write(submaker.generate_subs(args.text))
 
 
 async def amain() -> None:
@@ -116,12 +99,6 @@ async def amain() -> None:
     parser.add_argument("--rate", help="set TTS rate. Default +0%%.", default="+0%")
     parser.add_argument("--volume", help="set TTS volume. Default +0%%.", default="+0%")
     parser.add_argument("--pitch", help="set TTS pitch. Default +0Hz.", default="+0Hz")
-    parser.add_argument(
-        "--words-in-cue",
-        help="number of words in a subtitle cue. Default: 10.",
-        default=10,
-        type=float,
-    )
     parser.add_argument(
         "--write-media", help="send media output to file instead of stdout"
     )