"""Fryri API client for Python. Single file, one dependency (requests).

Drop this file into your project (pip package coming). Docs: https://fryri.com/docs/api

Quickstart (create a key in Settings, API keys, and fund your wallet first):

    from fryri import Fryri

    client = Fryri("fryri_sk_...")
    r = client.chat("When is my car rego due?")
    print(r["reply"])
    for s in r["sources"]:
        print("-", s["kind"], s.get("title"))

    # Stream a chat turn as it generates:
    for event in client.chat("Summarize my week.", stream=True):
        if event["type"] == "delta":
            print(event["content"], end="", flush=True)
        elif event["type"] == "done":
            print("\\n\\ntokens:", event["tokens_used"])

    # Write to your library:
    client.capture(text="Car rego is due 30 September.")

Errors raise FryriError with .status, .code, .message and .request_id (quote
the request_id if you report a problem).
"""
from __future__ import annotations

import json
from typing import Any, Dict, Generator, List, Optional

import requests

__version__ = "0.2.0"

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


class FryriError(Exception):
    """An error response from the API ({error: {code, message}, request_id})."""

    def __init__(self, status: int, code: str, message: str, request_id: Optional[str] = None,
                 details: Optional[dict] = None):
        super().__init__(f"[{status}] {code}: {message}")
        self.status = status
        self.code = code
        self.message = message
        self.request_id = request_id
        self.details = details or {}


