1"""Text to speech with Piper: sentence-by-sentence synthesis, MP3/WAV/OGG encoding and SRT/VTT subtitles."""
2from __future__ import annotations
3
4import html
5import io
6import json
7import os
8import re
9import subprocess
10import urllib.request
11import wave
12import zipfile
13from html.parser import HTMLParser
14
15import onnxruntime
16from piper import PiperVoice, SynthesisConfig
17from piper.config import PiperConfig
18
19HF = 'https://huggingface.co/rhasspy/piper-voices/resolve/main/'
20
21
22VOICES = {
23 'uk_UA-tetiana-high': ('Ukrainian', 'Apache-2.0'),
24 'uk_UA-mykyta-high': ('Ukrainian', 'Apache-2.0'),
25 'uk_UA-oleksa-high': ('Ukrainian', 'Apache-2.0'),
26 'en_US-ljspeech-high': ('English (US)', 'public domain'),
27 'en_US-john-medium': ('English (US)', 'public domain'),
28 'en_US-kristin-medium': ('English (US)', 'public domain'),
29 'en_US-bryce-medium': ('English (US)', 'public domain'),
30 'en_US-norman-medium': ('English (US)', 'public domain'),
31 'en_US-joe-medium': ('English (US)', 'CC0'),
32 'en_GB-cori-high': ('English (UK)', 'public domain'),
33 'en_GB-alba-medium': ('English (UK)', 'CC BY 4.0'),
34 'de_DE-thorsten-high': ('German', 'CC0'),
35 'de_DE-kerstin-low': ('German', 'CC0'),
36 'fr_FR-siwis-medium': ('French', 'CC BY 4.0'),
37 'fr_FR-gilles-low': ('French', 'CC0'),
38 'es_ES-davefx-medium': ('Spanish (Spain)', 'CC0'),
39 'es_MX-claude-high': ('Spanish (Mexico)', 'Apache-2.0'),
40 'it_IT-serena-high': ('Italian', 'CC BY 4.0'),
41 'pl_PL-gosia-medium': ('Polish', 'CC0'),
42 'pl_PL-darkman-medium': ('Polish', 'CC0'),
43 'pt_BR-faber-medium': ('Portuguese (Brazil)', 'CC0'),
44 'nl_NL-pim-medium': ('Dutch', 'CC0'),
45 'cs_CZ-jirka-medium': ('Czech', 'CC0'),
46 'ru_RU-denis-medium': ('Russian', 'CC0'),
47 'ru_RU-dmitri-medium': ('Russian', 'CC0'),
48 'ro_RO-mihai-medium': ('Romanian', 'CC0'),
49 'fi_FI-harri-medium': ('Finnish', 'CC0'),
50 'hu_HU-anna-medium': ('Hungarian', 'CC0'),
51 'sk_SK-lili-medium': ('Slovak', 'CC0'),
52 'sv_SE-nst-medium': ('Swedish', 'CC0'),
53 'da_DK-talesyntese-medium': ('Danish', 'CC0'),
54 'el_GR-rapunzelina-medium': ('Greek', 'CC0'),
55 'lv_LV-aivars-medium': ('Latvian', 'CC0'),
56 'sl_SI-artur-medium': ('Slovenian', 'CC BY 4.0'),
57 'zh_CN-chaowen-medium': ('Chinese', 'CC0'),
58 'fa_IR-amir-medium': ('Persian', 'CC0'),
59 'vi_VN-vais1000-medium': ('Vietnamese', 'CC BY 4.0'),
60}
61SENTENCE_PAUSE = 0.25
62PARAGRAPH_PAUSE = 0.7
63MAX_SENTENCE_CHARS = 400
64
65
66class TtsError(Exception):
67 pass
68
69
70def voice_path(voice: str) -> str:
71 lang, name, quality = voice.split('-', 2)
72 return f'{lang.split("_")[0]}/{lang}/{name}/{quality}/{voice}'
73
74
75def load_voice(voice: str, cache_dir: str) -> PiperVoice:
76 if voice not in VOICES:
77 raise TtsError(f'unknown voice "{voice}"')
78 os.makedirs(cache_dir, exist_ok=True)
79 model = os.path.join(cache_dir, voice + '.onnx')
80 for suffix in ('.onnx.json', '.onnx'):
81 dest = model[:-5] + suffix
82 if not os.path.exists(dest):
83 url = HF + urllib.request.quote(voice_path(voice)) + suffix
84 try:
85 with urllib.request.urlopen(url, timeout=300) as r, open(dest + '.part', 'wb') as f:
86 while chunk := r.read(1 << 20):
87 f.write(chunk)
88 except OSError as e:
89 raise TtsError(f'could not download voice {voice}: {e}') from e
90 os.replace(dest + '.part', dest)
91
92
93 memory = int(os.environ.get('ACTOR_MEMORY_MBYTES') or os.environ.get('APIFY_MEMORY_MBYTES') or 0)
94 threads = max(1, memory // 4096) if memory else (os.cpu_count() or 1)
95 opts = onnxruntime.SessionOptions()
96 opts.intra_op_num_threads = threads
97 opts.inter_op_num_threads = 1
98 with open(model + '.json', encoding='utf-8') as f:
99 config = PiperConfig.from_dict(json.load(f))
100 session = onnxruntime.InferenceSession(model, sess_options=opts, providers=['CPUExecutionProvider'])
101 return PiperVoice(session=session, config=config)
102
103
104class _Text(HTMLParser):
105 BLOCK = {'p', 'div', 'br', 'li', 'h1', 'h2', 'h3', 'h4', 'h5', 'h6', 'tr', 'section', 'article', 'blockquote'}
106 SKIP = {'script', 'style', 'noscript', 'nav', 'footer', 'header', 'svg'}
107
108 def __init__(self):
109 super().__init__()
110 self.parts, self.skip = [], 0
111
112 def handle_starttag(self, tag, attrs):
113 self.skip += tag in self.SKIP
114 if tag in self.BLOCK:
115 self.parts.append('\n\n')
116
117 def handle_endtag(self, tag):
118 self.skip -= tag in self.SKIP and self.skip > 0
119 if tag in self.BLOCK:
120 self.parts.append('\n\n')
121
122 def handle_data(self, data):
123 if not self.skip:
124 self.parts.append(data)
125
126
127def extract_text(path: str, name: str, ctype: str) -> str:
128 """Plain text from TXT, Markdown, HTML, PDF or DOCX."""
129 with open(path, 'rb') as f:
130 head = f.read(8)
131 ext = os.path.splitext(name or '')[1].lower()
132 if head.startswith(b'%PDF'):
133 import pypdfium2 as pdfium
134 try:
135 pdf = pdfium.PdfDocument(path)
136 except pdfium.PdfiumError as e:
137 raise TtsError('the PDF is damaged or password-protected') from e
138 text = '\n\n'.join(p.get_textpage().get_text_range() for p in pdf)
139 pdf.close()
140 if len(text.strip()) < 20:
141 raise TtsError('the PDF has no text layer (scanned?) — OCR it first')
142
143 return re.sub(r'(?<![.!?:;\n])\r?\n(?!\s*\n)', ' ', text)
144 if head.startswith(b'PK') and (ext == '.docx' or 'officedocument' in ctype):
145 try:
146 with zipfile.ZipFile(path) as z:
147 xml = z.read('word/document.xml').decode('utf-8', 'replace')
148 except (zipfile.BadZipFile, KeyError) as e:
149 raise TtsError('the DOCX file is damaged') from e
150 paras = re.findall(r'<w:p[ >].*?</w:p>', xml, re.S)
151 return '\n\n'.join(html.unescape(''.join(re.findall(r'<w:t[^>]*>([^<]*)</w:t>', p))) for p in paras)
152 raw = open(path, 'rb').read()
153 if b'\x00' in raw[:4096]:
154 raise TtsError('unsupported file: use TXT, Markdown, HTML, PDF or DOCX')
155 text = raw.decode('utf-8', 'replace')
156 if ext in ('.html', '.htm') or 'html' in ctype or re.search(r'<(html|body|p|div)\b', text[:2000], re.I):
157 parser = _Text()
158 parser.feed(text)
159 return html.unescape(''.join(parser.parts))
160 if ext in ('.md', '.markdown') or 'markdown' in ctype:
161 text = re.sub(r'!\[[^\]]*\]\([^)]*\)', '', text)
162 text = re.sub(r'\[([^\]]*)\]\([^)]*\)', r'\1', text)
163 text = re.sub(r'^\s{0,3}(#{1,6}|[-*+]|\d+\.)\s+', '', text, flags=re.M)
164 text = re.sub(r'[*_`~>|]+', '', text)
165 return text
166
167
168def split_text(text: str, max_chars: int) -> tuple[list[list[str]], int, bool]:
169 """Paragraphs of sentences; returns (paragraphs, characters used, truncated)."""
170 text = text.replace('\r', '')
171 paras, used, truncated = [], 0, False
172 for block in re.split(r'\n\s*\n', text):
173 block = re.sub(r'\s+', ' ', block).strip()
174 if not block:
175 continue
176 sentences = []
177 for s in re.split(r'(?<=[.!?…。！？])\s+', block):
178 while len(s) > MAX_SENTENCE_CHARS:
179 cut = max(s.rfind(', ', 0, MAX_SENTENCE_CHARS), s.rfind(' ', 0, MAX_SENTENCE_CHARS))
180 cut = cut if cut > 50 else MAX_SENTENCE_CHARS
181 sentences.append(s[:cut + 1].strip())
182 s = s[cut + 1:].strip()
183 if s:
184 sentences.append(s)
185 kept = []
186 for s in sentences:
187 if used + len(s) > max_chars:
188 truncated = True
189 break
190 kept.append(s)
191 used += len(s)
192 if kept:
193 paras.append(kept)
194 if truncated:
195 break
196 return paras, used, truncated
197
198
199
200QUOTES = re.compile('[«»“”„‟"]')
201
202
203
204STYLES = {'standard': (0.667, 0.8), 'lively': (0.9, 1.15)}
205
206
207def synthesize(voice: PiperVoice, paras: list[list[str]], speed: float, speaker: int | None = None,
208 ukrainian: bool = False, style: str = 'standard') -> tuple[bytes, int, list[tuple[float, float, str]]]:
209 """Returns (16-bit mono PCM, sample rate, cues [(start, end, sentence)]).
210 ukrainian=True puts the word stress from a dictionary (espeak often stresses Ukrainian words wrongly)."""
211 noise, noise_w = STYLES[style]
212 cfg = SynthesisConfig(length_scale=1 / speed, speaker_id=speaker, noise_scale=noise, noise_w_scale=noise_w)
213 rate = voice.config.sample_rate
214 pcm, cues, t = bytearray(), [], 0.0
215 for pi, sentences in enumerate(paras):
216 for si, s in enumerate(sentences):
217 spoken = QUOTES.sub('', s)
218 if ukrainian:
219 from .uk_stress import synthesize_uk
220 audio, _ = synthesize_uk(voice, spoken, cfg)
221 else:
222 audio = b''.join(c.audio_int16_bytes for c in voice.synthesize(spoken, cfg))
223 if not audio:
224 continue
225 dur = len(audio) / 2 / rate
226 cues.append((round(t, 3), round(t + dur, 3), s))
227 pcm += audio
228 pause = (SENTENCE_PAUSE if si < len(sentences) - 1 else PARAGRAPH_PAUSE if pi < len(paras) - 1 else 0) / speed
229 pcm += b'\x00\x00' * int(pause * rate)
230 t += dur + pause
231 return bytes(pcm), rate, cues
232
233
234def encode(pcm: bytes, rate: int, fmt: str, dest: str, bitrate: str = '64k') -> None:
235 buf = io.BytesIO()
236 with wave.open(buf, 'wb') as w:
237 w.setnchannels(1)
238 w.setsampwidth(2)
239 w.setframerate(rate)
240 w.writeframes(pcm)
241 if fmt == 'wav':
242 with open(dest, 'wb') as f:
243 f.write(buf.getvalue())
244 return
245 codec = {'mp3': ['-c:a', 'libmp3lame', '-b:a', bitrate], 'ogg': ['-c:a', 'libopus', '-b:a', bitrate]}[fmt]
246 p = subprocess.run(['ffmpeg', '-v', 'error', '-y', '-f', 'wav', '-i', 'pipe:0', *codec, dest],
247 input=buf.getvalue(), capture_output=True, timeout=900)
248 if p.returncode != 0:
249 raise TtsError('audio encoding failed: ' + p.stderr.decode('utf-8', 'replace')[-200:])
250
251
252def _ts(t: float, sep: str) -> str:
253 ms = int(round(t * 1000))
254 h, ms = divmod(ms, 3600000)
255 m, ms = divmod(ms, 60000)
256 s, ms = divmod(ms, 1000)
257 return f'{h:02d}:{m:02d}:{s:02d}{sep}{ms:03d}'
258
259
260def to_srt(cues) -> str:
261 return '\n'.join(f'{i}\n{_ts(a, ",")} --> {_ts(b, ",")}\n{t}\n' for i, (a, b, t) in enumerate(cues, 1))
262
263
264def to_vtt(cues) -> str:
265 return 'WEBVTT\n\n' + '\n'.join(f'{_ts(a, ".")} --> {_ts(b, ".")}\n{t}\n' for a, b, t in cues)
