| import os |
| import re |
| import asyncio |
| import tempfile |
| import logging |
| import threading |
| import time |
| import shutil |
| import subprocess |
|
|
| import gradio as gr |
| import edge_tts |
| from pydub import AudioSegment |
|
|
|
|
| |
| |
| |
|
|
| logging.basicConfig(level=logging.INFO) |
| logger = logging.getLogger("MyanmarEdgeTTS") |
|
|
| APP_TITLE = "Myanmar Multilingual TTS Studio" |
|
|
| WORDS_PER_SUBTITLE = 20 |
| MAX_CHUNK_CHARACTERS = 1800 |
| MAX_CONCURRENT_JOBS = 2 |
| MIN_AUDIO_BYTES = 100 |
|
|
| DEFAULT_VOICE_ID = "it-IT-GiuseppeMultilingualNeural" |
| DEFAULT_SPEED = 30 |
|
|
| OUTPUT_DIR = os.path.join( |
| tempfile.gettempdir(), |
| "myanmar_multilingual_tts", |
| ) |
| os.makedirs(OUTPUT_DIR, exist_ok=True) |
|
|
|
|
| |
| |
| |
|
|
| PINNED_VOICES = { |
| "my-MM-NilarNeural", |
| "my-MM-ThihaNeural", |
| "it-IT-GiuseppeMultilingualNeural", |
| } |
|
|
| VOICE_DISPLAY_NAMES = { |
| "it-IT-GiuseppeMultilingualNeural": "နေတိုး", |
| "my-MM-NilarNeural": "ဝတ်မှုန်ရွှေရည်", |
| "my-MM-ThihaNeural": "ပြေတီဦး", |
| } |
|
|
| VOICE_ALIAS_NAMES = [ |
| "အကယ်ဒမီမင်းမော်ကွန်း", |
| "ပိုင်တံခွန်", |
| "ရဲတိုက်", |
| "မြင့်မြတ်", |
| "နေမင်း", |
| "အောင်ရဲလင်း", |
| "ကျော်ကျော်ဗို", |
| "လွင်မိုး", |
| "ပြေတီဦး", |
| "နေတိုး", |
| "ဝတ်မှုန်ရွှေရည်", |
| "ဖွေးဖွေး", |
| "အိန္ဒြာကျော်ဇင်", |
| "သက်မွန်မြင့်", |
| "မိုးဟေကို", |
| "ရွှေမှုံရတီ", |
| "သင်ဇာဝင့်ကျော်", |
| "ပိုင်ဖြိုးသု", |
| "ခင်ဝင့်ဝါ", |
| "အေးမြတ်သူ", |
| "အိချောပို", |
| "နန်းဆုဦး", |
| "အေးဝတ်ရည်သောင်း", |
| "ဆုပန်ထွာ", |
| "မေမြင့်မိုရ်", |
| "ဝါဆိုမိုးဦး", |
| "သံသာမိုးသိမ့်", |
| "မြတ်နိုးအေး", |
| "ခင်သဇင်", |
| "ယွန်းယွန်း", |
| "မင်းဦး", |
| "ခန့်စည်သူ", |
| "ထွန်းထွန်း", |
| "ဒေါင်း", |
| "အောင်လေး", |
| "ဇေရဲထက်", |
| "ဟိန်းဝေယံ", |
| "မင်းရာဇာ", |
| "ရဲအောင်", |
| "ကျော်ဟိန်း", |
| "ဇင်ဝိုင်း", |
| "လူမင်း", |
| "ခိုင်သင်းကြည်", |
| "စိုးမြတ်သူဇာ", |
| "ထက်ထက်မိုးဦး", |
| "မေသန်းနု", |
| "နန္ဒာလှိုင်", |
| "စိုးပြည့်သဇင်", |
| "ဝိုင်းစုခိုင်သိန်း", |
| "ချစ်သုဝေ", |
| "မေကဗျာ", |
| "မိုးပြည့်ပြည့်မောင်", |
| "ယုသန္တာတင်", |
| "သဲစုညိမ်း", |
| "မေသက်ခိုင်", |
| "ရွှေသမီး", |
| "စန္ဒီမြင့်လွင်", |
| "နန်းမြတ်ဖြိုးသင်း", |
| "အဆိုတော်အမျိုးသား", |
| "အဆိုတော်အမျိုးသမီး", |
| ] |
|
|
| VOICE_CACHE = [] |
| VOICE_CACHE_TIME = 0 |
| VOICE_CACHE_TTL = 3600 |
|
|
| VOICE_LOCK = threading.Lock() |
| JOB_SEMAPHORE = threading.Semaphore(MAX_CONCURRENT_JOBS) |
|
|
|
|
| |
| |
| |
|
|
| async def fetch_voices_async(): |
| return await edge_tts.list_voices() |
|
|
|
|
| def get_fallback_voices(): |
| return [ |
| { |
| "ShortName": "my-MM-NilarNeural", |
| "FriendlyName": "Nilar", |
| "Gender": "Female", |
| "Locale": "my-MM", |
| }, |
| { |
| "ShortName": "my-MM-ThihaNeural", |
| "FriendlyName": "Thiha", |
| "Gender": "Male", |
| "Locale": "my-MM", |
| }, |
| { |
| "ShortName": "it-IT-GiuseppeMultilingualNeural", |
| "FriendlyName": "Giuseppe Multilingual", |
| "Gender": "Male", |
| "Locale": "it-IT", |
| }, |
| ] |
|
|
|
|
| def get_voices(force_refresh=False): |
| global VOICE_CACHE, VOICE_CACHE_TIME |
|
|
| now = time.time() |
|
|
| with VOICE_LOCK: |
| if ( |
| VOICE_CACHE |
| and not force_refresh |
| and now - VOICE_CACHE_TIME < VOICE_CACHE_TTL |
| ): |
| return VOICE_CACHE |
|
|
| try: |
| voices = asyncio.run(fetch_voices_async()) |
| selected = [] |
|
|
| for voice in voices: |
| short_name = voice.get("ShortName", "") |
| if not short_name: |
| continue |
|
|
| is_pinned = short_name in PINNED_VOICES |
| is_multilingual = "MultilingualNeural" in short_name |
|
|
| if is_pinned or is_multilingual: |
| selected.append(voice) |
|
|
| available_ids = { |
| voice["ShortName"] for voice in selected |
| } |
|
|
| fallback_map = { |
| voice["ShortName"]: voice |
| for voice in get_fallback_voices() |
| } |
|
|
| for short_name in PINNED_VOICES: |
| if short_name not in available_ids: |
| selected.append(fallback_map[short_name]) |
|
|
| selected.sort( |
| key=lambda voice: ( |
| 0 |
| if voice["ShortName"] == DEFAULT_VOICE_ID |
| else 1 |
| if voice["ShortName"] in PINNED_VOICES |
| else 2, |
| voice.get("Locale", ""), |
| voice.get("ShortName", ""), |
| ) |
| ) |
|
|
| VOICE_CACHE = selected |
| VOICE_CACHE_TIME = now |
|
|
| logger.info("Loaded %d selected voices", len(VOICE_CACHE)) |
| return VOICE_CACHE |
|
|
| except Exception: |
| logger.exception("Unable to retrieve Edge TTS voices") |
|
|
| if VOICE_CACHE: |
| return VOICE_CACHE |
|
|
| VOICE_CACHE = get_fallback_voices() |
| VOICE_CACHE_TIME = now |
| return VOICE_CACHE |
|
|
|
|
| def build_voice_display_map(): |
| """Create Myanmar display names; do not show original Voice IDs.""" |
|
|
| voices = get_voices() |
| result = {} |
| used_names = set() |
| alias_index = 0 |
|
|
| for voice in voices: |
| short_name = voice.get("ShortName", "") |
| if not short_name: |
| continue |
|
|
| if short_name in VOICE_DISPLAY_NAMES: |
| label = VOICE_DISPLAY_NAMES[short_name] |
| else: |
| label = None |
|
|
| while alias_index < len(VOICE_ALIAS_NAMES): |
| candidate = VOICE_ALIAS_NAMES[alias_index] |
| alias_index += 1 |
|
|
| if candidate not in used_names: |
| label = candidate |
| break |
|
|
| if label is None: |
| label = f"အသံဇာတ်ကောင် {len(result) + 1}" |
|
|
| base_label = label |
| suffix = 2 |
|
|
| while label in used_names: |
| label = f"{base_label} {suffix}" |
| suffix += 1 |
|
|
| used_names.add(label) |
| result[label] = short_name |
|
|
| return result |
|
|
|
|
| def voice_label(voice): |
| short_name = voice.get("ShortName", "") |
| return VOICE_DISPLAY_NAMES.get(short_name, "") |
|
|
|
|
| def get_voice_map(): |
| return build_voice_display_map() |
|
|
|
|
| def get_voice_choices(): |
| return list(get_voice_map().keys()) |
|
|
|
|
| def get_default_voice_label(): |
| voice_map = get_voice_map() |
|
|
| for label, voice_id in voice_map.items(): |
| if voice_id == DEFAULT_VOICE_ID: |
| return label |
|
|
| return next(iter(voice_map), None) |
|
|
|
|
| def refresh_voices(): |
| get_voices(force_refresh=True) |
| choices = get_voice_choices() |
| default_label = get_default_voice_label() |
|
|
| return ( |
| gr.update(choices=choices, value=default_label), |
| f"Voice List ပြန်ရယူပြီးပါပြီ။ Voice {len(choices)} ခု ရရှိထားသည်။", |
| ) |
|
|
|
|
| |
| |
| |
|
|
| def normalize_script(text): |
| """Normalize whitespace without changing word order.""" |
|
|
| text = (text or "").replace("\r\n", "\n").replace("\r", "\n") |
| lines = [] |
|
|
| for line in text.split("\n"): |
| line = re.sub(r"[ \t]+", " ", line).strip() |
| if line: |
| lines.append(line) |
|
|
| return "\n".join(lines) |
|
|
|
|
| def split_into_sentences(text): |
| """Split at sentence boundaries where possible.""" |
|
|
| parts = re.split(r"(?<=[.!?။！？၊])\s+", text) |
| return [part.strip() for part in parts if part.strip()] |
|
|
|
|
| def split_long_text(text, max_chars=MAX_CHUNK_CHARACTERS): |
| """Fallback split at whitespace without dropping words.""" |
|
|
| words = text.split() |
| chunks = [] |
| current_words = [] |
| current_length = 0 |
|
|
| for word in words: |
| added_length = len(word) |
| if current_words: |
| added_length += 1 |
|
|
| if current_words and current_length + added_length > max_chars: |
| chunks.append(" ".join(current_words)) |
| current_words = [word] |
| current_length = len(word) |
| else: |
| current_words.append(word) |
| current_length += added_length |
|
|
| if current_words: |
| chunks.append(" ".join(current_words)) |
|
|
| return chunks |
|
|
|
|
| def verify_text_integrity(original, subtitle_texts): |
| """Verify whitespace-separated tokens are preserved in order.""" |
|
|
| original_words = original.split() |
| output_words = [] |
|
|
| for subtitle in subtitle_texts: |
| output_words.extend(subtitle.split()) |
|
|
| if original_words != output_words: |
| raise RuntimeError( |
| "စာသားစစ်ဆေးမှု မအောင်မြင်ပါ။ " |
| "စာသားအစဉ် သို့မဟုတ် စကားလုံး ပျောက်နေသည်။" |
| ) |
|
|
| return True |
|
|
|
|
| def split_for_tts(text): |
| """ |
| TEST MODE: split after every Myanmar sentence-ending mark '။'. |
| Each sentence is synthesized as its own Edge TTS chunk. |
| Text order and punctuation are preserved. Long sections without |
| '။' use whitespace splitting as a safety fallback. |
| """ |
|
|
| text = (text or "").strip() |
| if not text: |
| return [] |
|
|
| |
| |
| parts = re.findall(r"[^။]*။|[^။]+$", text, flags=re.DOTALL) |
|
|
| chunks = [] |
| for part in parts: |
| part = part.strip() |
| if not part: |
| continue |
|
|
| if len(part) > MAX_CHUNK_CHARACTERS: |
| chunks.extend( |
| split_long_text(part, MAX_CHUNK_CHARACTERS) |
| ) |
| else: |
| chunks.append(part) |
|
|
| verify_text_integrity(text, chunks) |
| return chunks |
|
|
| |
| |
| |
|
|
| def split_for_srt(text, words_per_subtitle=WORDS_PER_SUBTITLE): |
| """Split SRT text at whitespace boundaries, around 20 tokens.""" |
|
|
| words = text.split() |
| result = [] |
|
|
| for index in range(0, len(words), words_per_subtitle): |
| result.append(" ".join(words[index:index + words_per_subtitle])) |
|
|
| verify_text_integrity(text, result) |
| return result |
|
|
|
|
| |
| |
| |
|
|
| def speed_to_rate(speed): |
| """ |
| Edge TTS rate format must include an explicit sign. |
| In particular, zero must be '+0%', never '0%'. |
| """ |
|
|
| speed = int(speed) |
| speed = max(-50, min(100, speed)) |
|
|
| if speed >= 0: |
| return f"+{speed}%" |
|
|
| return f"{speed}%" |
|
|
|
|
| |
| |
| |
|
|
| async def synthesize_async(text, voice, rate, output_path): |
| communicate = edge_tts.Communicate( |
| text=text, |
| voice=voice, |
| rate=rate, |
| ) |
|
|
| await communicate.save(output_path) |
|
|
| if ( |
| not os.path.isfile(output_path) |
| or os.path.getsize(output_path) < MIN_AUDIO_BYTES |
| ): |
| raise RuntimeError( |
| "Edge TTS က အသံဖိုင်အလွတ် ပြန်ပေးထားသည်။" |
| ) |
|
|
|
|
| def synthesize_chunk(text, voice, rate, output_path): |
| asyncio.run( |
| synthesize_async(text, voice, rate, output_path) |
| ) |
|
|
|
|
| |
| |
| |
| |
|
|
| def load_and_trim_audio_part(part_path): |
| """Load audio without trimming the beginning or ending.""" |
|
|
| return AudioSegment.from_file(part_path, format="mp3") |
|
|
|
|
| def combine_audio_parts(audio_parts, output_path): |
| """Join audio parts in order without trimming them.""" |
|
|
| if not audio_parts: |
| raise RuntimeError("ပေါင်းစပ်ရန် Audio အပိုင်း မရှိပါ။") |
|
|
| combined_audio = AudioSegment.empty() |
|
|
| for part_path in audio_parts: |
| segment = load_and_trim_audio_part(part_path) |
| combined_audio += segment |
|
|
| if len(combined_audio) <= 0: |
| raise RuntimeError( |
| "ပေါင်းစပ်ထားသော Audio ဖိုင်တွင် အသံမရှိပါ။" |
| ) |
|
|
| combined_audio.export( |
| output_path, |
| format="mp3", |
| bitrate="192k", |
| ) |
|
|
| if ( |
| not os.path.isfile(output_path) |
| or os.path.getsize(output_path) < MIN_AUDIO_BYTES |
| ): |
| raise RuntimeError( |
| "ပေါင်းစပ်ထားသော Audio ဖိုင်ကို ထုတ်မရပါ။" |
| ) |
|
|
| return len(combined_audio) |
|
|
|
|
| |
| |
| |
|
|
| def format_srt_time(milliseconds): |
| milliseconds = max(0, int(milliseconds)) |
|
|
| hours = milliseconds // 3600000 |
| milliseconds %= 3600000 |
|
|
| minutes = milliseconds // 60000 |
| milliseconds %= 60000 |
|
|
| seconds = milliseconds // 1000 |
| milliseconds %= 1000 |
|
|
| return ( |
| f"{hours:02d}:" |
| f"{minutes:02d}:" |
| f"{seconds:02d}," |
| f"{milliseconds:03d}" |
| ) |
|
|
|
|
| def make_srt_entries(subtitle_texts, total_duration_ms): |
| """Estimate timing proportionally by subtitle text length.""" |
|
|
| if not subtitle_texts: |
| return [] |
|
|
| weights = [ |
| max(1, len(text.replace(" ", ""))) |
| for text in subtitle_texts |
| ] |
|
|
| total_weight = sum(weights) |
| if total_weight <= 0: |
| return [] |
|
|
| entries = [] |
| elapsed = 0 |
|
|
| for index, (text, weight) in enumerate( |
| zip(subtitle_texts, weights) |
| ): |
| start = elapsed |
|
|
| if index == len(subtitle_texts) - 1: |
| end = total_duration_ms |
| else: |
| duration = round(total_duration_ms * weight / total_weight) |
| elapsed += duration |
| end = elapsed |
|
|
| entries.append( |
| { |
| "text": text, |
| "start": start, |
| "end": end, |
| } |
| ) |
|
|
| return entries |
|
|
|
|
| def build_srt(entries): |
| blocks = [] |
|
|
| for index, entry in enumerate(entries, start=1): |
| start = format_srt_time(entry["start"]) |
| end = format_srt_time(entry["end"]) |
|
|
| blocks.append( |
| f"{index}\n" |
| f"{start} --> {end}\n" |
| f"{entry['text']}" |
| ) |
|
|
| return "\n\n".join(blocks) + "\n" |
|
|
|
|
| |
| |
| |
|
|
| def generate_audio_and_srt( |
| script, |
| selected_voice, |
| speed, |
| progress=gr.Progress(), |
| ): |
| script = normalize_script(script) |
|
|
| if not script: |
| raise gr.Error("Script ထည့်ပေးပါ။") |
|
|
| voice_map = get_voice_map() |
| voice = voice_map.get(selected_voice) |
|
|
| if not voice: |
| raise gr.Error("Voice ကို ပြန်ရွေးပေးပါ။") |
|
|
| voice_display = selected_voice |
| rate = speed_to_rate(speed) |
| tts_chunks = split_for_tts(script) |
|
|
| if not tts_chunks: |
| raise gr.Error("အသံထုတ်ရန် Script မရှိပါ။") |
|
|
| work_dir = tempfile.mkdtemp(prefix="job_", dir=OUTPUT_DIR) |
| audio_path = os.path.join(work_dir, "output.mp3") |
| srt_path = os.path.join(work_dir, "output.srt") |
|
|
| try: |
| with JOB_SEMAPHORE: |
| audio_parts = [] |
| total_chunks = len(tts_chunks) |
|
|
| for index, chunk in enumerate(tts_chunks): |
| progress( |
| 0.75 * index / total_chunks, |
| desc=f"Audio ထုတ်နေသည် {index + 1}/{total_chunks}", |
| ) |
|
|
| part_path = os.path.join( |
| work_dir, |
| f"part_{index:05d}.mp3", |
| ) |
|
|
| synthesize_chunk( |
| chunk, |
| voice, |
| rate, |
| part_path, |
| ) |
| audio_parts.append(part_path) |
|
|
| progress(0.80, desc="Audio ပေါင်းနေသည်...") |
|
|
| duration_ms = combine_audio_parts( |
| audio_parts, |
| audio_path, |
| ) |
|
|
| progress(0.90, desc="SRT ပြုလုပ်နေသည်...") |
|
|
| subtitle_texts = split_for_srt(script) |
| verify_text_integrity(script, subtitle_texts) |
|
|
| entries = make_srt_entries( |
| subtitle_texts, |
| duration_ms, |
| ) |
|
|
| with open( |
| srt_path, |
| "w", |
| encoding="utf-8-sig", |
| newline="\n", |
| ) as file: |
| file.write(build_srt(entries)) |
|
|
| progress(1.0, desc="ပြီးပါပြီ!") |
|
|
| logger.info( |
| "Generation complete: voice=%s, rate=%s, chars=%d, chunks=%d", |
| voice, |
| rate, |
| len(script), |
| total_chunks, |
| ) |
|
|
| return ( |
| audio_path, |
| srt_path, |
| ( |
| "အောင်မြင်ပါသည်!\n" |
| f"Voice Character: {voice_display}\n" |
| f"Speed: {rate}\n" |
| f"Script: {len(script)} characters\n" |
| f"Audio chunks: {total_chunks}\n" |
| f"SRT lines: {len(subtitle_texts)}" |
| ), |
| ) |
|
|
| except Exception as exc: |
| logger.exception("Audio/SRT generation failed") |
| shutil.rmtree(work_dir, ignore_errors=True) |
|
|
| raise gr.Error( |
| f"Error: {type(exc).__name__}: {str(exc)[:400]}" |
| ) |
|
|
|
|
| |
| |
| |
|
|
| initial_choices = get_voice_choices() |
| initial_voice = get_default_voice_label() |
|
|
| with gr.Blocks( |
| title=APP_TITLE, |
| theme=gr.themes.Soft(), |
| ) as demo: |
| gr.Markdown( |
| """ |
| # 🎙️ Myanmar Multilingual TTS Studio |
| |
| မြန်မာအမည်များဖြင့် Voice ရွေးချယ်ပြီး |
| MP3 Audio နှင့် SRT Subtitle ကို တစ်ခါတည်း ထုတ်ယူပါ။ |
| """ |
| ) |
|
|
| with gr.Row(): |
| voice_input = gr.Dropdown( |
| choices=initial_choices, |
| value=initial_voice, |
| label="Voice Character", |
| interactive=True, |
| scale=5, |
| ) |
|
|
| refresh_button = gr.Button( |
| "🔄 Voice List ပြန်ယူမယ်", |
| scale=2, |
| ) |
|
|
| voice_status = gr.Markdown( |
| "နေတိုးကို Default Voice အဖြစ် ရွေးထားသည်။" |
| ) |
|
|
| script_input = gr.Textbox( |
| label="Script", |
| placeholder="အသံထုတ်လိုသော စာသားကို ဒီနေရာမှာ ထည့်ပါ...", |
| lines=12, |
| max_lines=30, |
| ) |
|
|
| speed_input = gr.Slider( |
| minimum=-50, |
| maximum=100, |
| value=DEFAULT_SPEED, |
| step=5, |
| label="အသံ Speed (%)", |
| ) |
|
|
| gr.Markdown( |
| """ |
| - Default Voice: နေတိုး |
| - Default Speed: +30% |
| - SRT: Space ခြားထားသော စကားလုံး ၂၀ ဝန်းကျင် |
| - Speed: -50% မှ +100% |
| """ |
| ) |
|
|
| generate_button = gr.Button( |
| "🎙️ Audio + SRT ထုတ်မယ်", |
| variant="primary", |
| size="lg", |
| ) |
|
|
| status_output = gr.Textbox( |
| label="အခြေအနေ", |
| interactive=False, |
| lines=5, |
| ) |
|
|
| audio_output = gr.Audio( |
| label="🎧 MP3 Audio", |
| type="filepath", |
| ) |
|
|
| srt_output = gr.File( |
| label="📝 SRT Subtitle Download", |
| ) |
|
|
| refresh_button.click( |
| fn=refresh_voices, |
| inputs=[], |
| outputs=[ |
| voice_input, |
| voice_status, |
| ], |
| ) |
|
|
| generate_button.click( |
| fn=generate_audio_and_srt, |
| inputs=[ |
| script_input, |
| voice_input, |
| speed_input, |
| ], |
| outputs=[ |
| audio_output, |
| srt_output, |
| status_output, |
| ], |
| api_name="generate", |
| ) |
|
|
|
|
| if __name__ == "__main__": |
| demo.queue( |
| max_size=10, |
| default_concurrency_limit=MAX_CONCURRENT_JOBS, |
| ).launch( |
| server_name="0.0.0.0", |
| server_port=int(os.environ.get("PORT", 7860)), |
| ) |
|
|
