dikte/api.py

"""OpenAI (transcription) and OpenRouter (cleanup) calls, stdlib only."""

import json
import mimetypes
import os
import secrets
import urllib.error
import urllib.request

from i18n import t

USER_AGENT = "dikte/1.0 (+https://github.com/yusufipk/dikte)"
OPENROUTER_URL = "https://openrouter.ai/api/v1"

# Only whisper-1 returns segment-level timestamps.
TIMESTAMP_MODEL = "whisper-1"


class ApiError(Exception):
    pass


def _request(url, data, headers, timeout=120):
    req = urllib.request.Request(url, data=data, headers=headers, method="POST")
    try:
        with urllib.request.urlopen(req, timeout=timeout) as resp:
            return json.loads(resp.read().decode("utf-8"))
    except urllib.error.HTTPError as exc:
        body = exc.read().decode("utf-8", "replace")
        raise ApiError(f"HTTP {exc.code}: {_extract_error(body)}") from exc
    except urllib.error.URLError as exc:
        raise ApiError(t("Could not connect: {reason}", reason=exc.reason)) from exc
    except json.JSONDecodeError as exc:
        raise ApiError(t("Could not parse the response: {error}", error=exc)) from exc


def _extract_error(body):
    try:
        payload = json.loads(body)
    except json.JSONDecodeError:
        return body[:300]
    err = payload.get("error")
    if isinstance(err, dict):
        return err.get("message") or json.dumps(err)[:300]
    if isinstance(err, str):
        return err
    return body[:300]


def _multipart(fields, file_field, file_path):
    """Build a multipart/form-data body; returns (body, content-type)."""
    boundary = "----dikte" + secrets.token_hex(16)
    out = bytearray()
    for name, value in fields:
        if value is None or value == "":
            continue
        out += f"--{boundary}\r\n".encode()
        out += f'Content-Disposition: form-data; name="{name}"\r\n\r\n'.encode()
        out += str(value).encode("utf-8") + b"\r\n"

    filename = os.path.basename(file_path)
    ctype = mimetypes.guess_type(filename)[0] or "application/octet-stream"
    with open(file_path, "rb") as fh:
        payload = fh.read()
    out += f"--{boundary}\r\n".encode()
    out += (
        f'Content-Disposition: form-data; name="{file_field}"; filename="{filename}"\r\n'
        f"Content-Type: {ctype}\r\n\r\n"
    ).encode()
    out += payload + b"\r\n"
    out += f"--{boundary}--\r\n".encode()
    return bytes(out), f"multipart/form-data; boundary={boundary}"


def _transcribe_request(wav_path, api_key, model, language, prompt, base_url,
                        response_format, granularity=None, timeout=300):
    if not api_key:
        raise ApiError(t("OpenAI API key is empty. Add it in Settings."))
    fields = [("model", model), ("response_format", response_format)]
    if language and language != "auto":
        fields.append(("language", language))
    if prompt:
        fields.append(("prompt", prompt))
    if granularity:
        fields.append(("timestamp_granularities[]", granularity))
    body, ctype = _multipart(fields, "file", wav_path)
    headers = {
        "Authorization": f"Bearer {api_key}",
        "Content-Type": ctype,
        "User-Agent": USER_AGENT,
    }
    return _request(
        f"{base_url.rstrip('/')}/audio/transcriptions", body, headers, timeout=timeout
    )


def transcribe(wav_path, api_key, model="gpt-4o-transcribe", language="", prompt="",
               base_url="https://api.openai.com/v1", timeout=300):
    data = _transcribe_request(
        wav_path, api_key, model, language, prompt, base_url, "json", timeout=timeout
    )
    text = (data.get("text") or "").strip()
    if not text:
        raise ApiError(t("Transcript came back empty."))
    return text


def transcribe_segments(wav_path, api_key, language="", prompt="",
                        base_url="https://api.openai.com/v1", timeout=300):
    """[(start_seconds, text)] using whisper-1's verbose response."""
    data = _transcribe_request(
        wav_path, api_key, TIMESTAMP_MODEL, language, prompt, base_url,
        "verbose_json", granularity="segment", timeout=timeout,
    )
    segments = data.get("segments") or []
    out = []
    for seg in segments:
        text = (seg.get("text") or "").strip()
        if text:
            out.append((float(seg.get("start") or 0.0), text))
    if not out:
        text = (data.get("text") or "").strip()
        if not text:
            raise ApiError(t("Transcript came back empty."))
        out = [(0.0, text)]
    return out


def cleanup(text, api_key, model, system_prompt, base_url=OPENROUTER_URL, timeout=180):
    if not api_key:
        raise ApiError(t("OpenRouter API key is empty. Add it in Settings."))
    payload = {
        "model": model,
        "temperature": 0,
        "messages": [
            {"role": "system", "content": system_prompt},
            {"role": "user", "content": f"<transcript>\n{text}\n</transcript>"},
        ],
    }
    headers = {
        "Authorization": f"Bearer {api_key}",
        "Content-Type": "application/json",
        "User-Agent": USER_AGENT,
        "HTTP-Referer": "https://github.com/yusufipk/dikte",
        "X-Title": "Dikte",
    }
    data = _request(
        f"{base_url.rstrip('/')}/chat/completions",
        json.dumps(payload).encode("utf-8"),
        headers,
        timeout=timeout,
    )
    choices = data.get("choices") or []
    if not choices:
        raise ApiError(_extract_error(json.dumps(data)))
    content = ((choices[0].get("message") or {}).get("content") or "").strip()
    if not content:
        raise ApiError(t("The cleanup model returned an empty reply."))
    return content


def _get_json(url, headers, timeout=20):
    req = urllib.request.Request(url, headers=headers)
    try:
        with urllib.request.urlopen(req, timeout=timeout) as resp:
            return json.loads(resp.read().decode("utf-8"))
    except urllib.error.HTTPError as exc:
        body = exc.read().decode("utf-8", "replace")
        raise ApiError(f"HTTP {exc.code}: {_extract_error(body)}") from exc
    except urllib.error.URLError as exc:
        raise ApiError(t("Could not connect: {reason}", reason=exc.reason)) from exc
    except json.JSONDecodeError as exc:
        raise ApiError(t("Could not parse the response: {error}", error=exc)) from exc


def openrouter_models(api_key=""):
    """Model ids available on OpenRouter (no key required)."""
    headers = {"User-Agent": USER_AGENT}
    if api_key:
        headers["Authorization"] = f"Bearer {api_key}"
    data = _get_json(f"{OPENROUTER_URL}/models", headers)
    return sorted(m["id"] for m in data.get("data", []) if m.get("id"))


def openai_models(api_key, base_url="https://api.openai.com/v1"):
    if not api_key:
        raise ApiError(t("OpenAI API key is empty. Add it in Settings."))
    data = _get_json(
        f"{base_url.rstrip('/')}/models",
        {"Authorization": f"Bearer {api_key}", "User-Agent": USER_AGENT},
    )
    ids = [m["id"] for m in data.get("data", []) if m.get("id")]
    audio = [i for i in ids if "transcribe" in i or "whisper" in i]
    return sorted(audio or ids)