"""HybridAI client: one file, no installs (Python 3.8+).

Copyright (c) 2026 HybridAI (hybridai.cloud). MIT License: https://hybridai.cloud/sdk/LICENSE.txt
SPDX-License-Identifier: MIT

Get it:   curl -O https://hybridai.cloud/sdk/hybridai.py
Use it:   from hybridai import HybridAI
          ai = HybridAI("YOUR_KEY")          # or set the HYBRIDAI_KEY environment variable
          print(ai.chat("Say hi in 5 words"))

Docs: https://hybridai.cloud/devkit
"""
import json
import mimetypes
import os
import time
import urllib.error
import urllib.parse
import urllib.request
import uuid

__version__ = "1.5.0"
DEFAULT_BASE = "https://api.hybridai.cloud"


class HybridAIError(Exception):
    """Raised when HybridAI answers with an error. Has .status, .message and .body."""

    def __init__(self, status, message, body=None):
        super().__init__("HTTP %s: %s" % (status, message))
        self.status = status
        self.message = message
        self.body = body


class Reply(object):
    """A chat answer. str(reply) is the text. The other fields say where it went."""

    def __init__(self, text, headers, raw):
        self.text = text
        self.route = headers.get("x-hybrid-route", "")
        self.model = headers.get("x-hybrid-model", "")
        self.privacy = headers.get("x-hybrid-privacy", "")
        self.why = headers.get("x-hybrid-why", "")
        self.receipt = headers.get("x-hybrid-receipt", "")
        self.output_guard = headers.get("x-hybrid-output-guard", "")   # e.g. "redact: email" when something was hidden
        self.headers = headers
        self.raw = raw

    def __str__(self):
        return self.text

    def __repr__(self):
        return "Reply(route=%r, model=%r, privacy=%r, text=%r)" % (self.route, self.model, self.privacy, self.text[:60])


def restore(text, mapping):
    """Put hidden values back into text. Works offline."""
    for placeholder, value in (mapping or {}).items():
        text = text.replace(placeholder, value)
    return text


def verify_webhook(secret, signature, body, tolerance=300, now=None):
    """Checks a HybridAI webhook on YOUR server. Pass your signing secret, the x-hybridai-signature header and
    the raw request body (bytes or str). Returns True only when the signature matches and is recent."""
    import hashlib, hmac, time
    if isinstance(body, str):
        body = body.encode("utf-8")
    parts = dict(p.split("=", 1) for p in str(signature or "").split(",") if "=" in p)
    t, v1 = parts.get("t", ""), parts.get("v1", "")
    if not t.isdigit() or not v1:
        return False
    if abs((now if now is not None else time.time()) - int(t)) > tolerance:
        return False
    want = hmac.new(secret.encode("utf-8"), t.encode("ascii") + b"." + body, hashlib.sha256).hexdigest()
    return hmac.compare_digest(want, v1)


class Stream(object):
    """A streamed chat answer. Loop over it to get the text as it arrives:

        for piece in ai.chat_stream("Tell me a story"):
            print(piece, end="", flush=True)

    Afterwards .text holds the whole answer. .route, .model, .privacy, .why and .receipt say where it went."""

    def __init__(self, resp):
        self._resp = resp
        self.headers = {k.lower(): v for k, v in resp.headers.items()}
        self.route = self.headers.get("x-hybrid-route", "")
        self.model = self.headers.get("x-hybrid-model", "")
        self.privacy = self.headers.get("x-hybrid-privacy", "")
        self.why = self.headers.get("x-hybrid-why", "")
        self.receipt = self.headers.get("x-hybrid-receipt", "")
        self.output_guard = self.headers.get("x-hybrid-output-guard", "")
        self.text = ""
        self.usage = None
        self.finish_reason = None

    def __iter__(self):
        try:
            for raw in self._resp:
                line = raw.decode("utf-8", "replace").strip()
                if not line.startswith("data:"):
                    continue
                data = line[5:].strip()
                if data == "[DONE]":
                    break
                try:
                    event = json.loads(data)
                except ValueError:
                    continue
                if event.get("error"):
                    raise HybridAIError(502, (event["error"] or {}).get("message", "The stream failed."), data)
                if event.get("usage"):
                    self.usage = event["usage"]
                for choice in event.get("choices") or []:
                    piece = (choice.get("delta") or {}).get("content") or ""
                    if piece:
                        self.text += piece
                        yield piece
                    if choice.get("finish_reason"):
                        self.finish_reason = choice["finish_reason"]
        finally:
            self.close()

    def read(self):
        """Waits for the whole answer and returns it."""
        for _ in self:
            pass
        return self.text

    def close(self):
        try:
            self._resp.close()
        except Exception:
            pass

    def __enter__(self):
        return self

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


