"""Narração neural grátis, via Edge TTS (vozes online da Microsoft). ── POR QUE ESTE MOTOR EXISTE ───────────────────────────────────────────── A corrente de voz tinha um buraco entre o ElevenLabs (premium, com chave e cota) e o Google (uma voz só, robótica). No meio ficava o Piper — neural e offline, mas que baixa 90 MB e, em Mac ARM, vinha com binário x86 e travava. Quando o Piper não sobe, a narração caía direto no Google e o vídeo saía com voz de robô. O Edge TTS preenche esse buraco: são as MESMAS vozes neurais do serviço da Microsoft (Antonio, Francisca, Thalita em pt-BR), de graça, SEM chave e SEM conta. Precisa de internet — mas quem está gerando vídeo já está online. A qualidade fica logo abaixo do ElevenLabs e muito acima do Google. ── COMO ENTRA NA CORRENTE ──────────────────────────────────────────────── 1. ELEVENLABS premium, se houver chave com crédito. 2. EDGE TTS neural, grátis, sem chave. ← ESTE 3. PIPER neural, offline, ilimitado. 4. VOICEBOX se o app estiver instalado. 5. GOOGLE última reserva; nunca fica mudo. ── SÍNTESE ─────────────────────────────────────────────────────────────── Rodamos o edge-tts por SUBPROCESSO (`python3 -m edge_tts`), não pela API async: o servidor do Studio é stdlib síncrono, e enfiar um event loop dentro de uma thread dele é fonte de travamento. O subprocesso é isolado, tem timeout de verdade, e devolve o MP3 pronto. A DURAÇÃO sai das legendas que o próprio edge-tts escreve (`--write-subtitles`): o último timestamp do VTT é o fim da fala, medido — não estimado. Com isso a cena sabe quanto dura de fato e a linha do tempo fica determinística. O pacote `edge-tts` não vem no PrimoCode (é pip, não stdlib). Se faltar, este motor se declara indisponível e a corrente segue para o Piper — nada trava. `instalar()` tenta trazê-lo em segundo plano, uma vez, sem bloquear ninguém. """ import os import re import subprocess import sys import tempfile import threading # Catálogo pt-BR do Edge. Só três vozes existem no serviço da Microsoft para # português do Brasil; não há por que inventar mais. A padrão é a Antonio, # masculina e de locução, que é o registro de abertura de vídeo — a mesma # lógica da voz grave padrão do Piper. CATALOGO = [ {"id": "antonio", "voz": "pt-BR-AntonioNeural", "nome": "Antonio", "genero": "m", "padrao": True}, {"id": "francisca", "voz": "pt-BR-FranciscaNeural", "nome": "Francisca", "genero": "f"}, {"id": "thalita", "voz": "pt-BR-ThalitaMultilingualNeural","nome": "Thalita", "genero": "f"}, ] POR_ID = {v["id"]: v for v in CATALOGO} PADRAO = next(v["voz"] for v in CATALOGO if v.get("padrao")) # Apelidos: as vozes do ElevenLabs que o usuário talvez peça no documento # mapeiam para o timbre mais próximo aqui, para o mesmo documento render # igual com qualquer motor ativo. Masculinas → Antonio; femininas → Francisca. APELIDOS = { "brian": "pt-BR-AntonioNeural", "daniel": "pt-BR-AntonioNeural", "adam": "pt-BR-AntonioNeural", "bill": "pt-BR-AntonioNeural", "liam": "pt-BR-AntonioNeural", "jeff": "pt-BR-AntonioNeural", "cadu": "pt-BR-AntonioNeural", "edresson": "pt-BR-AntonioNeural", "faber": "pt-BR-AntonioNeural", "sarah": "pt-BR-FranciscaNeural", "alice": "pt-BR-FranciscaNeural", } _verificado = None # cache do disponivel(): None = ainda não checou _instalando = False _tranca = threading.Lock() def _voz_de(voz_cfg): """Resolve o id/nome pedido no documento para um nome de voz do Edge.""" pedido = "" if isinstance(voz_cfg, dict): pedido = str(voz_cfg.get("voz") or "").strip() elif isinstance(voz_cfg, str): pedido = voz_cfg.strip() if not pedido: return PADRAO baixo = pedido.lower() if baixo in POR_ID: return POR_ID[baixo]["voz"] if baixo in APELIDOS: return APELIDOS[baixo] # Nome cru do Edge ("pt-BR-...Neural") passa direto. if re.match(r"^[a-z]{2}-[A-Z]{2}-\w+$", pedido): return pedido return PADRAO def _taxa(velocidade): """velocidade do documento (1.0 = normal) vira o --rate do edge (+N%).""" try: v = float(velocidade) except (TypeError, ValueError): v = 1.0 v = max(0.5, min(2.0, v)) pct = int(round((v - 1.0) * 100)) return ("+%d%%" % pct) if pct >= 0 else ("%d%%" % pct) def disponivel(): """O edge-tts está utilizável nesta máquina? Resultado é cacheado: a checagem roda o CLI e não vale repetir a cada narração.""" global _verificado if _verificado is not None: return _verificado try: r = subprocess.run( [sys.executable, "-m", "edge_tts", "--help"], stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL, timeout=20) _verificado = (r.returncode == 0) except Exception: _verificado = False return _verificado # Por que a instalação falhava calada, e o vídeo saía com a voz do Google: # `pip install --user` é recusado pelo Python do sistema em Debian/Ubuntu e # no do Homebrew ("externally-managed-environment", PEP 668), e em máquina # sem pip o `-m pip` nem existe. Um `pip install` só, em segundo plano, com a # saída jogada fora, era exatamente a falha que ninguém vê. Agora são várias # tentativas, em ordem, e a última linha do erro de cada uma fica guardada # para o status contar o que houve. ESTRATEGIAS = [ ["-m", "pip", "install", "--user", "--quiet", "edge-tts"], ["-m", "pip", "install", "--user", "--quiet", "--break-system-packages", "edge-tts"], ["-m", "pip", "install", "--quiet", "edge-tts"], ["-m", "pip", "install", "--quiet", "--break-system-packages", "edge-tts"], ] _ultimo_erro = "" _thread = None def _rotina_instalar(): global _instalando, _verificado, _ultimo_erro try: # Sem pip, primeiro o ensurepip — é o que o python.org e a Apple # trazem para o caso. try: r = subprocess.run([sys.executable, "-m", "pip", "--version"], stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL, timeout=30) if r.returncode != 0: subprocess.run([sys.executable, "-m", "ensurepip", "--user", "--default-pip"], stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL, timeout=120) except Exception: pass for args in ESTRATEGIAS: try: r = subprocess.run([sys.executable] + args, stdout=subprocess.DEVNULL, stderr=subprocess.PIPE, timeout=240) except Exception as e: _ultimo_erro = str(e)[-200:] continue _verificado = None if r.returncode == 0 and disponivel(): _ultimo_erro = "" sys.stderr.write("[edge] edge-tts instalado (%s)\n" % " ".join(args[2:])) return _ultimo_erro = (r.stderr or b"").decode("utf-8", "replace").strip().splitlines()[-1:][0][-200:] \ if (r.stderr or b"").strip() else "pip devolveu %s" % r.returncode sys.stderr.write("[edge] não consegui instalar o edge-tts: %s\n" % _ultimo_erro) finally: _verificado = None # força nova checagem with _tranca: _instalando = False def instalar(esperar=0): """Traz o pacote — em segundo plano por padrão; com `esperar` (segundos), quem chamou fica aguardando até esse teto. Esperar existe porque a narração de um vídeo NÃO pode sair pela voz de reserva só porque o pip ainda estava rodando: quem pede voz agora e tem internet merece a voz neural agora. Meio minuto de pip é menos ruim que um vídeo inteiro com voz de robô.""" global _instalando, _thread if disponivel(): return True with _tranca: if not _instalando: _instalando = True _thread = threading.Thread(target=_rotina_instalar, daemon=True) _thread.start() t = _thread if t and esperar: t.join(esperar) return disponivel() def ultimo_erro(): return _ultimo_erro def _dur_do_vtt(caminho): """Último timestamp do VTT = fim da fala, em segundos. Medido, não estimado. Formato do edge-tts: `HH:MM:SS,mmm --> HH:MM:SS,mmm`.""" try: with open(caminho, "r", encoding="utf-8") as fh: texto = fh.read() except OSError: return None fins = re.findall(r"-->\s*(\d\d):(\d\d):(\d\d)[,.](\d\d\d)", texto) if not fins: return None h, m, s, ms = fins[-1] return int(h) * 3600 + int(m) * 60 + int(s) + int(ms) / 1000.0 def sintetizar(texto, voz_cfg=None, timeout=90): """O texto vira (mp3_bytes, duracao_em_s). Devolve (None, None) quando o Edge não está disponível ou falha — e NUNCA levanta exceção, para a corrente de voz cair no motor seguinte sem tratamento especial.""" if not disponivel(): instalar() return None, None # Uma fala por vez: quebras viram uma frase só, com respiro de pontuação. limpo = re.sub(r"\s+", " ", str(texto or "")).strip() if not limpo: return None, None grave = bool(isinstance(voz_cfg, dict) and voz_cfg.get("grave")) velocidade = (voz_cfg.get("velocidade") if isinstance(voz_cfg, dict) else None) or 1.0 voz = _voz_de(voz_cfg) taxa = _taxa(velocidade) pasta = tempfile.mkdtemp(prefix="primo-edge-") try: mp3 = os.path.join(pasta, "voz.mp3") vtt = os.path.join(pasta, "voz.vtt") try: r = subprocess.run( [sys.executable, "-m", "edge_tts", "--voice", voz, "--rate", taxa, "--text", limpo, "--write-media", mp3, "--write-subtitles", vtt], stdout=subprocess.DEVNULL, stderr=subprocess.PIPE, timeout=timeout) except subprocess.TimeoutExpired: sys.stderr.write("[edge] tempo esgotado (%ss)\n" % timeout) return None, None if r.returncode != 0 or not os.path.exists(mp3) or os.path.getsize(mp3) < 800: msg = (r.stderr or b"").decode("utf-8", "replace")[-200:] sys.stderr.write("[edge] síntese falhou: %s\n" % msg) return None, None with open(mp3, "rb") as fh: dados = fh.read() dur = _dur_do_vtt(vtt) return dados, (round(dur, 2) if dur else None) except Exception as e: sys.stderr.write("[edge] erro inesperado: %s\n" % e) return None, None finally: import shutil shutil.rmtree(pasta, ignore_errors=True) def estado(): """O que a interface mostra sobre este motor.""" with _tranca: inst = _instalando return { "motor": "edge", "disponivel": disponivel(), "instalando": inst, "erro": _ultimo_erro or None, "precisa_internet": True, "vozes": [{"id": v["id"], "nome": v["nome"], "genero": v["genero"], "padrao": bool(v.get("padrao"))} for v in CATALOGO], }