Asenkron Metin Okuma API'si: Uzun Biçimli Ses Üretimini Yönetme

Senkronize TTS, kısa metinler için işe yarar. 10.000 kelimelik bir bölüm, 30 dakikalık bir podcast senaryosu veya 500 ürün açıklamasından oluşan bir toplu iş için asenkron desenlere ihtiyacınız vardır.

Senkronize ve Asenkronize TTS: Ne Zaman Hangisi Uygulanır?

Senkronize TTS (standart istek/yanıt):

• ~5.000 karakterin altındaki metin
• Kullanıcının sonucu beklediği etkileşimli uygulamalar
• Tamamlanma süresinin önemli olmadığı basit betikler

Asenkronize TTS (arka plan işleri):

• İstek başına karakter sınırlarını aşan uzun belgeler
• Birçok dosyanın toplu üretimi
• Engellemeyen arka plan işleme (kullanıcı beklemez)
• Dağıtılmış iş yükleri

Çoğu TTS API'si teknik olarak senkronizedir (bir istek yaparsınız, ses yanıtını beklersiniz). Pratikte "Asenkronize TTS", API'nin kendisinin asenkronize olduğu anlamına gelmez, kodunuzda birden çok isteği eşzamanlı olarak işlemek için asenkron desenler kullandığınız anlamına gelir.

AWS Polly bir istisnadır — işleri sıraya alan ve S3'e çıktı veren yerel bir asenkron API'ye (StartSpeechSynthesisTask) sahiptir. Diğer çoğu sağlayıcıda bu yoktur.

Desen 1: Parçalı Sıralı Üretim

Uzun belgeler için en basit yaklaşım: metni parçalara ayırın, her birini üretin, birleştirin.

import requests
import os
import re
from pathlib import Path

def split_text(text: str, max_chars: int = 3000) -> list[str]:
    """Kelimeleri ortadan kesmemek için cümle sınırlarında bölün."""
    sentences = re.split(r'(?<=[.!?])\s+', text.strip())
    chunks, current = [], ""

    for sentence in sentences:
        if len(current) + len(sentence) + 1 > max_chars:
            if current:
                chunks.append(current.strip())
            current = sentence
        else:
            current = (current + " " + sentence).strip()

    if current:
        chunks.append(current)

    return chunks

def generate_chunk(text: str, chunk_index: int, total: int) -> bytes:
    """Bir parçayı hız sınırı durumunda yeniden deneme ile üret."""
    import time

    for attempt in range(3):
        response = requests.post(
            "https://api.speekoapp.com/v1/tts",
            headers={"X-API-Key": os.environ["SPEEKO_API_KEY"]},
            json={"text": text, "voice": "en-US-1", "format": "mp3"},
            timeout=60,
        )

        if response.status_code == 429:
            wait = 2 ** attempt
            print(f"  Hız sınırı aşıldı, {wait}s bekleniyor...")
            time.sleep(wait)
            continue

        response.raise_for_status()
        print(f"  Parça {chunk_index + 1}/{total}: {len(text)} karakter → {len(response.content):,} bayt")
        return response.content

    raise RuntimeError(f"Parça {chunk_index} 3 denemeden sonra başarısız oldu")

def long_document_to_audio(text: str, output_file: str):
    chunks = split_text(text)
    print(f"{len(chunks)} parçaya bölündü")

    audio_parts = []
    for i, chunk in enumerate(chunks):
        audio = generate_chunk(chunk, i, len(chunks))
        audio_parts.append(audio)

    # Tüm MP3 parçalarını birleştir
    with open(output_file, "wb") as f:
        for part in audio_parts:
            f.write(part)

    total_size = sum(len(p) for p in audio_parts)
    print(f"Bitti: {output_file} ({total_size:,} bayt)")

Desen 2: asyncio ile Eşzamanlı Üretim

Daha hızlı toplu işleme için, parçaları paralel olarak üretin:

import asyncio
import aiohttp
import os
import re
from pathlib import Path

async def generate_chunk_async(
    session: aiohttp.ClientSession,
    text: str,
    chunk_index: int,
    semaphore: asyncio.Semaphore,
) -> tuple[int, bytes]:
    """Bir parçayı asenkron olarak üret, eşzamanlılık sınırına uy."""

    async with semaphore:  # Eşzamanlı istekleri sınırla
        for attempt in range(3):
            async with session.post(
                "https://api.speekoapp.com/v1/tts",
                headers={"X-API-Key": os.environ["SPEEKO_API_KEY"]},
                json={"text": text, "voice": "en-US-1", "format": "mp3"},
            ) as response:
                if response.status == 429:
                    wait = 2 ** attempt
                    await asyncio.sleep(wait)
                    continue

                response.raise_for_status()
                audio = await response.read()
                print(f"Parça {chunk_index}: bitti ({len(audio):,} bayt)")
                return chunk_index, audio

        raise RuntimeError(f"Parça {chunk_index} başarısız oldu")

async def batch_tts_async(texts: list[str], max_concurrent: int = 5) -> list[bytes]:
    """Birden çok metin için TTS'yi eşzamanlı olarak üret."""

    semaphore = asyncio.Semaphore(max_concurrent)

    async with aiohttp.ClientSession() as session:
        tasks = [
            generate_chunk_async(session, text, i, semaphore)
            for i, text in enumerate(texts)
        ]
        results = await asyncio.gather(*tasks)

    # Sırayı korumak için indekse göre sırala
    results.sort(key=lambda x: x[0])
    return [audio for _, audio in results]

