"""
kirelta_client -- the official Python SDK for Kirelta.

Copyright (c) 2026 Kirelta. All rights reserved. Provided for use with the Kirelta service
under its Terms of Service (https://kirelta.com/terms).

Zero dependencies. Stdlib only, matching the engine's own philosophy: a reliability product
whose client library drags in a dependency tree is undermining its own pitch, and `pip install
kirelta` should never be the reason a customer's build breaks.

    from kirelta_client import Kirelta

    kir = Kirelta(api_key="kir_...")
    kir.fit("prod-api", baseline_rows)            # picks sync or async for you
    report = kir.assess("prod-api", live_rows)

    if not report.trusted:
        print(report.explain())                   # why, in words, not just a verdict
        raise SystemExit("model input is not trustworthy right now")

Four things here are not sugar over `requests.post`, and are the reason this file exists rather
than a README telling people to call the API directly:

1. **Retries do not double-charge you.** `/assess` is metered. A naive client that retries a
   timeout bills the customer twice for one batch and writes two readings into their history.
   Every retryable call carries an `Idempotency-Key` that is generated ONCE per logical call and
   reused across attempts, so a retry is recognised server-side as the same call. This is the
   single most valuable thing in the file and the easiest one to get wrong by hand.

2. **`fit()` picks sync or async for you.** A 200k-row fit takes ~8s and a 1k-row fit takes
   ~1.6s; both are dominated by model selection, not transfer. Callers should not have to know
   where the timeout cliff is, so `fit()` submits large baselines as a job and polls, and small
   ones inline, and returns the same object either way.

3. **Errors say what to do.** `QuotaExceeded` carries the plan and the limit that was hit.
   `RateLimited` carries the retry delay. `ModelNotFitted` tells you to call `fit()` first.
   An HTTP status code is not an error message.

4. **Every error carries the server's request id.** When a customer opens a support ticket, the
   id in their traceback is the id in your logs. That is the difference between a five-minute
   investigation and an afternoon.
"""
from __future__ import annotations

import hashlib
import hmac
import json
import random
import time
import urllib.error
import urllib.parse
import urllib.request
import uuid

__version__ = "1.0.0"
__all__ = [
    "Kirelta", "Report", "Job", "Decision", "verify_webhook", "SignatureError",
    "KireltaError", "AuthError", "QuotaExceeded", "RateLimited", "ModelNotFitted",
    "ValidationError", "ServerError", "JobFailed", "NotTrusted",
]

DEFAULT_BASE_URL = "https://api.kirelta.com"

# Above this many rows, fit() goes async. Chosen from measured latency: a synchronous fit of
# ~50k rows lands near 3s and 200k near 8s, which is inside most client timeouts but outside
# most patience. Override with fit(..., async_=True/False) when you know better.
ASYNC_FIT_THRESHOLD = 25_000


# ── errors ───────────────────────────────────────────────────────────────────────────────

class KireltaError(Exception):
    """Base for every error this client raises."""

    def __init__(self, message, *, status=None, request_id=None, detail=None, body=None):
        super().__init__(message)
        self.status = status
        self.request_id = request_id      # matches X-Request-Id in Kirelta's logs
        self.detail = detail
        self.body = body or {}

    def __str__(self):
        base = super().__str__()
        return f"{base} (request_id={self.request_id})" if self.request_id else base


class AuthError(KireltaError):
    """401/403 -- the key or session is missing, wrong, or lacks permission."""


class ValidationError(KireltaError):
    """400 -- the request was malformed. The message is the server's, which is usually specific."""


class ModelNotFitted(KireltaError):
    """409 -- assess() was called on a model with no baseline. Call fit() first."""


class QuotaExceeded(KireltaError):
    """402/429-quota -- a plan limit was reached. Carries `plan` and `limit` where the server says."""

    def __init__(self, message, *, plan=None, limit=None, **kw):
        super().__init__(message, **kw)
        self.plan = plan
        self.limit = limit


class RateLimited(KireltaError):
    """429 -- too many requests per minute for the plan. `retry_after` is seconds."""

    def __init__(self, message, *, retry_after=None, **kw):
        super().__init__(message, **kw)
        self.retry_after = retry_after


class ServerError(KireltaError):
    """5xx, or a transport failure. Retried automatically before it reaches you."""


class JobFailed(KireltaError):
    """An async fit finished in the `failed` state."""


