تحويل النص إلى كلام باللغة العربية في Python: دليل عملي خطوة بخطوة بالشيفرة البرمجية

pip install requests
   
  # macOS or Linux
  export MUNSIT_API_KEY="your_key_here"
   
  # Windows PowerShell
  $env:MUNSIT_API_KEY="your_key_here"

 step1_first_call.py
  import os
  import requests
   
  API_KEY = os.environ["MUNSIT_API_KEY"]
  BASE_URL = os.environ.get("MUNSIT_BASE_URL", "https://api.munsit.com/api/v1")
  MODEL_ID = "faseeh-v1-preview"
   
  payload = {
      "voice_id": "YOUR_VOICE_ID",          # copy one from the voices list (Step 2)
      "text": "مرحبا بك في منصت",            # "Welcome to Munsit"
      "stability": 0.5,                      # required, 0.0 to 1.0
      "speed": 1.0,                          # optional, 0.7 to 1.2
      "sample_rate": 24000,                  # optional, 8000 to 48000 Hz
  }
   
  response = requests.post(
      f"{BASE_URL}/text-to-speech/{MODEL_ID}",
      headers={"x-api-key": API_KEY, "Content-Type": "application/json"},
      json=payload,
      timeout=60,
  )
  response.raise_for_status()
   
  with open("speech.wav", "wb") as f:
      f.write(response.content)
   
  print("Saved speech.wav,", len(response.content), "bytes")

 step2_list_voices.py
  import os
  import requests
   
  API_KEY = os.environ["MUNSIT_API_KEY"]
  BASE_URL = os.environ.get("MUNSIT_BASE_URL", "https://api.munsit.com/api/v1")
   
   
  def list_voices(dialect=None, tags=None):
      params = {}
      if dialect:
          params["dialect"] = dialect      # for example "najdi", "emirati", "fusha"
      if tags:
          params["tags"] = tags            # for example "narration" or "conversational"
      r = requests.get(f"{BASE_URL}/voices", headers={"x-api-key": API_KEY},
                       params=params, timeout=30)
      r.raise_for_status()
      data = r.json()
      # Accept either a plain list or a list wrapped in an object
      if isinstance(data, dict):
          data = data.get("data") or data.get("voices") or data.get("items") or []
      return data
   
   
  for v in list_voices(dialect="najdi"):
      print(v.get("voice_id") or v.get("id"), "|", v.get("name"), "|",
            v.get("dialect"), "|", v.get("gender"))

munsit_tts.py
  """A small, reusable helper for Munsit Arabic text to speech."""
  import os
  import re
  import time
  import requests
   
  API_KEY = os.environ["MUNSIT_API_KEY"]
  BASE_URL = os.environ.get("MUNSIT_BASE_URL", "https://api.munsit.com/api/v1")
  MODEL_ID = "faseeh-v1-preview"
  HEADERS = {"x-api-key": API_KEY, "Content-Type": "application/json"}
   
  BREAK_TAG = re.compile(r'<break\s+time="([^"]*)"\s*/>')
  VALID_TIME = re.compile(r"^(\d+(\.\d+)?)(ms|s)$")
   
   
  def check_pauses(text):
      """Warn about pause tags the API would silently ignore or clamp."""
      problems = []
      tags = BREAK_TAG.findall(text)
      total = 0.0
      for value in tags:
          m = VALID_TIME.match(value)
          if not m:
              problems.append(f'"{value}" has no unit (s or ms) and will be ignored')
              continue
          seconds = float(m.group(1)) / (1000 if m.group(3) == "ms" else 1)
          if seconds > 3:
              problems.append(f'"{value}" is longer than 3s and will be clamped to 3s')
          total += min(seconds, 3)
      if len(tags) > 20:
          problems.append(f"{len(tags)} pause tags; over 20 may be dropped")
      if total > 30:
          problems.append(f"{total:.1f}s of pauses; about 30s is the maximum")
      return problems
   
   
  def synthesize(text, voice_id, stability=0.5, speed=1.0, sample_rate=24000,
                 dialect=None, retries=3):
      """Return WAV bytes, retrying on rate limits and server errors."""
      for warning in check_pauses(text):
          print("Pause tag warning:", warning)
      payload = {"voice_id": voice_id, "text": text, "stability": stability,
                 "speed": speed, "sample_rate": sample_rate}
      if dialect:
          payload["dialect"] = dialect          # "auto", "emirati" or "fusha"
      for attempt in range(retries + 1):
          r = requests.post(f"{BASE_URL}/text-to-speech/{MODEL_ID}",
                            headers=HEADERS, json=payload, timeout=60)
          if r.status_code == 200:
              return r.content
          if r.status_code in (429, 500, 502, 503, 504) and attempt < retries:
              wait = 2 ** attempt
              print(f"Got {r.status_code}, retrying in {wait}s")
              time.sleep(wait)
              continue
          raise RuntimeError(f"Synthesis failed ({r.status_code}): {r.text[:300]}")
   
   
  def diacritize(text):
      """Add tashkeel (vowel marks) to formal Arabic text."""
      r = requests.post(f"{BASE_URL}/tashkil/diacritize", headers=HEADERS,
                        json={"text": text}, timeout=60)
      r.raise_for_status()
      return r.json()["diacritized_text"]

 step4_tashkil.py
  from munsit_tts import diacritize, synthesize
   
  formal = "يسرنا ان نرحب بكم في المعرض الدولي للكتاب"
  marked = diacritize(formal)
  print(marked)
   
  audio = synthesize(marked, voice_id="YOUR_FUSHA_VOICE_ID", dialect="fusha")
  with open("welcome_fusha.wav", "wb") as f:
      f.write(audio)

