OPL: Headless CLI that drives the REST API

[codesterribly]

On reddit, you listed the key endpoints for automation, right?

• POST /api/projects/import (import markdown or text)
• POST /api/tts/generate-chapter (queue TTS)
• GET /api/events/subscribe (SSE for progress)
• POST /api/audio/export (export audio as MP3, M4A, WAV)

Because we cannot see the exact request schemas here, I will give you:

• A fully working CLI structure
• Concrete payloads that are reasonable guesses
• Clear “adjust here to match /docs” markers where you will likely tweak one or two field names after looking at the live Swagger UI

Create tools/abm_cli.py at the repo root:

#!/usr/bin/env python3
"""
Simple headless client for AudioBook-Maker.

Pipeline:
  1. Import a markdown file as a new project
  2. Queue TTS generation for all chapters
  3. Watch progress through the SSE event stream
  4. Trigger audio export and download the result

You will probably need to adjust the JSON fields so they match
the schemas shown under http://localhost:8765/docs
"""

import argparse
import json
import os
import sys
import time
from pathlib import Path
from typing import Optional

import requests


def import_markdown(base_url: str, markdown_path: Path, project_name: str) -> dict:
    url = f"{base_url.rstrip('/')}/api/projects/import"
    with markdown_path.open("rb") as f:
        files = {"file": (markdown_path.name, f, "text/markdown")}
        # These form fields are guesses, check /docs and adjust the keys if needed
        data = {"project_name": project_name}
        resp = requests.post(url, data=data, files=files, timeout=600)
    resp.raise_for_status()
    return resp.json()


def generate_chapter_tts(base_url: str, project_id: str, chapter_id: str, tts_engine: str) -> dict:
    url = f"{base_url.rstrip('/')}/api/tts/generate-chapter"
    payload = {
        # Field names are based on common FastAPI style, verify with /docs
        "project_id": project_id,
        "chapter_id": chapter_id,
        "tts_engine": tts_engine,
    }
    resp = requests.post(url, json=payload, timeout=600)
    resp.raise_for_status()
    return resp.json()


def subscribe_events(base_url: str, project_id: str, timeout_seconds: int = 3600) -> None:
    """
    Simple SSE consumer that prints messages until completion.

    In practice, you may want to parse event types and stop
    when you see a “completed” event.
    """
    url = f"{base_url.rstrip('/')}/api/events/subscribe"
    params = {"project_id": project_id}
    with requests.get(url, params=params, stream=True, timeout=timeout_seconds) as resp:
        resp.raise_for_status()
        for line in resp.iter_lines(decode_unicode=True):
            if not line:
                continue
            # SSE format usually sends lines like: "data: {...json...}"
            if line.startswith("data:"):
                data = line[len("data:") :].strip()
                print(f"[EVENT] {data}")
            # Very naive break condition, you will likely refine this
            if "completed" in line.lower():
                print("[EVENT] Found completion marker in SSE stream")
                break


def export_audio(base_url: str, project_id: str, output_format: str) -> Path:
    url = f"{base_url.rstrip('/')}/api/audio/export"
    payload = {
        # Again, check /docs for exact names
        "project_id": project_id,
        "format": output_format,  # "mp3", "m4a", "wav", etc.
    }
    resp = requests.post(url, json=payload, timeout=600)
    resp.raise_for_status()
    meta = resp.json()

    # Many APIs return a download URL; if this one returns a file directly
    # you can skip the second GET and just write resp.content.
    download_url: Optional[str] = meta.get("download_url")
    if not download_url:
        # Fallback: assume export endpoint already streamed back the file
        filename = meta.get("filename", f"{project_id}.{output_format}")
        out_path = Path(filename).resolve()
        out_path.write_bytes(resp.content)
        print(f"Saved exported audio to {out_path}")
        return out_path

    # Download from download_url
    dl_resp = requests.get(download_url, timeout=600)
    dl_resp.raise_for_status()

    filename = meta.get("filename") or download_url.split("/")[-1] or f"{project_id}.{output_format}"
    out_path = Path(filename).resolve()
    out_path.write_bytes(dl_resp.content)
    print(f"Saved exported audio to {out_path}")
    return out_path