class SignatureError(KireltaError):
    """A webhook failed signature verification. Treat the delivery as hostile: drop it."""


class NotTrusted(KireltaError):
    """Raised by Report.raise_for_verdict() when the verdict is not TRUSTED."""

    def __init__(self, report):
        super().__init__(f"verdict is {report.verdict}: {report.action}")
        self.report = report


# ── result objects ───────────────────────────────────────────────────────────────────────

class Report:
    """One `/assess` result. Attribute access, plus the raw dict at `.raw` for anything new."""

    __slots__ = ("raw",)

    def __init__(self, raw):
        self.raw = raw or {}

    model = property(lambda s: s.raw.get("model"))
    verdict = property(lambda s: s.raw.get("verdict"))
    action = property(lambda s: s.raw.get("action"))
    drift_alarmed = property(lambda s: bool(s.raw.get("drift_alarmed")))
    flagged_rate = property(lambda s: s.raw.get("flagged_rate"))
    n_rows = property(lambda s: s.raw.get("n"))
    decision_id = property(lambda s: s.raw.get("decision_id"))
    evidence_id = property(lambda s: s.raw.get("evidence_id"))
    temporal = property(lambda s: s.raw.get("temporal") or {})

    @property
    def trusted(self):
        return self.verdict == "TRUSTED"

    @property
    def degraded(self):
        return self.verdict == "DEGRADED"

    def raise_for_verdict(self, allow_degraded=False):
        """Gate a pipeline in one line: `kir.assess(...).raise_for_verdict()`.

        `allow_degraded` exists because DEGRADED is a warning, not a failure -- some pipelines
        should stop on it and some should log it and continue, and that is the caller's call to
        make, not ours.
        """
        if self.trusted or (allow_degraded and self.degraded):
            return self
        raise NotTrusted(self)

    def explain(self):
        """A human-readable summary. For logs, alerts and incident tickets."""
        bits = [f"{self.model}: {self.verdict}"]
        if self.flagged_rate is not None:
            bits.append(f"{self.flagged_rate:.1%} of rows flagged")
        if self.drift_alarmed:
            bits.append("population drift alarm FIRED -- the batch as a whole no longer matches "
                        "the baseline, so a re-fit or an upstream fix is needed")
        elif not self.trusted:
            bits.append("drift alarm did not fire -- individual outliers rather than a "
                        "distribution shift, so investigate the rows, not the baseline")
        t = self.temporal
        if t.get("elevated"):
            bits.append("the rows in this batch are related to one another rather than "
                        "independent, so the drift alarm is weaker here -- the per-row flags "
                        "and the flagged rate are unaffected")
        if self.decision_id:
            bits.append(f"decision {self.decision_id}")
        return "; ".join(bits)

    def __repr__(self):
        return f"<Report {self.model} {self.verdict} flagged={self.flagged_rate}>"


class Job:
    """An async fit. `wait()` blocks until it settles."""

    __slots__ = ("raw", "_client")

    def __init__(self, raw, client=None):
        self.raw = raw or {}
        self._client = client

    id = property(lambda s: s.raw.get("id") or s.raw.get("job_id"))
    status = property(lambda s: s.raw.get("status"))
    model = property(lambda s: s.raw.get("model"))
    attempts = property(lambda s: s.raw.get("attempts"))
    worker = property(lambda s: s.raw.get("worker"))
    error = property(lambda s: s.raw.get("error"))
    duration_ms = property(lambda s: s.raw.get("duration_ms"))

    @property
    def done(self):
        return self.status in ("completed", "failed")

    def refresh(self):
        self.raw = self._client.job(self.id).raw
        return self

    def wait(self, timeout=600, poll=1.0):
        """Poll until the job settles. Raises JobFailed on failure, returns self on success."""
        deadline = time.time() + timeout
        while True:
            self.refresh()
            if self.status == "completed":
                return self
            if self.status == "failed":
                raise JobFailed(f"fit job {self.id} failed: {self.error or 'no reason given'}")
            if time.time() > deadline:
                raise KireltaError(
                    f"fit job {self.id} still {self.status} after {timeout}s -- it is not lost, "
                    f"poll it later with client.job({self.id!r})")
            time.sleep(poll)

    def __repr__(self):
        return f"<Job {self.id} {self.status} model={self.model}>"


