"""FinData OS Python SDK.

    from findata_sdk import FinData
    fd = FinData("https://findata.spboucher.ai", api_key="fd_xxx.yyy")
    bars = fd.bars("AAPL", "1m", "2026-10-26", "2026-10-30")          # pyarrow.Table (all pages, Arrow IPC)
    df = bars.to_pandas()                                              # or fd.bars(...).to_polars()
    export = fd.export(["AAPL", "MSFT"], "2024-01-01", "2026-10-30")   # async Parquet export → local file
    facts = fd.fundamentals("AAPL", as_of="2025-01-01T00:00:00Z", metrics=["rev", "net_inc"])
"""
from __future__ import annotations

import io
import os
import time
from datetime import date, datetime
from pathlib import Path
from typing import Any

import httpx
import pyarrow as pa
from pyarrow import ipc


class FinDataError(RuntimeError):
    def __init__(self, status: int, payload: Any):
        self.status = status
        self.payload = payload
        err = payload.get("error", {}) if isinstance(payload, dict) else {}
        super().__init__(f"HTTP {status} {err.get('code', '')}: {err.get('message', payload)} (request_id={err.get('request_id')})")


class FinData:
    def __init__(self, base_url: str | None = None, api_key: str | None = None, timeout: float = 120.0, transport: httpx.BaseTransport | None = None):
        self.base_url = (base_url or os.environ.get("FINDATA_URL", "https://findata.spboucher.ai")).rstrip("/")
        key = api_key or os.environ.get("FINDATA_API_KEY")
        if not key:
            raise ValueError("api_key required (or FINDATA_API_KEY env var)")
        self._c = httpx.Client(base_url=self.base_url, timeout=timeout, headers={"Authorization": f"Bearer {key}", "User-Agent": "findata-sdk/0.1"}, transport=transport)

    # ------------------------------------------------------------------ low level
    def _get(self, path: str, **params: Any) -> Any:
        r = self._c.get(f"/api/v1{path}", params={k: v for k, v in params.items() if v is not None})
        if r.status_code >= 400:
            raise FinDataError(r.status_code, _json(r))
        return r.json()

    def _post(self, path: str, body: dict[str, Any]) -> Any:
        r = self._c.post(f"/api/v1{path}", json=body)
        if r.status_code >= 400:
            raise FinDataError(r.status_code, _json(r))
        return r.json()

    # ------------------------------------------------------------------ catalog
    def datasets(self) -> list[dict[str, Any]]:
        return self._get("/catalog/datasets")["data"]

    def instruments(self, query: str = "", **kw: Any) -> list[dict[str, Any]]:
        return self._get("/instruments", query=query, **kw)["data"]

    def health(self) -> dict[str, Any]:
        return self._get("/health")

    # ------------------------------------------------------------------ bars
    def bars(self, symbol: str, interval: str, start: str | date, end: str | date, adjustment: str = "raw",
             max_pages: int = 200) -> pa.Table:
        """All pages of a bounded request as one Arrow table (Arrow IPC transport, cursor pagination)."""
        tables: list[pa.Table] = []
        cursor: str | None = None
        for _ in range(max_pages):
            params = {"symbol": symbol, "interval": interval, "start": str(start), "end": str(end), "adjustment": adjustment,
                      "format": "arrow", "limit": 10000}
            if cursor:
                params["cursor"] = cursor
            r = self._c.get("/api/v1/market/bars", params=params)
            if r.status_code >= 400:
                raise FinDataError(r.status_code, _json(r))
            with ipc.open_stream(io.BytesIO(r.content)) as reader:
                tables.append(reader.read_all())
            cursor = r.headers.get("X-Next-Cursor") or None
            if not cursor:
                break
        return pa.concat_tables(tables) if tables else pa.table({})

    def sessions(self, symbol: str, interval: str = "1m", **kw: Any) -> list[dict[str, Any]]:
        return self._get("/market/sessions", symbol=symbol, interval=interval, **kw)["data"]

    # ------------------------------------------------------------------ fundamentals
    def fundamentals(self, entity: str, as_of: str | datetime | None = None, metrics: list[str] | None = None,
                     period_type: str | None = None) -> dict[str, Any]:
        return self._get(f"/fundamentals/{entity}", as_of=str(as_of) if as_of else None, metrics=",".join(metrics) if metrics else None,
                         period_type=period_type)

    # ------------------------------------------------------------------ query
    def sql(self, sql: str, max_rows: int = 1000) -> pa.Table:
        res = self._post("/query", {"sql": sql, "max_rows": max_rows})
        if not res["rows"]:
            return pa.table({c["name"]: [] for c in res["columns"]})
        return pa.Table.from_pylist(res["rows"])

    # ------------------------------------------------------------------ exports
    def export(self, symbols: list[str], start: str | date, end: str | date, dataset_id: str = "market_bars.stock.1m.raw",
               adjustment: str = "raw", fmt: str = "parquet", dest: str | Path | None = None, poll_s: float = 2.0, timeout_s: float = 3600) -> Path:
        job = self._post("/exports", {"dataset_id": dataset_id, "symbols": symbols, "start": str(start), "end": str(end), "adjustment": adjustment, "format": fmt})
        eid = job["id"]
        t0 = time.monotonic()
        while True:
            st = self._get(f"/exports/{eid}")
            if st["status"] == "ready":
                break
            if st["status"] in ("failed", "expired"):
                raise FinDataError(500, {"error": {"code": st["status"], "message": st.get("error")}})
            if time.monotonic() - t0 > timeout_s:
                raise TimeoutError(f"export {eid} not ready after {timeout_s}s")
            time.sleep(poll_s)
        out = Path(dest) if dest else Path(f"findata-{eid}.{fmt}")
        with self._c.stream("GET", f"/api/v1/exports/{eid}/download") as r:
            if r.status_code >= 400:
                raise FinDataError(r.status_code, {"error": {"code": "download_failed"}})
            with open(out, "wb") as f:
                f.writelines(r.iter_bytes())
        return out


def _json(r: httpx.Response) -> Any:
    try:
        return r.json()
    except Exception:
        return {"error": {"code": "http", "message": r.text[:300]}}