"""A small Python client for YouTubeTranscriptAPI (https://youtubetranscriptapi.com).

    pip install httpx
    export YTT_API_KEY=tk_live_...

    from ytt_client import TranscriptClient, NoTranscript

    with TranscriptClient() as yt:
        data = yt.transcript("https://www.youtube.com/watch?v=aircAruvnKk")
        print(data["transcript"][0]["text"], yt.credits_remaining)

    # many videos at once
    import asyncio
    from ytt_client import AsyncTranscriptClient

    async def main():
        async with AsyncTranscriptClient() as yt:
            results = await yt.transcripts(["aircAruvnKk", "jNQXAC9IVRw"], concurrency=10)
    asyncio.run(main())

What it does for you:
  * a 60 s timeout on every call (a video nobody fetched before can take 10-30 s);
  * retries on 408, 429, 500 and 503, honouring Retry-After (seconds or a date),
    with exponential backoff and jitter otherwise;
  * typed errors: NoTranscript (404, not charged), OutOfCredits (402), AuthError
    (401), BadRequest (422);
  * credits_remaining and rate_limit_remaining, read from the response headers;
  * an async batch helper that caps concurrency and paces requests at 280 a
    minute by default, leaving headroom under the 300-a-minute rate limit.
"""

import asyncio
import os
import random
import time
from email.utils import parsedate_to_datetime

import httpx

BASE_URL = os.environ.get("YTT_API_BASE", "https://youtubetranscriptapi.com/api/v2")
RETRYABLE = {408, 429, 500, 503}


class TranscriptAPIError(Exception):
    def __init__(self, status, code, message, body=None):
        super().__init__(f"HTTP {status} {code}: {message}")
        self.status, self.code, self.message, self.body = status, code, message, body


class NoTranscript(TranscriptAPIError):
    """404: no captions, private or removed video, or unknown channel/playlist. Never charged."""


class OutOfCredits(TranscriptAPIError):
    """402: not enough credits. `action_url` points to the top-up or plan page."""

    @property
    def action_url(self):
        return self.body.get("detail", {}).get("action_url") if isinstance(self.body, dict) else None


class AuthError(TranscriptAPIError):
    """401: missing, malformed or revoked API key."""


class BadRequest(TranscriptAPIError):
    """422: a parameter is missing or invalid."""


ERRORS = {401: AuthError, 402: OutOfCredits, 404: NoTranscript, 422: BadRequest}


def _retry_delay(resp, attempt):
    """Seconds to wait before the next try: Retry-After if present, else backoff with jitter."""
    value = resp.headers.get("Retry-After") if resp is not None else None
    if value:
        try:
            return min(max(float(value), 0.0), 60.0)
        except ValueError:
            try:
                return min(max(parsedate_to_datetime(value).timestamp() - time.time(), 0.0), 60.0)
            except (TypeError, ValueError):
                pass
    return min(2 ** attempt, 30) + random.uniform(0, 0.5)


def _raise_for(resp):
    try:
        body = resp.json()
    except ValueError:
        body = {"detail": resp.text[:200]}
    detail = body.get("detail") if isinstance(body, dict) else body
    message = detail.get("message") if isinstance(detail, dict) else str(detail)
    code = body.get("code", "") if isinstance(body, dict) else ""
    raise ERRORS.get(resp.status_code, TranscriptAPIError)(resp.status_code, code, message, body)


def _bool(value):
    return "true" if value else "false"


def _transcript_params(video, language, text, timestamps, metadata):
    return {"video_url": video, "language": language, "format": "text" if text else "json",
            "include_timestamp": _bool(timestamps), "send_metadata": _bool(metadata)}


class _Base:
    def __init__(self, api_key=None, timeout=60.0, max_retries=4, base_url=BASE_URL):
        self.api_key = api_key or os.environ.get("YTT_API_KEY")
        if not self.api_key:
            raise AuthError(401, "missing_auth", "pass api_key or set YTT_API_KEY")
        self.base_url, self.timeout, self.max_retries = base_url.rstrip("/"), timeout, max_retries
        self.credits_remaining = None
        self.rate_limit_remaining = None
        self.credits_used = 0

    @property
    def _headers(self):
        return {"Authorization": f"Bearer {self.api_key}", "User-Agent": "ytt-client-python/1.0"}

    def _note(self, resp):
        h = resp.headers
        if h.get("X-Credits-Remaining") is not None:
            self.credits_remaining = int(h["X-Credits-Remaining"])
        if h.get("X-RateLimit-Remaining") is not None:
            self.rate_limit_remaining = int(h["X-RateLimit-Remaining"])
        self.credits_used += int(h.get("X-Credits-Charged", 0) or 0)


