transcriptfetch_api.py

"""A small client for the TranscriptFetch v2 API.

Every request goes to one fixed origin (API_BASE), with a finite timeout and
redirects not followed. The API key is sent only as an Authorization header
and never appears in an error message.
"""

from __future__ import annotations

import time
from typing import Any, Callable, Optional

import httpx

API_BASE = "https://transcriptfetch.com/api/v2"
USER_AGENT = "transcriptfetch-dify-plugin/0.1.0"

# One request: 10 s to connect. Transcripts of short media can be held open by
# the API for up to 45 s while they are transcribed, so reads get 60 s.
TIMEOUT = httpx.Timeout(60.0, connect=10.0)

# How long get_transcript waits for an AI transcription job before handing back
# the job id, and how often it checks. Dify gives a tool call 120 s in total.
JOB_WAIT_SECONDS = 75.0
JOB_POLL_SECONDS = 5.0


class TranscriptFetchError(ValueError):
    """An API error, with the API's own code and message."""

    def __init__(self, message: str, code: Optional[str] = None, status: Optional[int] = None):
        super().__init__(message)
        self.code = code
        self.status = status


def _message(status: int, payload: Any) -> TranscriptFetchError:
    error = payload.get("error") if isinstance(payload, dict) else None
    if isinstance(error, dict):
        code = error.get("code")
        text = error.get("message") or "Request failed."
        docs = error.get("docs")
        detail = f"{code}: {text}" if code else text
        if docs:
            detail += f" ({docs})"
        return TranscriptFetchError(detail, code=code, status=status)
    if status == 401:
        return TranscriptFetchError("Invalid or missing API key.", code="unauthorized", status=status)
    return TranscriptFetchError(f"TranscriptFetch answered HTTP {status}.", status=status)


class TranscriptFetchClient:
    def __init__(self, api_key: Optional[str], *, transport: Optional[httpx.BaseTransport] = None,
                 sleep: Optional[Callable[[float], None]] = None, clock: Optional[Callable[[], float]] = None):
        key = (api_key or "").strip()
        if not key:
            raise TranscriptFetchError("A TranscriptFetch API key is required.", code="unauthorized")
        self._client = httpx.Client(
            base_url=API_BASE,
            timeout=TIMEOUT,
            follow_redirects=False,
            headers={"Authorization": f"Bearer {key}", "User-Agent": USER_AGENT, "Accept": "application/json"},
            transport=transport,
        )
        self._sleep = sleep or time.sleep
        self._clock = clock or time.monotonic

    def close(self) -> None:
        self._client.close()

    def __enter__(self) -> "TranscriptFetchClient":
        return self

    def __exit__(self, *exc: Any) -> None:
        self.close()

    def _request(self, method: str, path: str, body: Optional[dict[str, Any]] = None) -> tuple[int, dict[str, Any]]:
        try:
            response = self._client.request(method, path, json=body)
        except httpx.TimeoutException:
            raise TranscriptFetchError("TranscriptFetch did not answer in time. Try again.", code="timeout") from None
        except httpx.HTTPError:
            raise TranscriptFetchError("Could not reach TranscriptFetch. Try again.", code="network") from None
        try:
            payload = response.json()
        except ValueError:
            payload = None
        if response.status_code >= 300 or not isinstance(payload, dict) or payload.get("ok") is False:
            raise _message(response.status_code, payload)
        return response.status_code, payload

    def me(self) -> dict[str, Any]:
        _, payload = self._request("GET", "/me")
        return payload.get("data") or {}

    def search(self, query: str, platform: str, limit: int) -> dict[str, Any]:
        _, payload = self._request("POST", "/transcripts/search", {"query": query, "platform": platform, "limit": limit})
        return _with_usage(payload)

    def channel(self, channel: str, limit: int, cursor: Optional[str] = None) -> dict[str, Any]:
        body: dict[str, Any] = {"channel": channel, "limit": limit}
        if cursor:
            body["cursor"] = cursor
        _, payload = self._request("POST", "/transcripts/channel", body)
        return _with_usage(payload)

    def transcript(self, video: str, mode: str, timestamps: bool) -> dict[str, Any]:
        """Fetch a transcript; when the API starts an AI transcription job, wait for it.

        The wait is bounded by JOB_WAIT_SECONDS. A job still running after that
        comes back as {"status": "processing", "job_id": ...}: calling again
        with the same video returns the finished transcript from the cache and
        costs credits only once.
        """
        status, payload = self._request(
            "POST", "/transcripts/video", {"video": video, "mode": mode, "timestamps": timestamps}
        )
        job_id = _job_id(payload)
        if status != 202 and not job_id:
            return _with_usage(payload)

        deadline = self._clock() + JOB_WAIT_SECONDS
        last = payload
        while job_id and self._clock() < deadline:
            self._sleep(JOB_POLL_SECONDS)
            _, last = self._request("GET", f"/transcripts/jobs/{job_id}")
            state = last.get("status")
            if state == "completed":
                result = _with_usage(last)
                if not timestamps:
                    result.pop("segments", None)
                return result
            if state == "failed":
                raise _message(422, last)
        return {
            "status": "processing",
            "job_id": job_id,
            "video": video,
            "message": (
                "AI transcription is still running. Call get_transcript again with the same video in a "
                "minute or two: the finished transcript is cached, so it costs credits only once."
            ),
            "progress": last.get("progress"),
        }


def _job_id(payload: dict[str, Any]) -> Optional[str]:
    data = payload.get("data") if isinstance(payload.get("data"), dict) else {}
    return payload.get("job_id") or data.get("job_id")


def _with_usage(payload: dict[str, Any]) -> dict[str, Any]:
    data = dict(payload.get("data") or {})
    usage = payload.get("usage")
    if isinstance(usage, dict):
        data["usage"] = {k: usage[k] for k in ("credits_spent", "balance") if k in usage}
    return data


def clamp_limit(value: Any, default: int = 10, maximum: int = 50) -> int:
    try:
        number = int(value)
    except (TypeError, ValueError):
        return default
    return max(1, min(maximum, number))