class Fryri:
    """Client for the Fryri /v1 API. One instance per API key."""

    def __init__(self, api_key: str, base_url: str = DEFAULT_BASE_URL, timeout: float = 120.0):
        if not api_key or not api_key.startswith("fryri_sk_"):
            raise ValueError("api_key must be a fryri_sk_... key")
        self._base = base_url.rstrip("/")
        self._timeout = timeout
        self._session = requests.Session()
        self._session.headers.update({
            "Authorization": f"Bearer {api_key}",
            "User-Agent": f"fryri-python/{__version__}",
        })

    # -- plumbing -----------------------------------------------------------

    def _request(self, method: str, path: str, *, params: Optional[dict] = None,
                 body: Optional[dict] = None, headers: Optional[dict] = None,
                 stream: bool = False) -> requests.Response:
        resp = self._session.request(
            method, f"{self._base}{path}", params=params, json=body,
            headers=headers, timeout=self._timeout, stream=stream,
        )
        if resp.status_code >= 400:
            try:
                payload = resp.json()
            except ValueError:
                raise FryriError(resp.status_code, "error", resp.text[:500])
            err = payload.get("error") or {}
            raise FryriError(
                resp.status_code,
                err.get("code", "error"),
                err.get("message", resp.text[:500]),
                payload.get("request_id"),
                {k: v for k, v in err.items() if k not in ("code", "message")},
            )
        return resp

    def _get(self, path: str, **params) -> Any:
        clean = {k: v for k, v in params.items() if v is not None}
        return self._request("GET", path, params=clean).json()

    def _post(self, path: str, body: Optional[dict] = None, headers: Optional[dict] = None) -> Any:
        return self._request("POST", path, body=body, headers=headers).json()

    # -- chat (the managed run) ----------------------------------------------

    def chat(self, question: str, *, model_choice: Optional[str] = None, web_search: bool = False,
             store: str = "library", include_trace: bool = False, stream: bool = False,
             provider: Optional[str] = None, model: Optional[str] = None,
             api_key: Optional[str] = None, max_output_tokens: Optional[int] = None):
        """A full chat turn grounded in your library:
        {reply, artifacts, sources, tokens_used, run_id} (+ facts_saved /
        search / trace / wallet_warning when they apply).

        `sources` is one list of grounding cards, each with a "kind":
        library items are file/photo/video (a file card's file_id chains
        into document()/grep()); web results are kind "web" with a url.

        With stream=True, returns a generator of event dicts instead:
        {"type": "delta", "content": ...} as text generates,
        {"type": "reset"} (discard text so far),
        {"type": "tool_call", "phase": "start"|"done", "tool": ..., "label": ...,
         "ok": ..., "count": ...} (live tool activity, names/counts only),
        then a terminal {"type": "done", ...} carrying the exact
        non-streaming shape (its `reply` is authoritative, swap your
        concatenated deltas for it), or {"type": "error", ...} on failure.

        Bring your own key (paid plans): pass provider ("openrouter" |
        "anthropic" | "openai" | "xai" | "google" | "deepseek" | "zai" |
        "xiaomi" | "minimax") + model (the raw slug on that provider) to run
        the turn on your own account. api_key is request-scoped (or omit it
        to use the key saved via set_byo_key). max_output_tokens defaults
        to 8192.
        """
        body = {
            "question": question, "model_choice": model_choice, "web_search": web_search,
            "store": store, "include_trace": include_trace, "stream": stream,
            "provider": provider, "model": model, "api_key": api_key,
            "max_output_tokens": max_output_tokens,
        }
        if not stream:
            return self._post("/v1/chat", body)
        return self._stream_chat(body)

    def _stream_chat(self, body: dict) -> Generator[Dict[str, Any], None, None]:
        resp = self._request("POST", "/v1/chat", body=body, stream=True)
        for raw in resp.iter_lines(decode_unicode=True):
            if not raw or not raw.startswith("data:"):
                continue  # keep-alive comments and blank lines
            try:
                event = json.loads(raw[len("data:"):].strip())
            except ValueError:
                continue
            yield event
            if event.get("type") in ("done", "error"):
                return

    # -- bring your own key ---------------------------------------------------

    def byo_key_status(self, provider: str = "openrouter") -> Dict[str, Any]:
        """Whether a key is saved for a provider: {enabled, has_key, key_prefix}."""
        return self._get("/v1/byo-key", provider=provider)

    def set_byo_key(self, key: str, provider: str = "openrouter") -> Dict[str, Any]:
        """Save a provider key (validated live, encrypted at rest, never returned)."""
        return self._post("/v1/byo-key", {"key": key, "provider": provider})

    def clear_byo_key(self, provider: str = "openrouter") -> Dict[str, Any]:
        """Remove a saved provider key; turns fall back to plan billing."""
        return self._request("DELETE", "/v1/byo-key", params={"provider": provider}).json()

    # -- capture (write into the library) -----------------------------------

    def capture(self, *, text: Optional[str] = None, content_base64: Optional[str] = None,
                filename: Optional[str] = None, content_type: Optional[str] = None,
                idempotency_key: Optional[str] = None) -> Dict[str, Any]:
        """Add one thing to your library: plain text (max 200k chars) OR
        base64 bytes + filename + content_type (files up to ~30MB). Returns
        {ok, detail, file_id}; file_id chains into document()/grep() (None
        for photos)."""
        headers = {"Idempotency-Key": idempotency_key} if idempotency_key else None
        return self._post("/v1/capture", {
            "text": text, "content_base64": content_base64,
            "filename": filename, "content_type": content_type,
        }, headers=headers)

    def capture_batch(self, items: List[dict], *, idempotency_key: Optional[str] = None) -> Dict[str, Any]:
        """Capture up to 100 items in one call (each item the same shape as a
        capture body). Returns {total, succeeded, results: [{index, ok,
        detail, file_id}]}; one bad item never aborts the rest."""
        headers = {"Idempotency-Key": idempotency_key} if idempotency_key else None
        return self._post("/v1/capture/batch", {"items": items}, headers=headers)

    # -- library reads ---------------------------------------------------------

    def search(self, query: str, *, limit: int = 10) -> Dict[str, Any]:
        """Semantic (meaning-based) search: {query, count, results}. limit
        caps at 50. The one metered read (spends a query embedding)."""
        return self._get("/v1/search", q=query, limit=limit)

    def grep(self, pattern: str, *, file_id: Optional[int] = None,
             case_insensitive: Optional[bool] = None, context_lines: Optional[int] = None,
             max_matches: Optional[int] = None, max_files: Optional[int] = None,
             skip: Optional[int] = None, extension: Optional[str] = None) -> Dict[str, Any]:
        """Exact-text (regex) search, two scopes. With file_id: search that
        one document (context_lines <= 5, max_matches <= 250; response has
        mode "file"). Without: search a page of the whole library, newest
        first (max_files <= 20 per call, page with skip, filter with
        extension; response has mode "library" and lists only files WITH
        matches; scanned_files says how many were checked). Free."""
        return self._get("/v1/grep", pattern=pattern, file_id=file_id,
                         case_insensitive=case_insensitive, context_lines=context_lines,
                         max_matches=max_matches, max_files=max_files,
                         skip=skip, extension=extension)

    def documents(self, *, extension: Optional[str] = None, since: Optional[str] = None,
                  skip: int = 0, limit: int = 25) -> Dict[str, Any]:
        """List your documents, newest first: {count, documents: [{file_id,
        name, media_type, size, extension, uploaded_at, preview_snippet}]}.
        `since` is an ISO-8601 timestamp; limit caps at 100. Free."""
        return self._get("/v1/documents", extension=extension, since=since,
                         skip=skip, limit=limit)

    def document(self, file_id: int, *, view: str = "structured",
                 offset: Optional[int] = None, limit: Optional[int] = None,
                 line_numbers: Optional[bool] = None) -> Dict[str, Any]:
        """Read one document. view="structured" (default) is the extraction
        the system already computed (document_type, summary_one_line,
        primary_date, structured fields, entities). view="text" is the
        extracted text, line-windowed with offset (1-indexed) / limit
        (default 200, max 2000 lines); raw text by default, pass
        line_numbers=True to match grep()'s line references. Free."""
        return self._get(f"/v1/documents/{file_id}", view=view, offset=offset,
                         limit=limit, line_numbers=line_numbers)

    def document_download(self, file_id: int, *, download: bool = False) -> Dict[str, Any]:
        """A 300-second presigned URL to the ORIGINAL uploaded bytes:
        {file_id, url, filename, expires_in}. download=True forces a
        save-to-disk Content-Disposition."""
        return self._get(f"/v1/documents/{file_id}/download", download=download or None)

    # -- facts ---------------------------------------------------------------

    def facts(self, *, include_superseded: bool = False, skip: int = 0, limit: int = 200) -> Dict[str, Any]:
        return self._get("/v1/facts", include_superseded=include_superseded, skip=skip, limit=limit)

    def create_fact(self, *, predicate: str, value: str, category: str,
                    subject: Optional[str] = None, label: Optional[str] = None,
                    valid_from: Optional[str] = None) -> Dict[str, Any]:
        return self._post("/v1/facts", {
            "predicate": predicate, "value": value, "category": category,
            "subject": subject, "label": label, "valid_from": valid_from,
        })

    def update_fact(self, fact_id: int, **fields) -> Dict[str, Any]:
        return self._request("PATCH", f"/v1/facts/{fact_id}", body=fields).json()

    def delete_fact(self, fact_id: int) -> Dict[str, Any]:
        return self._request("DELETE", f"/v1/facts/{fact_id}").json()

    # -- research ------------------------------------------------------------

    def research(self, instructions: str, columns: List[Any],
                 entity: Optional[str] = None, *,
                 idempotency_key: Optional[str] = None) -> Dict[str, Any]:
        """Kick off an async deep-research job that fills a comparison table
        (premium plans, daily-capped). `columns` are the table columns, as
        strings (["model", "price"]) or dicts ({"key": "price", "name":
        "Price"}). Returns {job_id, status, poll_url}; poll research_result()
        until succeeded/failed."""
        cols = [{"key": c} if isinstance(c, str) else c for c in columns]
        body: Dict[str, Any] = {"instructions": instructions, "columns": cols}
        if entity:
            body["entity"] = entity
        headers = {"Idempotency-Key": idempotency_key} if idempotency_key else None
        return self._post("/v1/research", body, headers=headers)

    def research_result(self, job_id: int) -> Dict[str, Any]:
        """Poll a research job. `table` is None until status is succeeded."""
        return self._get(f"/v1/research/{job_id}")

    # -- entity research (company/entity intelligence) ------------------------

    def entity_research(self, query: str, *, effort: str = "medium",
                        output_schema: Optional[Dict[str, Any]] = None,
                        input_data: Optional[List[Dict[str, Any]]] = None,
                        input_exclusion: Optional[List[Dict[str, Any]]] = None,
                        data_sources: Optional[List[str]] = None,
                        system_prompt: Optional[str] = None,
                        idempotency_key: Optional[str] = None) -> Dict[str, Any]:
        """Start an entity/company research run: list building, row
        enrichment, KYB profiles, multi-step research with citations.

        `effort` sets the per-run price, billed at cost: minimal $0.012,
        low $0.025, medium $0.10, high $0.50, xhigh $1.00; "auto" scales
        with the task. `data_sources` plugs premium providers into the run
        (e.g. ["similarweb"], max 5), per-call, at cost. Returns {job_id,
        status, poll_url}; poll entity_research_result() until status is
        completed, failed, or cancelled. Paid plans only."""
        body: Dict[str, Any] = {"query": query, "effort": effort}
        if output_schema:
            body["output_schema"] = output_schema
        if input_data:
            body["input_data"] = input_data
        if input_exclusion:
            body["input_exclusion"] = input_exclusion
        if data_sources:
            body["data_sources"] = data_sources
        if system_prompt:
            body["system_prompt"] = system_prompt
        headers = {"Idempotency-Key": idempotency_key} if idempotency_key else None
        return self._post("/v1/research/entity", body, headers=headers)

    def entity_research_result(self, job_id: int) -> Dict[str, Any]:
        """Poll a run. `output` is None until terminal, then carries
        {text, structured, grounding}; `cost_usd` is the run's real cost."""
        return self._get(f"/v1/research/entity/{job_id}")

    # -- webhooks (proactive out) ---------------------------------------------

    def webhooks(self) -> Any:
        return self._get("/v1/webhooks")

    def create_webhook(self, url: str, events: Optional[List[str]] = None) -> Dict[str, Any]:
        """Register a URL. The signing secret is in the response ONCE.
        `events` is a subset of: research.completed, research.failed,
        entity_research.completed, entity_research.failed, capture.completed,
        capture.failed, radar.brief, or omit it (["*"]) for everything."""
        return self._post("/v1/webhooks", {"url": url, "events": events or ["*"]})

    def delete_webhook(self, webhook_id: int) -> None:
        self._request("DELETE", f"/v1/webhooks/{webhook_id}")

    def test_webhook(self, webhook_id: int) -> Dict[str, Any]:
        return self._post(f"/v1/webhooks/{webhook_id}/test")

    # -- billing & account -----------------------------------------------------

    def usage(self, *, days: int = 30) -> Dict[str, Any]:
        """Itemized spend for this key and the whole account: {key, account}."""
        return self._get("/v1/usage", days=days)

    def run_usage(self, run_id: str) -> Dict[str, Any]:
        """Cost/token breakdown for ONE run. run_id is the chat response's
        `run_id` (also the X-Request-ID response header on every call)."""
        return self._get(f"/v1/runs/{run_id}/usage")

    def wallet(self) -> Dict[str, Any]:
        """Funding state: {tier, wallet_balance_usd, wallet_funded,
        over_budget, topup_endpoint, ...}. over_budget=True means the next
        paid call would 402."""
        return self._get("/v1/wallet")

    def topup_link(self, amount_usd: int) -> Dict[str, Any]:
        """A Stripe-hosted checkout URL a human opens to fund the wallet.
        amount_usd must be one of 5, 10, 20, 50, 100."""
        return self._post("/v1/wallet/topup-link", {"amount_usd": amount_usd})

    def topup_token(self, spt: str, amount_usd: int) -> Dict[str, Any]:
        """Headless funding: charge a Stripe Shared Payment Token (spt_...)
        your owner granted. Idempotent; amount_usd one of 5, 10, 20, 50, 100."""
        return self._post("/v1/wallet/topup-token", {"spt": spt, "amount_usd": amount_usd})

    # -- beta: /v1/evolving (1-3 month deprecation window; may change) ---------

    def query(self, operation: str, source: str = "all", **params) -> Dict[str, Any]:
        """BETA. Structural counts/lists over the library: operation is
        count|list|oldest|newest; source is photos|files|chats|conversations|
        thoughts|events|all."""
        return self._get("/v1/evolving/query", operation=operation, source=source, **params)

    def trace(self, name: str, *, hops: Optional[int] = None) -> Dict[str, Any]:
        """BETA. Entity connection graph from a person/place/org:
        {nodes, edges, documents}. hops is 1-3 (default 2)."""
        return self._get("/v1/evolving/trace", name=name, hops=hops)

    def skills(self, *, cursor: Optional[str] = None, limit: int = 20) -> Dict[str, Any]:
        """BETA. List your skills, newest updated first (cursor paging)."""
        return self._get("/v1/evolving/skills", cursor=cursor, limit=limit)

    def skill(self, skill_id: int) -> Dict[str, Any]:
        """BETA. One skill, including the full recipe body."""
        return self._get(f"/v1/evolving/skills/{skill_id}")

    def create_skill(self, *, description: Optional[str] = None, **draft) -> Dict[str, Any]:
        """BETA. Create from a plain-English description (the server authors
        the recipe) or pass a full draft (title=, body=, parameters=, ...)."""
        body = dict(draft)
        if description is not None:
            body["description"] = description
        return self._post("/v1/evolving/skills", body)

    def update_skill(self, skill_id: int, **fields) -> Dict[str, Any]:
        return self._request("PATCH", f"/v1/evolving/skills/{skill_id}", body=fields).json()

    def delete_skill(self, skill_id: int) -> Dict[str, Any]:
        return self._request("DELETE", f"/v1/evolving/skills/{skill_id}").json()

    def run_skill(self, skill_id: int, *, params: Optional[dict] = None,
                  idempotency_key: Optional[str] = None) -> Dict[str, Any]:
        """BETA. Start a run (202). Poll skill_run() until succeeded/failed.
        Idempotency-Key dedupes at the run level: a replay returns the same
        run."""
        headers = {"Idempotency-Key": idempotency_key} if idempotency_key else None
        return self._post(f"/v1/evolving/skills/{skill_id}/run", {"params": params}, headers=headers)

    def skill_run(self, run_id: int) -> Dict[str, Any]:
        """BETA. Poll one run. A finished run's response includes `output`,
        the run's final answer text."""
        return self._get(f"/v1/evolving/skills/runs/{run_id}")

    def skill_runs(self, skill_id: int, *, limit: int = 20) -> Dict[str, Any]:
        """BETA. Recent runs of one skill, newest first."""
        return self._get(f"/v1/evolving/skills/{skill_id}/runs", limit=limit)

    def agent_runs(self, *, status: Optional[str] = None, before_id: Optional[int] = None,
                   limit: int = 50) -> Dict[str, Any]:
        """BETA. The account's background agent runs, newest first (read-only
        supervision of runs started in the app). Page by passing the returned
        next_before_id back as before_id."""
        return self._get("/v1/evolving/agent-runs", status=status, before_id=before_id, limit=limit)

    def agent_run(self, run_id: int) -> Dict[str, Any]:
        """BETA. Poll one agent run."""
        return self._get(f"/v1/evolving/agent-runs/{run_id}")

    def agent_run_events(self, run_id: int, *, after_seq: int = 0, limit: int = 200) -> Dict[str, Any]:
        """BETA. The run's event trail (tool names/labels/counts only, never
        arguments or outputs). You're caught up when your highest event seq
        equals the run's seq."""
        return self._get(f"/v1/evolving/agent-runs/{run_id}/events", after_seq=after_seq, limit=limit)

    def cancel_agent_run(self, run_id: int) -> Dict[str, Any]:
        """BETA. Stop a run, keeping the work it already did. 409 if it
        already finished."""
        return self._post(f"/v1/evolving/agent-runs/{run_id}/cancel")