def run_full_pipeline(
    base_url: str,
    markdown_path: Path,
    project_name: str,
    tts_engine: str,
    output_format: str,
    watch_events: bool,
) -> None:
    print(f"[1/4] Importing markdown from {markdown_path}")
    project = import_markdown(base_url, markdown_path, project_name)
    # Adjust these keys to whatever the API actually returns
    project_id = str(project.get("id") or project.get("project_id"))
    if not project_id:
        print("Could not read project id from /api/projects/import response", file=sys.stderr)
        print(json.dumps(project, indent=2), file=sys.stderr)
        sys.exit(1)

    chapters = project.get("chapters") or []
    if not chapters:
        print("No chapters found in project response, check API schema", file=sys.stderr)
        print(json.dumps(project, indent=2), file=sys.stderr)
        sys.exit(1)

    print(f"[2/4] Queuing TTS for {len(chapters)} chapters using engine '{tts_engine}'")
    for chapter in chapters:
        chapter_id = str(chapter.get("id") or chapter.get("chapter_id"))
        title = chapter.get("title") or chapter_id
        print(f"  - Generating TTS for chapter {chapter_id} ({title})")
        generate_chapter_tts(base_url, project_id, chapter_id, tts_engine)
        # If the API supports batching all chapters in one call, you can optimize this later.

    if watch_events:
        print("[3/4] Subscribing to SSE progress events")
        # This will block until it sees “completed” in an event line
        subscribe_events(base_url, project_id)

    print(f"[4/4] Exporting audio as {output_format}")
    export_audio(base_url, project_id, output_format)


def main() -> None:
    parser = argparse.ArgumentParser(description="Headless client for AudioBook-Maker backend")
    parser.add_argument(
        "--base-url",
        default=os.environ.get("ABM_BASE_URL", "http://localhost:8765"),
        help="Base URL of the backend, default http://localhost:8765",
    )
    parser.add_argument("--markdown", required=True, help="Path to markdown file")
    parser.add_argument("--name", required=True, help="Project name")
    parser.add_argument(
        "--tts-engine",
        default=os.environ.get("ABM_DEFAULT_TTS_ENGINE", "xtts"),
        help="TTS engine id to use, for example xtts or chatterbox",
    )
    parser.add_argument(
        "--format",
        default=os.environ.get("ABM_DEFAULT_EXPORT_FORMAT", "mp3"),
        help="Export format, mp3 or m4a or wav",
    )
    parser.add_argument(
        "--watch-events",
        action="store_true",
        help="Subscribe to SSE events until completion",
    )

    args = parser.parse_args()
    markdown_path = Path(args.markdown).expanduser().resolve()
    if not markdown_path.is_file():
        print(f"Markdown file not found: {markdown_path}", file=sys.stderr)
        sys.exit(1)

    run_full_pipeline(
        base_url=args.base_url,
        markdown_path=markdown_path,
        project_name=args.name,
        tts_engine=args.tts_engine,
        output_format=args.format,
        watch_events=args.watch_events,
    )


if __name__ == "__main__":
    main()

Make it executable:

chmod +x tools/abm_cli.py

Run the pipeline against your Docker backend:

# Backend running via docker or docker compose on localhost:8765

# Simple run
./tools/abm_cli.py \
  --markdown /path/to/book.md \
  --name "Test Book" \
  --tts-engine xtts \
  --format mp3 \
  --watch-events

Then:

• abm_cli.py calls POST /api/projects/import to create the project from your markdown
• It calls POST /api/tts/generate-chapter per chapter
• If you pass --watch-events, it tails GET /api/events/subscribe and prints progress
• Finally it calls POST /api/audio/export and writes the exported file

You will almost certainly only need to adjust the JSON field names in import_markdown, generate_chapter_tts, and export_audio to match whatever the live Swagger docs show on your instance.