# Kullanım
async def main():
    texts = [
        "Bölüm 1: Konuya giriş.",
        "Bölüm 2: Temel kavramlar ayrıntılı olarak açıklanmıştır.",
        "Bölüm 3: Pratik uygulamalar ve örnekler.",
    ]

    audio_parts = await batch_tts_async(texts, max_concurrent=3)

    with open("book.mp3", "wb") as f:
        for part in audio_parts:
            f.write(part)

    print(f"{len(audio_parts)} parça üretildi")

asyncio.run(main())

max_concurrent=3 sınırı, API'yi aşırı yüklemeyi önler. Hız sınırı katmanınıza göre ayarlayın.

Desen 3: Celery ile Arka Plan İşi

Kullanıcının bir istek gönderdiği ve ses hazır olduğunda bir bildirim aldığı web uygulamaları için:

from celery import Celery
import requests
import os

celery = Celery('tts', broker=os.environ['REDIS_URL'])

@celery.task(bind=True, max_retries=3)
def generate_audio_task(self, job_id: str, text: str, voice: str = "en-US-1"):
    """Arka plan TTS görevi."""
    try:
        # Gerekirse böl
        chunks = split_text(text)
        audio_parts = []

        for chunk in chunks:
            response = requests.post(
                "https://api.speekoapp.com/v1/tts",
                headers={"X-API-Key": os.environ["SPEEKO_API_KEY"]},
                json={"text": chunk, "voice": voice, "format": "mp3"},
                timeout=60,
            )
            response.raise_for_status()
            audio_parts.append(response.content)

        # Sonucu birleştir ve sakla
        audio = b"".join(audio_parts)

        # Veritabanınızda/dosya depolamanızda saklayın
        save_audio(job_id, audio)
        update_job_status(job_id, "completed")

    except requests.HTTPError as exc:
        if exc.response.status_code == 429:
            # Hız sınırı aşıldı — gecikmeyle yeniden dene
            raise self.retry(countdown=30, exc=exc)
        raise

# Web görünümünüzde:
# job_id = str(uuid.uuid4())
# generate_audio_task.delay(job_id, text)
# return {"job_id": job_id, "status": "processing"}

Parçaları Temiz Bir Şekilde Yeniden Birleştirme

Basit bayt birleştirme MP3 için işe yarar ancak sınırlarda ses aksaklıklarına neden olabilir. Daha temiz birleştirmeler için ffmpeg kullanın:

import subprocess
import tempfile
import os

def concat_with_ffmpeg(audio_chunks: list[bytes], output_file: str):
    """MP3 parçalarını temiz birleştirmeler için ffmpeg kullanarak birleştir."""

    # Parçaları geçici dosyalara yaz
    temp_files = []
    with tempfile.TemporaryDirectory() as tmpdir:
        for i, chunk in enumerate(audio_chunks):
            path = os.path.join(tmpdir, f"chunk_{i:04d}.mp3")
            with open(path, "wb") as f:
                f.write(chunk)
            temp_files.append(path)

        # Birleştirme listesini yaz
        list_file = os.path.join(tmpdir, "list.txt")
        with open(list_file, "w") as f:
            for path in temp_files:
                f.write(f"file '{path}'\n")

        # Birleştir
        subprocess.run(
            ["ffmpeg", "-f", "concat", "-safe", "0",
             "-i", list_file, "-c", "copy", output_file],
            check=True,
            capture_output=True,
        )

Uzun İşler için Hata Kurtarma

Yarıda başarısız olabilecek işler için:

from pathlib import Path
import json

def resumable_batch_tts(texts: list[str], job_dir: str):
    """Metinler için ses üret, zaten tamamlanmış parçaları atla."""

    output_dir = Path(job_dir)
    output_dir.mkdir(exist_ok=True)
    progress_file = output_dir / "progress.json"

    # Mevcut ilerlemeyi yükle
    progress = {}
    if progress_file.exists():
        progress = json.loads(progress_file.read_text())

    for i, text in enumerate(texts):
        chunk_key = str(i)

        if chunk_key in progress:
            print(f"Parça {i}: zaten bitti, atlanıyor")
            continue

        chunk_file = output_dir / f"chunk_{i:04d}.mp3"
        response = requests.post(
            "https://api.speekoapp.com/v1/tts",
            headers={"X-API-Key": os.environ["SPEEKO_API_KEY"]},
            json={"text": text, "voice": "en-US-1", "format": "mp3"},
        )
        response.raise_for_status()

        chunk_file.write_bytes(response.content)
        progress[chunk_key] = str(chunk_file)
        progress_file.write_text(json.dumps(progress))
        print(f"Parça {i}: bitti")

    # Tamamlanmış tüm parçaları sırayla birleştir
    chunks = [progress[str(i)] for i in range(len(texts))]
    concat_with_ffmpeg([Path(p).read_bytes() for p in chunks], str(output_dir / "final.mp3"))

Başlarken

speekoapp.com/register adresinden kaydolun — 5$ ücretsiz kredi, kart gerekmez.

İlgili: Metni Programlı Olarak Konuşmaya Dönüştürme, Blogdan Podcast'e Otomasyon, Yapay Zeka Sesli Kitap Oluşturma.