Podcast Transcription with Speaker Labels

GNANI_API_KEY=your-api-key-here

import os

API_KEY = os.getenv("GNANI_API_KEY")
HEADERS = {"X-API-Key-ID": API_KEY}

import os
import requests
from pathlib import Path

BATCH_SUBMIT = "https://api.vachana.ai/stt/v3/batch/submit"

def submit_job(
    audio_path: str,
    language_code: str = "hi-IN",
    itn: bool = True,
) -> str:
    api_key = os.getenv("GNANI_API_KEY")
    headers = {"X-API-Key-ID": api_key}

    with open(audio_path, "rb") as f:
        files = [("audio_files", (Path(audio_path).name, f, "audio/wav"))]
        data  = {
            "language_code":   language_code,
            "is_multi_channel": "false",
            "format":          "transcribe" if itn else "verbatim",
        }
        resp = requests.post(BATCH_SUBMIT, headers=headers, files=files, data=data)

    resp.raise_for_status()
    job_id = resp.json()["job_id"]
    print(f"Submitted. job_id: {job_id}")
    return job_id

import time
from typing import Optional

BATCH_STATUS  = "https://api.vachana.ai/stt/v3/batch/status/{job_id}"
POLL_INTERVAL = 60  # seconds — enforced minimum; do not reduce

def poll_until_complete(job_id: str) -> Optional[list]:
    api_key = os.getenv("GNANI_API_KEY")
    headers = {"X-API-Key-ID": api_key}
    url     = BATCH_STATUS.format(job_id=job_id)

    print(f"Polling job {job_id} every {POLL_INTERVAL}s...")

    while True:
        time.sleep(POLL_INTERVAL)

        resp = requests.get(url, headers=headers)
        resp.raise_for_status()
        payload  = resp.json()
        status   = payload["status"]
        progress = payload.get("overall_progress", "–")

        print(f"  [{status}]  progress: {progress}%")

        if status == "completed":
            print(f"Job complete. {payload['completed_files']} file(s) transcribed.")
            return payload.get("results", [])

        if status == "failed":
            print(f"Job failed: {payload.get('error')}")
            return None

import json
from pathlib import Path
from typing import Dict

def parse_results(results: list, output_dir: Path) -> Dict[str, dict]:
    outputs = {}

    for file_result in results:
        fname    = Path(file_result["filename"]).stem
        segments = file_result.get("segments", [])

        if not segments or file_result.get("status") == "failed":
            print(f"Skipping {fname}: {file_result.get('error', 'no segments')}")
            continue

        lines, speaker_times, segment_meta = [], {}, []

        for seg in segments:
            spk   = seg.get("speaker_id", "UNKNOWN")
            text  = seg.get("text", "").strip()
            start = seg.get("start_time", 0.0)
            end   = seg.get("end_time",   0.0)

            ts = f"{int(start // 60):02d}:{int(start % 60):02d}"
            lines.append(f"[{ts}] SPEAKER_{spk}: {text}")
            speaker_times[spk] = speaker_times.get(spk, 0.0) + (end - start)
            segment_meta.append({
                "segment_id":        seg.get("segment_id"),
                "speaker_id":        spk,
                "start_time":        start,
                "end_time":          end,
                "text":              text,
                "confidence":        seg.get("confidence"),
                "language_detected": seg.get("language_detected"),
            })

        transcript_path = output_dir / f"{fname}_transcript.txt"
        transcript_path.write_text("\n".join(lines), encoding="utf-8")

        metadata_path = output_dir / f"{fname}_metadata.json"
        metadata_path.write_text(json.dumps({
            "filename":          file_result["filename"],
            "total_duration_s":  file_result.get("total_duration"),
            "speaker_talk_time": {f"SPEAKER_{k}": round(v, 2) for k, v in speaker_times.items()},
            "segments":          segment_meta,
        }, indent=2, ensure_ascii=False), encoding="utf-8")

        outputs[fname] = {
            "transcript_path": str(transcript_path),
            "metadata_path":   str(metadata_path),
        }
        print(f"Parsed: {fname} → {len(lines)} segments, {len(speaker_times)} speaker(s)")

    return outputs

import os
import json
import time
import requests
from pathlib import Path
from typing  import Dict, List, Optional