class Decision:
    """One decision-journal entry."""

    __slots__ = ("raw",)

    def __init__(self, raw):
        self.raw = raw or {}

    id = property(lambda s: s.raw.get("id"))
    model = property(lambda s: s.raw.get("model"))
    status = property(lambda s: s.raw.get("status"))
    verdict = property(lambda s: s.raw.get("verdict"))
    created = property(lambda s: s.raw.get("created"))

    def __repr__(self):
        return f"<Decision {self.id} {self.status} {self.verdict}>"


# ── client ───────────────────────────────────────────────────────────────────────────────

def _coerce_rows(rows):
    """Accept lists, numpy arrays or pandas DataFrames without importing either.

    Duck-typed on purpose: requiring numpy in a client library, only to call .tolist() on it,
    would be a dependency bought for nothing.
    """
    if rows is None:
        raise ValidationError("rows is required")
    if hasattr(rows, "to_numpy"):            # pandas DataFrame / Series
        rows = rows.to_numpy()
    if hasattr(rows, "tolist"):              # numpy ndarray
        rows = rows.tolist()
    if not isinstance(rows, (list, tuple)):
        raise ValidationError(f"rows must be a sequence of sequences, got {type(rows).__name__}")
    if len(rows) == 0:
        raise ValidationError("rows is empty -- send at least one row")
    return list(rows)


