feat(api): extract pure API clients for image generation and TTS

- Add bin/api/ package with generate_image() and synthesize_tts()
- Refactor bin/commands/image.py to call bin.api.image.generate_image
- Refactor bin/commands/voice.py to call bin.api.voice.synthesize_tts
- API layer is free of Click CLI dependencies, usable by external scripts

bin/api/__init__.py

@@ -0,0 +1,6 @@
+"""Pure API clients for external services (no CLI deps)."""
+
+from .image import generate_image
+from .voice import synthesize_tts
+
+__all__ = ["generate_image", "synthesize_tts"]

bin/api/image.py

@@ -0,0 +1,57 @@
+"""Image generation API client (Doubao/Seedream)."""
+
+import base64
+import os
+import urllib.request
+from pathlib import Path
+
+_RATIO_MAP = {
+    "1:1": "1024x1024",
+    "4:3": "1024x768",
+    "3:4": "768x1024",
+    "16:9": "1280x720",
+    "9:16": "720x1280",
+}
+
+
+def generate_image(
+    api_key: str,
+    prompt: str,
+    size: str | None = None,
+    ratio: str | None = None,
+    output: str | None = None,
+    watermark: bool = True,
+    b64: bool = False,
+) -> str:
+    """Generate image using Doubao/Seedream API.
+
+    Returns absolute file path if saved, or URL string if output is None.
+    """
+    from openai import OpenAI
+
+    resolved_size = size or (ratio and _RATIO_MAP.get(ratio)) or "2K"
+
+    client = OpenAI(
+        base_url="https://ark.cn-beijing.volces.com/api/v3",
+        api_key=api_key,
+    )
+
+    resp = client.images.generate(
+        model="doubao-seedream-5-0-260128",
+        prompt=prompt,
+        size=resolved_size,
+        response_format="b64_json" if b64 else "url",
+        extra_body={"watermark": watermark},
+    )
+
+    if b64:
+        image_bytes = base64.b64decode(resp.data[0].b64_json)
+        save_path = output or "generated_image.png"
+        Path(save_path).write_bytes(image_bytes)
+        return os.path.abspath(save_path)
+    else:
+        url = resp.data[0].url
+        if output:
+            urllib.request.urlretrieve(url, output)
+            return os.path.abspath(output)
+        return url

bin/api/voice.py

@@ -0,0 +1,93 @@
+"""Voice / TTS API client via Volcengine (ByteDance) WebSocket API."""
+
+import asyncio
+import json
+import uuid
+from pathlib import Path
+
+from bin.commands.protocols import EventType, MsgType, full_client_request, receive_message
+
+_RESOURCE_ID = "volc.service_type.10029"
+_ENDPOINT = "wss://openspeech.bytedance.com/api/v3/tts/unidirectional/stream"
+
+_VOICE_CHOICES = [
+    "zh_female_cancan_mars_bigtts",
+    "zh_female_shuangkuaisisi_moon_bigtts",
+    "zh_male_wennuanahu_moon_bigtts",
+    "zh_male_sunwukong_moon_bigtts",
+]
+
+_FORMAT_CHOICES = ["mp3", "wav", "pcm"]
+
+
+def synthesize_tts(
+    appid: str,
+    token: str,
+    text: str,
+    speaker: str = "zh_male_wennuanahu_moon_bigtts",
+    fmt: str = "mp3",
+    speech_rate: int = 0,
+    output_path: str = "output.mp3",
+) -> str:
+    """Synchronous TTS synthesis via Volcengine WebSocket API.
+
+    Returns absolute path to the output audio file.
+    """
+    try:
+        import websockets
+    except ImportError as e:
+        raise RuntimeError(f"Missing dependency: {e}. Run: pip install websockets") from e
+
+    headers = {
+        "X-Api-App-Id": appid,
+        "X-Api-Access-Key": token,
+        "X-Api-Resource-Id": _RESOURCE_ID,
+        "X-Api-Connect-Id": str(uuid.uuid4()),
+    }
+
+    async def _run() -> bytearray:
+        ws = await websockets.connect(
+            _ENDPOINT,
+            additional_headers=headers,
+            max_size=10 * 1024 * 1024,
+            open_timeout=30,
+        )
+
+        request = {
+            "user": {"uid": str(uuid.uuid4())},
+            "req_params": {
+                "speaker": speaker,
+                "audio_params": {
+                    "format": fmt,
+                    "sample_rate": 24000,
+                    "speech_rate": speech_rate,
+                },
+                "text": text,
+            },
+        }
+
+        await full_client_request(ws, json.dumps(request).encode())
+
+        audio_data = bytearray()
+        while True:
+            msg = await receive_message(ws)
+            if msg.type == MsgType.FullServerResponse and msg.event == EventType.SessionFinished:
+                break
+            elif msg.type == MsgType.AudioOnlyServer:
+                audio_data.extend(msg.payload)
+            elif msg.type == MsgType.Error:
+                await ws.close()
+                raise RuntimeError(
+                    f"TTS service error: {msg.payload.decode('utf-8', errors='replace')}"
+                )
+
+        await ws.close()
+        return audio_data
+
+    audio_data = asyncio.run(_run())
+    if not audio_data:
+        raise RuntimeError("No audio data received from TTS service")
+
+    out = Path(output_path)
+    out.write_bytes(audio_data)
+    return str(out.resolve())

bin/commands/image.py

@@ -1,13 +1,11 @@
 """Image utilities."""
 
-import base64
-import os
 import subprocess
-import urllib.request
 from pathlib import Path
 
 import click
 
+from bin.api.image import _RATIO_MAP, generate_image
 from bin.config import config
 from bin.utils import handle_errors, run_command
 