BATCH_SUBMIT  = "https://api.vachana.ai/stt/v3/batch/submit"
BATCH_STATUS  = "https://api.vachana.ai/stt/v3/batch/status/{job_id}"
POLL_INTERVAL = 60
OUTPUT_DIR    = "outputs"

Path(OUTPUT_DIR).mkdir(exist_ok=True)


def submit_job(audio_paths: List[str], language_code: str = "hi-IN", itn: bool = True) -> str:
    api_key = os.getenv("GNANI_API_KEY")
    headers = {"X-API-Key-ID": api_key}

    files = [("audio_files", (Path(p).name, open(p, "rb"), "audio/wav")) for p in audio_paths]
    data  = {"language_code": language_code, "is_multi_channel": "false",
              "format": "transcribe" if itn else "verbatim"}

    resp = requests.post(BATCH_SUBMIT, headers=headers, files=files, data=data)
    resp.raise_for_status()

    for _, (_, fh, _) in files:
        fh.close()

    job_id = resp.json()["job_id"]
    print(f"Submitted {len(audio_paths)} file(s). job_id: {job_id}")
    return job_id


def poll_until_complete(job_id: str) -> Optional[list]:
    api_key = os.getenv("GNANI_API_KEY")
    headers = {"X-API-Key-ID": api_key}
    url     = BATCH_STATUS.format(job_id=job_id)

    print(f"Polling every {POLL_INTERVAL}s...")
    while True:
        time.sleep(POLL_INTERVAL)
        resp    = requests.get(url, headers=headers)
        resp.raise_for_status()
        payload = resp.json()
        status  = payload["status"]
        print(f"  [{status}]  {payload.get('overall_progress', '–')}%")
        if status == "completed":
            return payload.get("results", [])
        if status == "failed":
            print(f"Job failed: {payload.get('error')}")
            return None


def parse_results(results: list, output_dir: Path) -> Dict[str, dict]:
    outputs = {}

    for file_result in results:
        fname    = Path(file_result["filename"]).stem
        segments = file_result.get("segments", [])

        if not segments or file_result.get("status") == "failed":
            print(f"Skipping {fname}: {file_result.get('error', 'no segments')}")
            continue

        lines, speaker_times, segment_meta = [], {}, []

        for seg in segments:
            spk   = seg.get("speaker_id", "UNKNOWN")
            text  = seg.get("text", "").strip()
            start = seg.get("start_time", 0.0)
            end   = seg.get("end_time",   0.0)

            ts = f"{int(start // 60):02d}:{int(start % 60):02d}"
            lines.append(f"[{ts}] SPEAKER_{spk}: {text}")
            speaker_times[spk] = speaker_times.get(spk, 0.0) + (end - start)
            segment_meta.append({
                "segment_id":        seg.get("segment_id"),
                "speaker_id":        spk,
                "start_time":        start,
                "end_time":          end,
                "text":              text,
                "confidence":        seg.get("confidence"),
                "language_detected": seg.get("language_detected"),
            })

        transcript_path = output_dir / f"{fname}_transcript.txt"
        transcript_path.write_text("\n".join(lines), encoding="utf-8")

        metadata_path = output_dir / f"{fname}_metadata.json"
        metadata_path.write_text(json.dumps({
            "filename":          file_result["filename"],
            "total_duration_s":  file_result.get("total_duration"),
            "speaker_talk_time": {f"SPEAKER_{k}": round(v, 2) for k, v in speaker_times.items()},
            "segments":          segment_meta,
        }, indent=2, ensure_ascii=False), encoding="utf-8")

        outputs[fname] = {"transcript_path": str(transcript_path), "metadata_path": str(metadata_path)}
        print(f"Saved: {transcript_path.name}")

    return outputs


if __name__ == "__main__":
    job_id = submit_job(audio_paths=["/path/to/episode_01.wav"], language_code="hi-IN", itn=True)

    results = poll_until_complete(job_id)

    if results:
        output_dir = Path(OUTPUT_DIR) / f"job_{job_id}"
        output_dir.mkdir(parents=True, exist_ok=True)
        outputs = parse_results(results, output_dir)
        print(f"\nDone. {len(outputs)} transcript(s) saved to {output_dir}/")