class HybridAI(object):
    def __init__(self, key=None, base=None, timeout=60, byok=None):
        """key: your HybridAI key. base: only if you run your own copy.
        byok: your own provider keys for one request each, for example {"groq": "gsk_..."}."""
        self.key = (key or os.environ.get("HYBRIDAI_KEY") or "").strip()
        if not self.key:
            raise ValueError("Pass your key, or set the HYBRIDAI_KEY environment variable.")
        self.base = (base or os.environ.get("HYBRIDAI_BASE") or DEFAULT_BASE).rstrip("/")
        self.timeout = timeout
        self.byok = dict(byok or {})

    # ---- plumbing
    def _headers(self, extra=None, byok_only=False):
        h = {"Authorization": "Bearer " + self.key, "User-Agent": "hybridai-py/" + __version__}
        for name, value in self.byok.items():
            h["x-byok-" + name] = value
        if byok_only:
            h["x-byok-only"] = "1"
        h.update(extra or {})
        return h

    def _open(self, method, path, body=None, data=None, headers=None, byok_only=False, timeout=None):
        hdrs = self._headers(headers, byok_only)
        payload = data
        if body is not None:
            payload = json.dumps(body).encode("utf-8")
            hdrs["Content-Type"] = "application/json"
        req = urllib.request.Request(self.base + path, data=payload, method=method, headers=hdrs)
        try:
            return urllib.request.urlopen(req, timeout=timeout or self.timeout)
        except urllib.error.HTTPError as e:
            raw = e.read()
            message = ""
            try:
                message = json.loads(raw.decode("utf-8")).get("error", {}).get("message", "")
            except Exception:
                pass
            raise HybridAIError(e.code, message or e.reason, raw.decode("utf-8", "replace"))
        except urllib.error.URLError as e:
            raise HybridAIError(0, "Could not reach " + self.base + ": " + str(e.reason))

    def _call(self, method, path, **kw):
        with self._open(method, path, **kw) as r:
            return r.status, {k.lower(): v for k, v in r.headers.items()}, r.read()

    def _json(self, method, path, body=None, **kw):
        _, headers, raw = self._call(method, path, body=body, **kw)
        return json.loads(raw.decode("utf-8")), headers

    # ---- chat
    def chat(self, prompt, system=None, model="hybridai-auto", max_tokens=None, temperature=None,
             cache=True, byok_only=False, output_guard=None, **extra):
        """prompt: text, or a list of {"role": ..., "content": ...} messages. Returns a Reply.
        output_guard="redact" hides emails, phones, cards and keys the model writes; "block" holds the whole answer back."""
        if isinstance(prompt, str):
            messages = [{"role": "user", "content": prompt}]
        else:
            messages = list(prompt)
        if system:
            messages = [{"role": "system", "content": system}] + messages
        body = {"model": model, "messages": messages}
        if max_tokens is not None:
            body["max_tokens"] = max_tokens
        if temperature is not None:
            body["temperature"] = temperature
        body.update(extra)
        headers = {} if cache else {"x-hybrid-cache": "off"}
        if output_guard:
            headers["x-output-guard"] = output_guard
        data, h = self._json("POST", "/v1/chat/completions", body, headers=headers, byok_only=byok_only,
                             timeout=max(self.timeout, 120) if (self.byok or byok_only) else None)
        text = ((data.get("choices") or [{}])[0].get("message") or {}).get("content") or ""
        return Reply(text, h, data)

    def chat_stream(self, prompt, system=None, model="hybridai-auto", max_tokens=None, temperature=None,
                    cache=True, byok_only=False, include_usage=False, output_guard=None, **extra):
        """Like chat, but the answer arrives piece by piece. Returns a Stream you loop over.
        Plain cloud answers stream as the provider writes them; saved answers, private (Cloudflare) answers and
        requests with tools arrive whole, in small pieces."""
        if isinstance(prompt, str):
            messages = [{"role": "user", "content": prompt}]
        else:
            messages = list(prompt)
        if system:
            messages = [{"role": "system", "content": system}] + messages
        body = {"model": model, "messages": messages, "stream": True}
        if include_usage:
            body["stream_options"] = {"include_usage": True}
        if max_tokens is not None:
            body["max_tokens"] = max_tokens
        if temperature is not None:
            body["temperature"] = temperature
        body.update(extra)
        headers = {} if cache else {"x-hybrid-cache": "off"}
        if output_guard:
            headers["x-output-guard"] = output_guard
        resp = self._open("POST", "/v1/chat/completions", body=body, headers=headers, byok_only=byok_only,
                          timeout=max(self.timeout, 120) if (self.byok or byok_only) else None)
        return Stream(resp)

    def webhook(self):
        """Your webhook address and the last delivery result. The owner must have switched webhooks on for your key."""
        return self._json("GET", "/v1/webhook")[0]

    def set_webhook(self, url):
        """Sets the https address. The first time, the result holds a signing secret (shown once): keep it."""
        return self._json("PUT", "/v1/webhook", {"url": url})[0]

    def delete_webhook(self):
        return self._json("DELETE", "/v1/webhook")[0]

    def test_webhook(self):
        """Sends a test ping now. Returns {"delivered": bool, "status": int, "error": ...}."""
        return self._json("POST", "/v1/webhook/test")[0]

    def rotate_webhook_secret(self):
        """Makes a new signing secret (the old one stops working) and returns it once."""
        return self._json("POST", "/v1/webhook/secret")[0]

    def fallback(self):
        """Your saved last-resort endpoint (host, model, last four key characters). The owner must have switched custom endpoints on for your key."""
        return self._json("GET", "/v1/fallback")[0]

    def set_fallback(self, url, key, model):
        """Saves your own OpenAI-compatible endpoint. HybridAI tries it last, after every free provider has failed."""
        return self._json("PUT", "/v1/fallback", {"url": url, "key": key, "model": model})[0]

    def delete_fallback(self):
        return self._json("DELETE", "/v1/fallback")[0]

    def test_fallback(self):
        """Sends one tiny message to your saved endpoint. Limited to 10 a day."""
        return self._json("POST", "/v1/fallback/test")[0]

    def agent_keys(self):
        """Your agent keys: status (active, locked, revoked, expired), today's use and why a key locked. The owner must have switched agent keys on for your key."""
        return self._json("GET", "/v1/agents")[0]["data"]

    def create_agent_key(self, name, daily_limit=None, burst_per_min=None, expires_in_minutes=None):
        """Makes a key for one script or agent. The result holds the new key once (result["key"]). It locks itself if it sends
        more than burst_per_min requests a minute (default 30) or keeps sending blocked prompts."""
        body = {"name": name}
        if daily_limit is not None:
            body["daily_limit"] = daily_limit
        if burst_per_min is not None:
            body["burst_per_min"] = burst_per_min
        if expires_in_minutes is not None:
            body["expires_in_minutes"] = expires_in_minutes
        return self._json("POST", "/v1/agents", body)[0]

    def revoke_agent_key(self, agent_id):
        return self._json("POST", "/v1/agents/" + str(int(agent_id)) + "/revoke")[0]

    def unlock_agent_key(self, agent_id):
        """Lets a locked agent key work again."""
        return self._json("POST", "/v1/agents/" + str(int(agent_id)) + "/unlock")[0]

    def chat_private(self, prompt, strict=True, **kw):
        """Hide private details on your side, ask, and put them back in the answer.
        The model only ever sees placeholders such as [EMAIL_1]."""
        hidden = self.redact(prompt, strict=strict)
        if not hidden["complete"]:
            raise HybridAIError(422, "Some private details could not be hidden safely, so nothing was sent.")
        reply = self.chat(hidden["redacted"], **kw)
        reply.redacted_prompt = hidden["redacted"]
        reply.found = hidden["found"]
        reply.model_text = reply.text
        reply.text = restore(reply.text, hidden["map"])
        return reply

    def models(self):
        return self._json("GET", "/v1/models")[0]

    def usage(self):
        return self._json("GET", "/v1/usage")[0]

    # ---- redact
    def redact(self, text, strict=False):
        """Returns a dict: redacted, found, map, complete. strict=True also hides spoken digits,
        passwords, names and addresses that follow phrases like 'my password is'."""
        body = {"text": text}
        if strict:
            body["strict"] = True
        return self._json("POST", "/v1/redact", body)[0]

    restore = staticmethod(restore)

    # ---- voice
    def transcribe(self, audio, redact=False, language=None, filename=None):
        """audio: a file path or bytes. Returns {"text": ...} (and redacted, found, map, complete if redact=True)."""
        if isinstance(audio, (bytes, bytearray)):
            blob, name = bytes(audio), filename or "audio.mp3"
        else:
            with open(audio, "rb") as f:
                blob = f.read()
            name = filename or os.path.basename(audio)
        fields = {}
        if redact:
            fields["redact"] = "true"
        if language:
            fields["language"] = language
        boundary = uuid.uuid4().hex
        parts = []
        for k, v in fields.items():
            parts.append(("--%s\r\nContent-Disposition: form-data; name=\"%s\"\r\n\r\n%s\r\n" % (boundary, k, v)).encode())
        ctype = mimetypes.guess_type(name)[0] or "application/octet-stream"
        parts.append(("--%s\r\nContent-Disposition: form-data; name=\"file\"; filename=\"%s\"\r\nContent-Type: %s\r\n\r\n"
                      % (boundary, name.replace('"', ""), ctype)).encode() + blob + b"\r\n")
        parts.append(("--%s--\r\n" % boundary).encode())
        data = b"".join(parts)
        return self._json("POST", "/v1/audio/transcriptions", data=data,
                          headers={"Content-Type": "multipart/form-data; boundary=" + boundary}, timeout=max(self.timeout, 120))[0]

    def speak(self, text, out_path=None, lang="en", voice=None, redact=False):
        """Returns mp3 bytes. With redact=True, private details are spoken as 'email address removed' and so on."""
        body = {"input": text, "lang": lang}
        if voice:
            body["voice"] = voice
        if redact:
            body["redact"] = True
        _, _, audio = self._call("POST", "/v1/audio/speech", body=body, timeout=max(self.timeout, 120))
        if out_path:
            with open(out_path, "wb") as f:
                f.write(audio)
        return audio

    # ---- approvals (a person says yes or no)
    def create_approval(self, title, details="", expires_in_minutes=None):
        body = {"title": title, "details": details}
        if expires_in_minutes is not None:
            body["expires_in_minutes"] = expires_in_minutes
        return self._json("POST", "/v1/approvals", body)[0]

    def get_approval(self, approval_id):
        return self._json("GET", "/v1/approvals/" + approval_id)[0]

    def cancel_approval(self, approval_id):
        return self._json("POST", "/v1/approvals/" + approval_id + "/cancel")[0]

    def list_approvals(self):
        return self._json("GET", "/v1/approvals")[0]["data"]

    def wait_for_approval(self, approval_id, timeout=600, poll=3):
        """Waits until a person decides (or it expires). Returns the final record."""
        end = time.time() + timeout
        while True:
            a = self.get_approval(approval_id)
            if a["status"] != "pending":
                return a
            if time.time() >= end:
                return a
            time.sleep(poll)

    def ask_approval(self, title, details="", expires_in_minutes=60, wait=True, timeout=None, poll=3, show=print):
        """Ask a person. Prints a link to open on any device, waits, and returns the record with
        an extra key "approved" (True only if a person approved)."""
        a = self.create_approval(title, details, expires_in_minutes)
        if show:
            show("Open to decide: " + a["review_url"])
        if not wait:
            a["approved"] = False
            return a
        final = self.wait_for_approval(a["id"], timeout=timeout or expires_in_minutes * 60, poll=poll)
        final["approved"] = final["status"] == "approved"
        final["review_url"] = a["review_url"]
        return final

    # ---- private memory (encrypted notes only your key can read)
    def remember(self, name, value, expires_in_minutes=None, tags=None):
        """Saves a note. value can be text or anything JSON can hold (up to 16 KB).
        tags: up to 5 short words (letters, numbers . _ : -) to find the note later. Leave tags out to keep the
        note's existing tags, pass a new list to replace them, or tags=[] to clear them."""
        body = {"value": value}
        if expires_in_minutes is not None:
            body["expires_in_minutes"] = expires_in_minutes
        if tags is not None:
            body["tags"] = list(tags)
        return self._json("PUT", "/v1/memory/" + urllib.parse.quote(name, safe="._:-"), body)[0]

    def recall(self, name, default=None):
        """Returns the note's value, or default if there is no such note."""
        try:
            return self._json("GET", "/v1/memory/" + urllib.parse.quote(name, safe="._:-"))[0]["value"]
        except HybridAIError as e:
            if e.status == 404:
                return default
            raise

    def forget(self, name):
        """Deletes one note. Returns True if it existed."""
        return self._json("DELETE", "/v1/memory/" + urllib.parse.quote(name, safe="._:-"))[0]["deleted"]

    def memories(self, tag=None):
        """Lists your notes (names, tags, sizes, times). The text is not included; use recall.
        With tag="work" only notes carrying that tag are listed."""
        path = "/v1/memory" + ("?tag=" + urllib.parse.quote(tag, safe="") if tag else "")
        return self._json("GET", path)[0]["data"]

    def note_tags(self, name):
        """Returns the tags of one note, or None if there is no such note."""
        try:
            return self._json("GET", "/v1/memory/" + urllib.parse.quote(name, safe="._:-"))[0].get("tags", [])
        except HybridAIError as e:
            if e.status == 404:
                return None
            raise

    def forget_all(self):
        """Erases every note for this key. Returns how many were deleted."""
        return self._json("DELETE", "/v1/memory?all=yes")[0]["deleted"]

    # ---- public pages (no key needed on the server side, but the client sends yours anyway)
    def status(self):
        """Which free models worked in real use over the last week."""
        return self._json("GET", "/api/status")[0]

    def transparency(self):
        """Which provider answered what share of requests over the last 7 days."""
        return self._json("GET", "/api/transparency")[0]

    # ---- receipts
    def verify_receipt(self, receipt):
        data = json.dumps({"receipt": receipt}).encode("utf-8")
        return self._json("POST", "/api/receipt/verify", data=data, headers={"Content-Type": "application/json"})[0]