@@ -187,15 +185,6 @@ def compress_images(files, target_mb, max_width, quality, suffix, dry_run):
             click.echo(f"Compressed: {img_file} -> {output} ({size_mb:.2f} MB)")
 
 
-_RATIO_MAP = {
-    "1:1": "1024x1024",
-    "4:3": "1024x768",
-    "3:4": "768x1024",
-    "16:9": "1280x720",
-    "9:16": "720x1280",
-}
-
-
 @image.command("generate")
 @click.argument("prompt", required=False)
 @click.option("-f", "--file", type=click.Path(exists=True), help="从文件读取 prompt")
@@ -207,8 +196,6 @@ _RATIO_MAP = {
 @handle_errors
 def generate_image_cmd(prompt, file, size, ratio, output, no_watermark, b64):
     """使用豆包/Seedream 模型生成图片。"""
-    from openai import OpenAI
-
     if file:
         prompt = Path(file).read_text(encoding="utf-8").strip()
     elif not prompt:
@@ -222,30 +209,13 @@ def generate_image_cmd(prompt, file, size, ratio, output, no_watermark, b64):
     if size and ratio:
         click.echo("Warning: --size 和 --ratio 同时指定，--size 优先", err=True)
 
-    resolved_size = size or (ratio and _RATIO_MAP[ratio]) or "2K"
+    result = generate_image(
-
-    client = OpenAI(
-        base_url="https://ark.cn-beijing.volces.com/api/v3",
         api_key=api_key,
-    )
-
-    resp = client.images.generate(
-        model="doubao-seedream-5-0-260128",
         prompt=prompt,
-        size=resolved_size,
+        size=size,
-        response_format="b64_json" if b64 else "url",
+        ratio=ratio,
-        extra_body={"watermark": not no_watermark},
+        output=output,
+        watermark=not no_watermark,
+        b64=b64,
     )
-
+    click.echo(result)
-    if b64:
-        image_bytes = base64.b64decode(resp.data[0].b64_json)
-        save_path = output or "generated_image.png"
-        Path(save_path).write_bytes(image_bytes)
-        click.echo(os.path.abspath(save_path))
-    else:
-        url = resp.data[0].url
-        if output:
-            urllib.request.urlretrieve(url, output)
-            click.echo(os.path.abspath(output))
-        else:
-            click.echo(url)

bin/commands/voice.py

@@ -1,19 +1,14 @@
 """Voice / TTS utilities via Volcengine (ByteDance) WebSocket API."""
 
-import asyncio
-import json
 import os
-import uuid
 from pathlib import Path
 
 import click
 
+from bin.api.voice import synthesize_tts
 from bin.config import config
 from bin.utils import handle_errors
 
-_RESOURCE_ID = "volc.service_type.10029"
-_ENDPOINT = "wss://openspeech.bytedance.com/api/v3/tts/unidirectional/stream"
-
 _VOICE_CHOICES = [
     "zh_female_cancan_mars_bigtts",
     "zh_female_shuangkuaisisi_moon_bigtts",
@@ -41,17 +36,6 @@ def _synthesize_sync(
     output_path: str,
 ) -> str:
     """Synchronous wrapper around async WebSocket TTS."""
-    try:
-        from bin.commands.protocols import (
-            EventType,
-            MsgType,
-            full_client_request,
-            receive_message,
-        )
-        import websockets
-    except ImportError as e:
-        raise RuntimeError(f"Missing dependency: {e}. Run: pip install websockets")
-
     appid, access_token = _get_credentials()
     if not appid or not access_token:
         raise click.UsageError(
@@ -60,61 +44,15 @@ def _synthesize_sync(
             "  mytoolkit env set volc_access_token <token>"
         )
 
-    headers = {
+    return synthesize_tts(
-        "X-Api-App-Id": appid,
+        appid=appid,
-        "X-Api-Access-Key": access_token,
+        token=access_token,
-        "X-Api-Resource-Id": _RESOURCE_ID,
+        text=text,
-        "X-Api-Connect-Id": str(uuid.uuid4()),
+        speaker=speaker,
-    }
+        fmt=fmt,
-
+        speech_rate=speech_rate,
-    async def _run() -> bytearray:
+        output_path=output_path,
-        try:
+    )
-            ws = await websockets.connect(
-                _ENDPOINT,
-                additional_headers=headers,
-                max_size=10 * 1024 * 1024,
-                open_timeout=30,
-            )
-        except Exception as e:
-            raise RuntimeError(f"WebSocket connection failed: {e}") from e
-
-        request = {
-            "user": {"uid": str(uuid.uuid4())},
-            "req_params": {
-                "speaker": speaker,
-                "audio_params": {
-                    "format": fmt,
-                    "sample_rate": 24000,
-                    "speech_rate": speech_rate,
-                },
-                "text": text,
-            },
-        }
-
-        await full_client_request(ws, json.dumps(request).encode())
-
-        audio_data = bytearray()
-        while True:
-            msg = await receive_message(ws)
-            if msg.type == MsgType.FullServerResponse and msg.event == EventType.SessionFinished:
-                break
-            elif msg.type == MsgType.AudioOnlyServer:
-                audio_data.extend(msg.payload)
-            elif msg.type == MsgType.Error:
-                await ws.close()
-                raise RuntimeError(
-                    f"TTS service error: {msg.payload.decode('utf-8', errors='replace')}"
-                )
-
-        await ws.close()
-        return audio_data
-
-    audio_data = asyncio.run(_run())
-    if not audio_data:
-        raise RuntimeError("No audio data received from TTS service")
-
-    Path(output_path).write_bytes(audio_data)
-    return output_path
 
 
 @click.group()