class TranscriptClient(_Base):
    """Blocking client. Use it as a context manager so the connection pool is closed."""

    def __init__(self, *a, **kw):
        super().__init__(*a, **kw)
        self._http = httpx.Client(base_url=self.base_url, headers=self._headers, timeout=self.timeout)

    def __enter__(self):
        return self

    def __exit__(self, *exc):
        self.close()

    def close(self):
        self._http.close()

    def get(self, path, **params):
        resp = None
        for attempt in range(self.max_retries + 1):
            try:
                resp = self._http.get(path, params=params)
            except httpx.TransportError:
                if attempt == self.max_retries:
                    raise
                time.sleep(_retry_delay(None, attempt))
                continue
            if resp.status_code not in RETRYABLE or attempt == self.max_retries:
                break
            time.sleep(_retry_delay(resp, attempt))
        self._note(resp)
        if resp.status_code != 200:
            _raise_for(resp)
        return resp.json()

    def transcript(self, video, language="en", text=False, timestamps=True, metadata=False):
        """1 credit. Raises NoTranscript (free) when the video has no captions."""
        return self.get("/youtube/transcript", **_transcript_params(video, language, text, timestamps, metadata))

    def search(self, q, type="video", limit=20):
        return self.get("/youtube/search", q=q, type=type, limit=limit)  # 4 credits

    def channel_search(self, channel, q, limit=30):
        return self.get("/youtube/channel/search", channel=channel, q=q, limit=limit)  # 1 credit

    def channel_videos(self, channel):
        return self.get("/youtube/channel/videos", channel=channel)  # 20 credits, newest 100

    def channel_latest(self, channel):
        return self.get("/youtube/channel/latest", channel=channel)  # free, newest 15

    def playlist_videos(self, playlist):
        return self.get("/youtube/playlist/videos", playlist=playlist)  # 20 credits, first 100

    def whoami(self):
        return self.get("/account/whoami")  # free: plan and balance


class AsyncTranscriptClient(_Base):
    """Async client. Share one instance across tasks; it reuses one connection pool."""

    def __init__(self, *a, per_minute=280, **kw):  # headroom under the 300/min limit
        super().__init__(*a, **kw)
        self._http = httpx.AsyncClient(base_url=self.base_url, headers=self._headers, timeout=self.timeout)
        self._interval = 60.0 / per_minute if per_minute else 0.0
        self._next_slot = 0.0
        self._pace_lock = asyncio.Lock()

    async def __aenter__(self):
        return self

    async def __aexit__(self, *exc):
        await self.aclose()

    async def aclose(self):
        await self._http.aclose()

    async def _pace(self):
        """Space request starts evenly so a burst never exceeds the per-minute limit."""
        async with self._pace_lock:
            now = time.monotonic()
            wait = self._next_slot - now
            self._next_slot = max(now, self._next_slot) + self._interval
        if wait > 0:
            await asyncio.sleep(wait)

    async def get(self, path, _sem=None, _stop=None, **params):
        resp = None
        for attempt in range(self.max_retries + 1):
            await self._pace()
            try:
                if _sem:
                    async with _sem:  # hold a slot only while the request is in flight
                        if _stop is not None and _stop.is_set():
                            raise OutOfCredits(402, "insufficient_credits", "skipped after an earlier 402")
                        resp = await self._http.get(path, params=params)
                else:
                    resp = await self._http.get(path, params=params)
            except httpx.TransportError:
                if attempt == self.max_retries:
                    raise
                await asyncio.sleep(_retry_delay(None, attempt))
                continue
            if resp.status_code not in RETRYABLE or attempt == self.max_retries:
                break
            await asyncio.sleep(_retry_delay(resp, attempt))  # sleep outside the slot
        self._note(resp)
        if resp.status_code != 200:
            _raise_for(resp)
        return resp.json()

    async def transcript(self, video, language="en", text=False, timestamps=True, metadata=False, _sem=None, _stop=None):
        return await self.get("/youtube/transcript", _sem=_sem, _stop=_stop,
                              **_transcript_params(video, language, text, timestamps, metadata))

    async def transcripts(self, videos, concurrency=10, stop_on_402=True, **kw):
        """Fetch many transcripts. Returns {video: result_dict or exception}, in input order.

        Duplicates are fetched once. After a 402 the remaining videos are skipped (left as
        OutOfCredits) unless stop_on_402=False, so you don't burn requests on an empty balance.
        """
        sem = asyncio.Semaphore(concurrency)
        broke = asyncio.Event()
        unique = list(dict.fromkeys(videos))

        async def one(v):
            if broke.is_set():
                return v, OutOfCredits(402, "insufficient_credits", "skipped after an earlier 402")
            try:
                return v, await self.transcript(v, _sem=sem, _stop=broke if stop_on_402 else None, **kw)
            except OutOfCredits as e:
                if stop_on_402:
                    broke.set()
                return v, e
            except (TranscriptAPIError, httpx.HTTPError) as e:
                return v, e

        done = dict(await asyncio.gather(*(one(v) for v in unique)))
        return {v: done[v] for v in unique}

    async def search(self, q, type="video", limit=20):
        return await self.get("/youtube/search", q=q, type=type, limit=limit)

    async def channel_search(self, channel, q, limit=30):
        return await self.get("/youtube/channel/search", channel=channel, q=q, limit=limit)

    async def channel_videos(self, channel):
        return await self.get("/youtube/channel/videos", channel=channel)

    async def channel_latest(self, channel):
        return await self.get("/youtube/channel/latest", channel=channel)

    async def playlist_videos(self, playlist):
        return await self.get("/youtube/playlist/videos", playlist=playlist)

    async def whoami(self):
        return await self.get("/account/whoami")


if __name__ == "__main__":
    import sys

    if len(sys.argv) < 2:
        sys.exit("usage: python ytt_client.py VIDEO_ID_OR_URL [...]")
    with TranscriptClient() as yt:
        for v in sys.argv[1:]:
            try:
                print(yt.transcript(v, text=True, timestamps=False)["transcript"][:300], "...\n")
            except NoTranscript:
                print(f"{v}: no captions (not charged)\n")
        print(f"credits used: {yt.credits_used}, remaining: {yt.credits_remaining}")