class Kirelta:
    """Kirelta API client.

    Args:
        api_key: your `kir_...` key. Used for /fit and /assess.
        session: a session token, if you are driving account endpoints instead.
        base_url: defaults to production.
        timeout: per-request seconds.
        max_retries: attempts for retryable failures (429 and 5xx). Idempotency-Key makes
            these safe on metered endpoints.
    """

    def __init__(self, api_key=None, *, session=None, base_url=DEFAULT_BASE_URL,
                 timeout=60.0, max_retries=3, user_agent=None):
        if not api_key and not session:
            raise ValueError("one of api_key= or session= is required")
        self.api_key = api_key
        self.session = session
        self.base_url = (base_url or DEFAULT_BASE_URL).rstrip("/")
        self.timeout = float(timeout)
        self.max_retries = max(1, int(max_retries))
        self.user_agent = user_agent or f"kirelta-python/{__version__}"

    # -- transport -------------------------------------------------------------------------

    def _headers(self, idem_key=None, request_id=None):
        h = {"Content-Type": "application/json", "Accept": "application/json",
             "User-Agent": self.user_agent,
             "X-Request-Id": request_id or ("cli_" + uuid.uuid4().hex[:16])}
        if self.api_key:
            h["X-API-Key"] = self.api_key
        if self.session:
            h["Authorization"] = f"Bearer {self.session}"
        if idem_key:
            h["Idempotency-Key"] = idem_key
        return h

    def _raise(self, status, body, request_id):
        err = (body or {}).get("error") or ""
        detail = (body or {}).get("detail") or ""
        msg = detail or err or f"HTTP {status}"
        common = dict(status=status, request_id=request_id, detail=detail, body=body)
        if status in (401, 403):
            raise AuthError(msg, **common)
        if status == 402 or err == "quota_exceeded":
            raise QuotaExceeded(msg, plan=(body or {}).get("plan"),
                                limit=(body or {}).get("limit"), **common)
        if status == 429:
            ra = (body or {}).get("retry_after")
            raise RateLimited(msg, retry_after=ra, **common)
        if status == 409:
            raise ModelNotFitted(msg, **common)
        if 400 <= status < 500:
            raise ValidationError(msg, **common)
        raise ServerError(msg, **common)

    def _request(self, method, path, *, params=None, payload=None, idempotent=False,
                 timeout=None):
        url = self.base_url + path
        if params:
            url += "?" + urllib.parse.urlencode({k: v for k, v in params.items() if v is not None})
        data = json.dumps(payload).encode() if payload is not None else None

        # Generated ONCE, outside the retry loop, and reused by every attempt. That is what
        # makes a retry of a metered call safe: the server recognises attempt 2 as the same
        # logical call and neither bills it again nor writes a second reading.
        idem_key = ("sdk_" + uuid.uuid4().hex) if idempotent else None
        request_id = "cli_" + uuid.uuid4().hex[:16]

        last = None
        for attempt in range(self.max_retries):
            req = urllib.request.Request(
                url, data=data, method=method,
                headers=self._headers(idem_key=idem_key, request_id=request_id))
            try:
                with urllib.request.urlopen(req, timeout=timeout or self.timeout) as r:
                    raw = r.read()
                    return json.loads(raw) if raw else {}
            except urllib.error.HTTPError as e:
                rid = e.headers.get("X-Request-Id") if e.headers else request_id
                try:
                    body = json.loads(e.read() or b"{}")
                except Exception:
                    body = {}
                if e.code == 429 or e.code >= 500:
                    last = (e.code, body, rid)
                    if attempt < self.max_retries - 1:
                        # Honour Retry-After when the server sends one; otherwise exponential
                        # backoff with jitter so a fleet of clients does not retry in lockstep.
                        wait = e.headers.get("Retry-After") if e.headers else None
                        try:
                            delay = float(wait)
                        except (TypeError, ValueError):
                            delay = (2 ** attempt) + random.uniform(0, 0.5)
                        time.sleep(min(delay, 30.0))
                        continue
                self._raise(e.code, body, rid)
            except (urllib.error.URLError, TimeoutError, OSError) as e:
                last = (None, {"detail": f"{type(e).__name__}: {e}"}, request_id)
                if attempt < self.max_retries - 1:
                    time.sleep((2 ** attempt) + random.uniform(0, 0.5))
                    continue
                raise ServerError(f"could not reach Kirelta at {self.base_url}: {e}",
                                  request_id=request_id) from e
        status, body, rid = last or (None, {}, request_id)
        self._raise(status or 500, body, rid)

    # -- core ------------------------------------------------------------------------------

    def health(self):
        """Liveness plus the honest `durable` flag (false means models are not on a volume)."""
        return self._request("GET", "/health")

    def fit(self, model, rows, *, feature_names=None, async_=None, wait=True, timeout=600):
        """Learn a baseline.

        Chooses sync or async by size unless you force it with `async_`. With `wait=True` an
        async fit is polled to completion, so the call is a drop-in for the sync one.
        """
        rows = _coerce_rows(rows)
        use_async = async_ if async_ is not None else (len(rows) > ASYNC_FIT_THRESHOLD)
        payload = {"rows": rows}
        if feature_names is not None:
            payload["feature_names"] = list(feature_names)
        params = {"model": model}
        if use_async:
            params["async"] = 1
        out = self._request("POST", "/fit", params=params, payload=payload, idempotent=True,
                            timeout=timeout)
        if use_async:
            job = Job({"id": out.get("job_id"), "status": out.get("status", "queued"),
                       "model": model}, client=self)
            return job.wait(timeout=timeout) if wait else job
        return out

    def fit_async(self, model, rows, **kw):
        """Submit a fit and return immediately with a Job you can poll."""
        kw.setdefault("wait", False)
        return self.fit(model, rows, async_=True, **kw)

    def assess(self, model, rows, *, idempotency_key=None):
        """Score a batch against the model's baseline. Returns a Report.

        This endpoint is metered, which is exactly why the retry path carries an
        Idempotency-Key: a retried timeout must not bill the batch twice.
        """
        rows = _coerce_rows(rows)
        out = self._request("POST", "/assess", params={"model": model},
                            payload={"rows": rows}, idempotent=True)
        return Report(out)

    def check(self, model, rows, *, allow_degraded=False):
        """assess() + raise_for_verdict() -- the one-line pipeline gate."""
        return self.assess(model, rows).raise_for_verdict(allow_degraded=allow_degraded)

    # -- models, usage ---------------------------------------------------------------------

    def models(self):
        return self._request("GET", "/models")

    def usage(self):
        return self._request("GET", "/usage")

    # -- jobs ------------------------------------------------------------------------------

    def job(self, job_id):
        out = self._request("GET", f"/jobs/{urllib.parse.quote(str(job_id))}")
        return Job(out.get("job") or out, client=self)

    def jobs(self, *, status=None, limit=50):
        """Job history and queue state. `status` in queued|running|completed|failed."""
        out = self._request("GET", "/jobs", params={"status": status, "limit": limit})
        return {"jobs": [Job(j, client=self) for j in out.get("jobs", [])],
                "summary": out.get("summary", {})}

    # -- decisions -------------------------------------------------------------------------

    def decisions(self, *, status=None, model=None, limit=50, offset=0):
        out = self._request("GET", "/decisions", params={
            "status": status, "model": model, "limit": limit, "offset": offset})
        return {"decisions": [Decision(d) for d in out.get("decisions", [])],
                "total": out.get("total", 0)}

    def iter_decisions(self, *, status=None, model=None, page_size=100):
        """Walk every decision without hand-rolling pagination."""
        offset = 0
        while True:
            page = self.decisions(status=status, model=model, limit=page_size, offset=offset)
            if not page["decisions"]:
                return
            for d in page["decisions"]:
                yield d
            offset += len(page["decisions"])
            if offset >= page.get("total", 0):
                return

    def decision(self, decision_id):
        out = self._request("GET", f"/decisions/{urllib.parse.quote(str(decision_id))}")
        return Decision(out.get("decision") or out)

    def transition(self, decision_id, to, note=""):
        """Move a decision through its lifecycle: reviewing/approved/rejected/executed/archived."""
        out = self._request("POST", f"/decisions/{urllib.parse.quote(str(decision_id))}/transition",
                            payload={"to": to, "note": note})
        return Decision(out.get("decision") or out)

    def evidence(self, decision_id):
        """The sealed evidence snapshot, including its `integrity_ok` hash check."""
        out = self._request("GET",
                            f"/decisions/{urllib.parse.quote(str(decision_id))}/evidence")
        return out.get("evidence") or out

    # -- integration -----------------------------------------------------------------------

    def event_registry(self):
        """The published webhook contract: kinds, severities, guaranteed fields, version."""
        return self._request("GET", "/events/registry")

    def __repr__(self):
        who = "api_key" if self.api_key else "session"
        return f"<Kirelta {self.base_url} auth={who}>"


