kie AI로 구글 Gemini TTS 쓰는 법 | 가입·충전·API 호출부터 목소리 70개 샘플까지

import json, os, struct, time, urllib.request, urllib.error

KEY = os.environ["KIE_API_KEY"]          # 키는 코드에 적지 않습니다
CREATE = "https://api.kie.ai/api/v1/jobs/createTask"
QUERY = "https://api.kie.ai/api/v1/jobs/recordInfo?taskId="
MODEL = "google/gemini-3-8-flash-lite-tts"


def api(method, url, body=None):
    data = None if body is None else json.dumps(body, ensure_ascii=False).encode("utf-8")
    req = urllib.request.Request(url, data=data, method=method)
    req.add_header("Authorization", "Bearer " + KEY)
    req.add_header("Content-Type", "application/json")
    try:
        with urllib.request.urlopen(req, timeout=90) as r:
            return json.loads(r.read().decode("utf-8"))
    except urllib.error.HTTPError as e:
        return json.loads(e.read().decode("utf-8", "replace") or "{}")


def create(voice, text, style=""):
    body = {
        "model": MODEL,
        "input": {
            "temperature": 1,
            "speakers": [{"speaker_id": "Speaker 1", "voice_name": voice}],
            "dialogue_turns": [{"speaker_id": "Speaker 1", "text": text, "style": style}],
        },
    }
    for attempt in range(4):
        res = api("POST", CREATE, body)
        code = res.get("code")
        if code == 200:
            return res["data"]["taskId"]
        if code in (429, 455, 500):          # 한도·일시 오류는 쉬었다가 재시도
            time.sleep(4 + attempt * 3)
            continue
        raise RuntimeError(f"create failed: {code} {res.get('msg')}")  # 401·402·422는 즉시 중단
    raise RuntimeError("retry exhausted")


def wait(task_id):
    for _ in range(40):
        time.sleep(3)
        node = api("GET", QUERY + task_id).get("data") or {}
        if node.get("state") == "success":
            result = json.loads(node.get("resultJson") or "{}")
            return result["resultUrls"][0], node.get("creditsConsumed")
        if node.get("state") == "fail":
            raise RuntimeError(f"fail: {node.get('failCode')} {node.get('failMsg')}")
    raise TimeoutError(task_id)


def as_wav(blob):
    if blob.startswith(b"RIFF"):
        return blob
    rate, ch, bits = 24000, 1, 16                # 헤더 없는 PCM이면 WAV 헤더를 붙입니다
    head = struct.pack("<4sI4s4sIHHIIHH4sI", b"RIFF", 36 + len(blob), b"WAVE", b"fmt ", 16, 1,
                       ch, rate, rate * ch * bits // 8, ch * bits // 8, bits, b"data", len(blob))
    return head + blob


task = create("Charon", "우회전 빨간불이면 정지선 바로 앞에 완전히 멈춥니다.")
url, credits = wait(task)
with urllib.request.urlopen(url, timeout=120) as r:
    open("charon.wav", "wb").write(as_wav(r.read()))   # 14일 뒤 삭제되므로 바로 저장
print("saved charon.wav, credits:", credits)