480 lines
16 KiB
Python
480 lines
16 KiB
Python
# src/voice_gen/cli.py
|
|
"""argparse-based CLI for voice-gen."""
|
|
from __future__ import annotations
|
|
|
|
import argparse
|
|
import json
|
|
import shutil
|
|
import subprocess
|
|
import sys
|
|
from datetime import datetime, timezone
|
|
from pathlib import Path
|
|
|
|
from .registry import VoiceRegistry
|
|
|
|
_MMX_CONFIG_PATH = Path.home() / ".mmx" / "config.json"
|
|
_oauth_token_cache: str | None = None
|
|
|
|
|
|
def _load_oauth_token() -> str:
|
|
"""Load the OAuth access token from ~/.mmx/config.json (cached)."""
|
|
global _oauth_token_cache
|
|
if _oauth_token_cache is None:
|
|
if not _MMX_CONFIG_PATH.exists():
|
|
raise FileNotFoundError(
|
|
f"mmx config not found at {_MMX_CONFIG_PATH}; "
|
|
"run `mmx auth login` first"
|
|
)
|
|
data = json.loads(_MMX_CONFIG_PATH.read_text(encoding="utf-8"))
|
|
_oauth_token_cache = data["oauth"]["access_token"]
|
|
return _oauth_token_cache
|
|
|
|
|
|
def _get_mmx_resource_url() -> str:
|
|
"""Return the mmx API base URL from the user's config."""
|
|
if not _MMX_CONFIG_PATH.exists():
|
|
return "https://api.minimaxi.com"
|
|
data = json.loads(_MMX_CONFIG_PATH.read_text(encoding="utf-8"))
|
|
return data.get("oauth", {}).get("resource_url", "https://api.minimaxi.com")
|
|
|
|
|
|
def build_parser() -> argparse.ArgumentParser:
|
|
parser = argparse.ArgumentParser(
|
|
prog="voice-gen",
|
|
description="Batch TTS via mmx CLI with custom voice clones.",
|
|
)
|
|
sub = parser.add_subparsers(dest="command", required=True)
|
|
|
|
sub.add_parser("init", help="Create voices.json, scripts/, output/ in CWD")
|
|
|
|
voices = sub.add_parser("voices", help="Manage the voice registry")
|
|
voices_sub = voices.add_subparsers(dest="voices_command", required=True)
|
|
list_p = voices_sub.add_parser("list", help="List registered voices (default: local)")
|
|
list_p.add_argument("scope", nargs="?", default="local", choices=["local", "server"],
|
|
help="Which voices to list (default: local)")
|
|
add_p = voices_sub.add_parser("add", help="Upload an audio file; friendly name is derived from the filename stem")
|
|
add_p.add_argument("audio", help="Path to the reference audio file")
|
|
pull_p = voices_sub.add_parser("pull", help="Sync voices from server into the local registry")
|
|
pull_p.add_argument("--auto", action="store_true",
|
|
help="Always auto-derive friendly name from filename (default if non-TTY)")
|
|
pull_p.add_argument("--interactive", action="store_true",
|
|
help="Always prompt for each friendly name (default if TTY)")
|
|
pull_p.add_argument("--force", action="store_true",
|
|
help="Overwrite local entries that have the same name")
|
|
pull_p.add_argument("--dry-run", action="store_true",
|
|
help="Show what would be added/changed without modifying voices.json")
|
|
voices_sub.add_parser("cleanup", help="Delete duplicate voices that share the same source filename")
|
|
|
|
gen = sub.add_parser("gen", help="Synthesize audio from a directory of .md scripts")
|
|
gen.add_argument("--voice", required=True, help="Voice name from registry")
|
|
gen.add_argument("input", help="Input directory containing .md files")
|
|
gen.add_argument("--out", default="output", help="Output directory (default: output)")
|
|
gen.add_argument("--format", default="mp3", choices=["mp3", "wav", "flac", "pcm", "opus"])
|
|
return parser
|
|
|
|
|
|
def _registry() -> VoiceRegistry:
|
|
return VoiceRegistry.load(Path("voices.json"))
|
|
|
|
|
|
def _upload_voice_file(path: Path, purpose: str = "voice_clone") -> dict:
|
|
"""Upload a file to MiniMax storage via direct API call.
|
|
|
|
Bypasses `mmx file upload` because mmx CLI v1.0.16 POSTs to the wrong
|
|
endpoint (`/v1/files`) and gets a 404. The correct endpoint is
|
|
`/v1/files/upload` (multipart form).
|
|
"""
|
|
from .errors import MmxError
|
|
|
|
token = _load_oauth_token()
|
|
base_url = _get_mmx_resource_url()
|
|
curl = shutil.which("curl")
|
|
if not curl:
|
|
raise MmxError(
|
|
exit_code=1,
|
|
stderr="curl not found on PATH; required for direct API upload",
|
|
command="curl",
|
|
)
|
|
|
|
url = f"{base_url}/v1/files/upload"
|
|
cmd = [
|
|
curl, "-s", "-X", "POST", url,
|
|
"-H", f"Authorization: Bearer {token}",
|
|
"-F", f"file=@{path}",
|
|
"-F", f"purpose={purpose}",
|
|
]
|
|
result = subprocess.run(cmd, capture_output=True, text=True)
|
|
if result.returncode != 0:
|
|
raise MmxError(
|
|
exit_code=result.returncode,
|
|
stderr=result.stderr,
|
|
command=f"curl ... {url}",
|
|
)
|
|
|
|
try:
|
|
data = json.loads(result.stdout)
|
|
except json.JSONDecodeError as e:
|
|
raise MmxError(
|
|
exit_code=1,
|
|
stderr=f"invalid JSON from upload API: {result.stdout[:500]}",
|
|
command=f"curl ... {url}",
|
|
)
|
|
|
|
if "file" not in data or "file_id" not in data["file"]:
|
|
raise MmxError(
|
|
exit_code=1,
|
|
stderr=f"unexpected response from upload API: {data}",
|
|
command=f"curl ... {url}",
|
|
)
|
|
|
|
return {
|
|
"file_id": data["file"]["file_id"],
|
|
"filename": data["file"].get("filename", path.name),
|
|
}
|
|
|
|
|
|
def _delete_remote_file(file_id: int) -> bool:
|
|
"""Run `mmx file delete`. Raises MmxError on failure."""
|
|
from .errors import MmxError
|
|
cmd = [
|
|
"mmx", "file", "delete",
|
|
"--non-interactive", "--quiet",
|
|
"--file-id", str(file_id),
|
|
]
|
|
from .synthesizer import _run_mmx
|
|
result = _run_mmx(cmd)
|
|
if result.returncode != 0:
|
|
raise MmxError(exit_code=result.returncode, stderr=result.stderr, command=" ".join(cmd))
|
|
return True
|
|
|
|
|
|
def _clone_voice_file(file_id: int, voice_id: str) -> None:
|
|
"""Register an uploaded file as a voice clone via /v1/voice_clone.
|
|
|
|
After this call, `voice_id` (the user's friendly name) is usable in the
|
|
synthesis endpoint (`/v1/t2a_v2`). Without this step, the file_id alone
|
|
cannot be used as a voice_id 鈥?the API will respond with
|
|
"voice id not exist".
|
|
"""
|
|
from .errors import MmxError
|
|
|
|
token = _load_oauth_token()
|
|
base_url = _get_mmx_resource_url()
|
|
curl = shutil.which("curl")
|
|
if not curl:
|
|
raise MmxError(
|
|
exit_code=1,
|
|
stderr="curl not found on PATH; required for direct API call",
|
|
command="curl",
|
|
)
|
|
|
|
url = f"{base_url}/v1/voice_clone"
|
|
payload = json.dumps({
|
|
"file_id": file_id,
|
|
"voice_id": voice_id,
|
|
"model": "speech-2.8-hd",
|
|
"text": "test",
|
|
"need_noise_reduction": False,
|
|
"need_volume_normalization": False,
|
|
})
|
|
cmd = [
|
|
curl, "-s", "-X", "POST", url,
|
|
"-H", f"Authorization: Bearer {token}",
|
|
"-H", "Content-Type: application/json",
|
|
"-d", payload,
|
|
]
|
|
result = subprocess.run(cmd, capture_output=True, text=True)
|
|
if result.returncode != 0:
|
|
raise MmxError(
|
|
exit_code=result.returncode,
|
|
stderr=result.stderr,
|
|
command=f"curl ... {url}",
|
|
)
|
|
|
|
try:
|
|
data = json.loads(result.stdout)
|
|
except json.JSONDecodeError as e:
|
|
raise MmxError(
|
|
exit_code=1,
|
|
stderr=f"invalid JSON from voice_clone API: {result.stdout[:500]}",
|
|
command=f"curl ... {url}",
|
|
)
|
|
|
|
resp = data.get("base_resp", {})
|
|
if resp.get("status_code") != 0:
|
|
raise MmxError(
|
|
exit_code=1,
|
|
stderr=f"voice_clone failed: status_code={resp.get('status_code')} "
|
|
f"msg={resp.get('status_msg')} raw={data}",
|
|
command=f"curl ... {url}",
|
|
)
|
|
|
|
|
|
def cmd_init(_args: argparse.Namespace) -> int:
|
|
cwd = Path.cwd()
|
|
(cwd / "scripts").mkdir(exist_ok=True)
|
|
(cwd / "output").mkdir(exist_ok=True)
|
|
registry_path = cwd / "voices.json"
|
|
if not registry_path.exists():
|
|
registry_path.write_text(
|
|
json.dumps({"voices": {}}, ensure_ascii=False, indent=2) + "\n",
|
|
encoding="utf-8",
|
|
)
|
|
print(f"initialized in {cwd}")
|
|
return 0
|
|
|
|
|
|
def cmd_voices_list(args: argparse.Namespace) -> int:
|
|
if args.scope == "server":
|
|
return _list_server_voices()
|
|
reg = _registry()
|
|
names = reg.list_names()
|
|
if not names:
|
|
print("(no voices registered; run `voice-gen voices add <audio>` or `voice-gen voices pull`)")
|
|
return 0
|
|
width = max(len(n) for n in names)
|
|
for n in names:
|
|
e = reg.get(n)
|
|
print(f"{n.ljust(width)} file_id={e.file_id} {e.filename}")
|
|
return 0
|
|
|
|
|
|
def _list_server_voices() -> int:
|
|
"""List server-side voice_clone files via mmx file list."""
|
|
from .synthesizer import _run_mmx
|
|
from .errors import MmxError
|
|
cmd = ["mmx", "file", "list", "--purpose", "voice_clone",
|
|
"--output", "json", "--quiet"]
|
|
result = _run_mmx(cmd)
|
|
if result.returncode != 0:
|
|
raise MmxError(exit_code=result.returncode, stderr=result.stderr,
|
|
command=" ".join(cmd))
|
|
try:
|
|
data = json.loads(result.stdout)
|
|
except json.JSONDecodeError as e:
|
|
print(f"(server returned invalid JSON: {e})", file=sys.stderr)
|
|
return 1
|
|
files = data.get("files", [])
|
|
if not files:
|
|
print("(no voice_clone files on server)")
|
|
return 0
|
|
width = max(len(f.get("filename", "")) for f in files)
|
|
for f in files:
|
|
filename = f.get("filename", "?")
|
|
file_id = f.get("file_id", "?")
|
|
print(f"{filename.ljust(width)} file_id={file_id}")
|
|
return 0
|
|
|
|
|
|
def _derive_voice_name(audio_path: Path) -> str:
|
|
"""Derive a friendly voice name from an audio filename (stem, spaces to underscores)."""
|
|
return audio_path.stem.replace(" ", "_")
|
|
|
|
|
|
def cmd_voices_add(args: argparse.Namespace) -> int:
|
|
from .registry import VoiceEntry # local import keeps top of file clean
|
|
|
|
audio = Path(args.audio)
|
|
if not audio.exists():
|
|
print(f"audio file not found: {audio}", file=sys.stderr)
|
|
return 2
|
|
|
|
name = _derive_voice_name(audio)
|
|
reg = _registry()
|
|
if reg.has(name):
|
|
print(f"voice {name!r} already exists locally; pick a different filename", file=sys.stderr)
|
|
return 1
|
|
|
|
info = _upload_voice_file(audio)
|
|
_clone_voice_file(info["file_id"], name)
|
|
entry = VoiceEntry(
|
|
file_id=info["file_id"],
|
|
filename=info["filename"],
|
|
created_at=datetime.now(timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ"),
|
|
)
|
|
reg.add(name, entry)
|
|
reg.save()
|
|
print(f"registered {name!r} -> file_id={entry.file_id}")
|
|
return 0
|
|
|
|
|
|
def cmd_voices_pull(args: argparse.Namespace) -> int:
|
|
"""Pull server-side voice_clone files into the local registry."""
|
|
from .synthesizer import _run_mmx
|
|
from .errors import MmxError
|
|
from .registry import VoiceEntry
|
|
|
|
cmd = ["mmx", "file", "list", "--purpose", "voice_clone",
|
|
"--output", "json", "--quiet"]
|
|
result = _run_mmx(cmd)
|
|
if result.returncode != 0:
|
|
raise MmxError(exit_code=result.returncode, stderr=result.stderr,
|
|
command=" ".join(cmd))
|
|
try:
|
|
data = json.loads(result.stdout)
|
|
except json.JSONDecodeError as e:
|
|
print(f"(server returned invalid JSON: {e})", file=sys.stderr)
|
|
return 1
|
|
files = data.get("files", [])
|
|
if not files:
|
|
print("(no voice_clone files on server)")
|
|
return 0
|
|
|
|
reg = _registry()
|
|
interactive = args.interactive or (not args.auto and sys.stdin.isatty())
|
|
|
|
added, skipped, overwritten = 0, 0, 0
|
|
for f in files:
|
|
filename = f.get("filename", "")
|
|
file_id = f.get("file_id")
|
|
if not filename or file_id is None:
|
|
continue
|
|
default_name = _derive_voice_name(Path(filename))
|
|
if interactive:
|
|
try:
|
|
user_input = input(f" file_id={file_id} filename={filename}\n friendly name [{default_name}]: ").strip()
|
|
except EOFError:
|
|
user_input = ""
|
|
name = user_input or default_name
|
|
else:
|
|
name = default_name
|
|
|
|
existing = reg.has(name)
|
|
if existing and not args.force:
|
|
print(f" 路 skip {name!r} (already local; use --force to overwrite)")
|
|
skipped += 1
|
|
continue
|
|
|
|
entry = VoiceEntry(
|
|
file_id=file_id,
|
|
filename=filename,
|
|
created_at=datetime.now(timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ"),
|
|
)
|
|
if args.dry_run:
|
|
action = "overwrite" if existing else "add"
|
|
print(f" 路 would {action} {name!r} file_id={file_id} filename={filename}")
|
|
continue
|
|
|
|
if existing:
|
|
reg.remove(name)
|
|
overwritten += 1
|
|
else:
|
|
added += 1
|
|
reg.add(name, entry)
|
|
print(f" + {name!r} file_id={file_id} filename={filename}")
|
|
|
|
if not args.dry_run:
|
|
reg.save()
|
|
summary = []
|
|
if added: summary.append(f"{added} added")
|
|
if overwritten: summary.append(f"{overwritten} overwritten")
|
|
if skipped: summary.append(f"{skipped} skipped")
|
|
if not summary:
|
|
summary.append("no changes")
|
|
prefix = "would pull" if args.dry_run else "pulled"
|
|
print(f"{prefix}: {', '.join(summary)}")
|
|
return 0
|
|
|
|
|
|
def cmd_voices_cleanup(_args: argparse.Namespace) -> int:
|
|
reg = _registry()
|
|
by_filename: dict[str, list[str]] = {}
|
|
for name in reg.list_names():
|
|
e = reg.get(name)
|
|
by_filename.setdefault(e.filename, []).append(name)
|
|
|
|
removed = 0
|
|
for filename, names in by_filename.items():
|
|
if len(names) <= 1:
|
|
continue
|
|
# keep the oldest by created_at
|
|
names_sorted = sorted(names, key=lambda n: reg.get(n).created_at)
|
|
keep = names_sorted[0]
|
|
for drop in names_sorted[1:]:
|
|
entry = reg.get(drop)
|
|
_delete_remote_file(entry.file_id)
|
|
reg.remove(drop)
|
|
print(f"removed duplicate {drop!r} (file_id={entry.file_id}, filename={filename})")
|
|
removed += 1
|
|
if removed:
|
|
reg.save()
|
|
print(f"cleanup done; removed {removed} duplicate(s)")
|
|
return 0
|
|
|
|
|
|
def cmd_gen(args: argparse.Namespace) -> int:
|
|
from .extractor import extract_text
|
|
from .synthesizer import synthesize
|
|
from .errors import VoiceNotFoundError
|
|
|
|
reg = _registry()
|
|
try:
|
|
voice = reg.get(args.voice)
|
|
except VoiceNotFoundError as e:
|
|
print(str(e), file=sys.stderr)
|
|
return 1
|
|
|
|
in_dir = Path(args.input)
|
|
if not in_dir.is_dir():
|
|
print(f"input is not a directory: {in_dir}", file=sys.stderr)
|
|
return 2
|
|
out_dir = Path(args.out)
|
|
out_dir.mkdir(parents=True, exist_ok=True)
|
|
|
|
md_files = sorted(in_dir.glob("*.md"))
|
|
if not md_files:
|
|
print(f"no .md files in {in_dir}", file=sys.stderr)
|
|
return 1
|
|
|
|
successes, failures = 0, 0
|
|
for md in md_files:
|
|
out_path = out_dir / f"{md.stem}.{args.format}"
|
|
try:
|
|
text = extract_text(md)
|
|
# Use the friendly name (dict key) as the voice_id for synthesis.
|
|
# The file_id in the registry entry is just for tracking the
|
|
# source audio; the API uses the friendly name as voice_id after
|
|
# voice_clone registration.
|
|
synthesize(
|
|
text=text,
|
|
voice_id=args.voice,
|
|
out_path=out_path,
|
|
format=args.format,
|
|
)
|
|
print(f"OK {md.name} -> {out_path}")
|
|
successes += 1
|
|
except Exception as e: # noqa: BLE001 - log and continue
|
|
print(f"ERROR {md.name}: {e}", file=sys.stderr)
|
|
failures += 1
|
|
print(f"done: {successes} ok, {failures} failed")
|
|
return 0 if failures == 0 else 1
|
|
|
|
|
|
def main(argv: list[str] | None = None) -> int:
|
|
parser = build_parser()
|
|
try:
|
|
args = parser.parse_args(argv)
|
|
except SystemExit as e:
|
|
# argparse calls sys.exit on parse errors; convert to return code
|
|
# so callers (e.g. tests) can observe it without exiting the process.
|
|
return e.code if isinstance(e.code, int) else 2
|
|
if args.command == "init":
|
|
return cmd_init(args)
|
|
if args.command == "voices" and args.voices_command == "list":
|
|
return cmd_voices_list(args)
|
|
if args.command == "voices" and args.voices_command == "add":
|
|
return cmd_voices_add(args)
|
|
if args.command == "voices" and args.voices_command == "pull":
|
|
return cmd_voices_pull(args)
|
|
if args.command == "voices" and args.voices_command == "cleanup":
|
|
return cmd_voices_cleanup(args)
|
|
if args.command == "gen":
|
|
return cmd_gen(args)
|
|
parser.error("unknown command")
|
|
return 2 # unreachable
|
|
|
|
|
|
if __name__ == "__main__":
|
|
sys.exit(main())
|
|
|