# ── webhook verification ─────────────────────────────────────────────────────────────────

WEBHOOK_TOLERANCE_SECONDS = 300


def verify_webhook(raw_body, signature_header, secret, *, tolerance=WEBHOOK_TOLERANCE_SECONDS):
    """Verify an inbound Kirelta webhook and return the parsed event.

    Raises SignatureError on ANY failure. There is no "probably fine" branch: an endpoint that
    logs a warning and processes the event anyway is an unverified endpoint with extra steps.

        from kirelta_client import verify_webhook, SignatureError

        @app.post("/kirelta-hook")
        def hook(request):
            try:
                event = verify_webhook(request.body,                    # RAW bytes
                                       request.headers["X-Kirelta-Signature"],
                                       os.environ["KIRELTA_WEBHOOK_SECRET"])
            except SignatureError:
                return 400
            ...

    `raw_body` MUST be the bytes as received. Passing `json.dumps(request.json)` will fail:
    re-serialising changes whitespace and key order, and the MAC covers bytes, not meaning.
    This is the single most common way people get webhook verification wrong, and it fails
    closed here rather than silently -- which is the correct direction to fail.
    """
    if not secret:
        raise SignatureError("no signing secret supplied")
    if isinstance(raw_body, str):
        raw_body = raw_body.encode()
    if not isinstance(raw_body, (bytes, bytearray)):
        raise SignatureError("raw_body must be the bytes exactly as received")
    if not signature_header:
        raise SignatureError("missing X-Kirelta-Signature header")

    parts = {}
    for chunk in str(signature_header).split(","):
        k, _, v = chunk.strip().partition("=")
        if k:
            parts[k.strip()] = v.strip()
    ts, sig = parts.get("t"), parts.get("v1")
    if not ts or not sig:
        raise SignatureError(f"malformed signature header: {signature_header!r}")
    try:
        ts_i = int(ts)
    except ValueError:
        raise SignatureError("signature timestamp is not an integer") from None

    # Freshness first: a stale-but-valid signature is a replay, and cheap to reject.
    skew = abs(time.time() - ts_i)
    if tolerance and skew > tolerance:
        raise SignatureError(
            f"signature timestamp is {skew:.0f}s away (tolerance {tolerance}s) -- replay, or "
            f"your server clock is wrong")

    expected = hmac.new(str(secret).encode(), f"{ts_i}.".encode() + bytes(raw_body),
                        hashlib.sha256).hexdigest()
    if not hmac.compare_digest(expected, sig):
        raise SignatureError("signature does not match -- treat this delivery as forged")

    try:
        return json.loads(bytes(raw_body).decode("utf-8"))
    except (UnicodeDecodeError, json.JSONDecodeError) as e:
        raise SignatureError(f"body verified but is not valid JSON: {e}") from None