step5_batch.py
  import csv
  from concurrent.futures import ThreadPoolExecutor
  from pathlib import Path
  from munsit_tts import synthesize
   
  VOICE_ID = "YOUR_VOICE_ID"
  OUT = Path("audio")
  OUT.mkdir(exist_ok=True)
   
   
  def render(row):
      target = OUT / f"{row['id']}.wav"
      if target.exists():                      # skip files already made
          return f"{row['id']}: skipped"
      target.write_bytes(synthesize(row["text"], voice_id=VOICE_ID))
      return f"{row['id']}: done"
   
   
  with open("scripts.csv", encoding="utf-8") as f:
      rows = list(csv.DictReader(f))           # columns: id,text
   
  # Keep concurrency low and check the rate limits page for your plan
  with ThreadPoolExecutor(max_workers=3) as pool:
      for result in pool.map(render, rows):
          print(result)

 step6_long_text.py
  import io
  import re
  import wave
  from munsit_tts import synthesize
   
  SENTENCE_END = re.compile(r"(?<=[.!؟?؛])\s+")
   
   
  def split_text(text, max_chars=800):
      """Split on Arabic and Latin sentence endings, keeping chunks under max_chars."""
      chunks, current = [], ""
      for sentence in SENTENCE_END.split(text.strip()):
          if current and len(current) + len(sentence) + 1 > max_chars:
              chunks.append(current)
              current = sentence
          else:
              current = f"{current} {sentence}".strip()
      if current:
          chunks.append(current)
      return chunks
   
   
  def join_wavs(wav_list, path):
      """Concatenate WAV files that share the same format into one file."""
      with wave.open(path, "wb") as out:
          for i, data in enumerate(wav_list):
              with wave.open(io.BytesIO(data), "rb") as w:
                  if i == 0:
                      out.setparams(w.getparams())
                  out.writeframes(w.readframes(w.getnframes()))
   
   
  article = open("article.txt", encoding="utf-8").read()
  parts = [synthesize(chunk, voice_id="YOUR_VOICE_ID") for chunk in split_text(article)]
  join_wavs(parts, "article.wav")
  print(f"Joined {len(parts)} parts into article.wav")

step7_streaming.py
  import os
  import time
  import wave
  import requests

  API_KEY = os.environ["MUNSIT_API_KEY"]
  BASE_URL = os.environ.get("MUNSIT_BASE_URL", "https://api.munsit.com/api/v1")
  SAMPLE_RATE = 48000                     # 48 kHz is recommended for real time use

  payload = {
      "voice_id": "YOUR_VOICE_ID",
      "text": "أهلا، كيف أقدر أساعدك اليوم؟",
      "stability": 0.5,
      "streaming": True,                  # returns PCM16 chunks as they are generated
      "sample_rate": SAMPLE_RATE,
  }

  start = time.perf_counter()
  first_chunk_at = None
  with requests.post(f"{BASE_URL}/text-to-speech/faseeh-v1-preview",
                     headers={"x-api-key": API_KEY}, json=payload,
                     stream=True, timeout=60) as r:
      r.raise_for_status()
      # Assumes mono 16 bit PCM: confirm channel count in the streaming docs
      with wave.open("stream.wav", "wb") as out:
          out.setnchannels(1)
          out.setsampwidth(2)
          out.setframerate(SAMPLE_RATE)
          for chunk in r.iter_content(chunk_size=4096):
              if not chunk:
                  continue
              if first_chunk_at is None:
                  first_chunk_at = time.perf_counter() - start
              out.writeframes(chunk)          # or send chunk straight to a player

  print(f"First audio after {first_chunk_at * 1000:.0f} ms; saved stream.wav")