"""Typed, redacting client for the FinData OS API (https://findata.spboucher.ai/api/v1).

Design rules (docs/API_INTEGRATION.md):
- the bearer token is read from the environment or a mode-600 file, never logged, never placed in a URL;
- bounded exponential backoff with jitter on 429/5xx and transport errors (idempotent GETs; export creation only when
  no export id was obtained);
- every call appends one provenance line (method, path, status, request id, dataset id/version, elapsed) to a JSONL log;
- Parquet exports are validated against the published bars schema before being accepted.
"""
from __future__ import annotations

import hashlib
import json
import os
import random
import re
import time
from dataclasses import dataclass, asdict
from pathlib import Path
from typing import Any, Iterator

import httpx
import threading

_TOKEN_RE = re.compile(r"fd_[A-Za-z0-9_.\-]+")


def _redact(text: str) -> str:
    return _TOKEN_RE.sub("fd_<REDACTED>", text)


class FinDataError(RuntimeError):
    def __init__(self, status: int, code: str, message: str, request_id: str | None = None):
        super().__init__(f"HTTP {status} {code}: {_redact(message)} (request_id={request_id})")
        self.status, self.code, self.request_id = status, code, request_id


def load_api_key() -> str:
    key = os.environ.get("FINDATA_API_KEY")
    if key:
        return key.strip()
    path = os.path.expanduser(os.environ.get("FINDATA_API_KEY_FILE", "~/findata-secrets/api-key-wp39.txt"))
    if not os.path.exists(path):
        path = os.path.expanduser("~/Desktop/Cluster/secrets/findata/api-key-wp39.txt")
    with open(path) as fh:
        return fh.read().strip()


@dataclass
class ExportResult:
    export_id: str
    dataset_id: str
    dataset_version: int | None
    symbols: list[str]
    start: str
    end: str
    adjustment: str
    row_count: int
    bytes: int
    sha256: str
    path: str
    request_id: str | None
    exported_at: str


class _RateLimiter:
    """Process-wide token bucket shared by every client/thread (default 6 req/s; override with WP39_RATE)."""
    _lock = threading.Lock()
    _last = 0.0

    @classmethod
    def wait(cls) -> None:
        rate = float(os.environ.get("WP39_RATE", "6"))
        if rate <= 0:
            return
        with cls._lock:
            gap = 1.0 / rate - (time.monotonic() - cls._last)
            if gap > 0:
                time.sleep(gap)
            cls._last = time.monotonic()


class FinDataClient:
    def __init__(
        self,
        base_url: str | None = None,
        api_key: str | None = None,
        log_path: str | os.PathLike | None = None,
        timeout: float = 60.0,
        max_attempts: int = 8,
    ):
        self.base_url = (base_url or os.environ.get("FINDATA_BASE_URL") or "https://findata.spboucher.ai/api/v1").rstrip("/")
        self._key = api_key or load_api_key()
        self.max_attempts = max_attempts
        self.log_path = Path(log_path) if log_path else None
        self._client = httpx.Client(timeout=timeout, headers={"Authorization": f"Bearer {self._key}", "User-Agent": "wp39-uqo/0.1"})

    # ------------------------------------------------------------------ core
    def _log(self, rec: dict[str, Any]) -> None:
        if not self.log_path:
            return
        self.log_path.parent.mkdir(parents=True, exist_ok=True)
        with open(self.log_path, "a") as fh:
            fh.write(json.dumps(rec, default=str) + "\n")

    def _request(self, method: str, path: str, *, params=None, json_body=None, stream_to: Path | None = None,
                 retry: bool = True) -> httpx.Response:
        url = path if path.startswith("http") else f"{self.base_url}{path}"
        attempt, delay = 0, 0.5
        while True:
            attempt += 1
            _RateLimiter.wait()
            t0 = time.perf_counter()
            try:
                if stream_to is not None:
                    with self._client.stream(method, url, params=params, json=json_body) as r:
                        if r.status_code >= 400:
                            r.read()
                            resp = r
                        else:
                            stream_to.parent.mkdir(parents=True, exist_ok=True)
                            h = hashlib.sha256()
                            with open(stream_to, "wb") as fh:
                                for chunk in r.iter_bytes():
                                    fh.write(chunk)
                                    h.update(chunk)
                            r.sha256 = h.hexdigest()  # type: ignore[attr-defined]
                            resp = r
                else:
                    resp = self._client.request(method, url, params=params, json=json_body)
                elapsed = time.perf_counter() - t0
                self._log({
                    "ts": time.strftime("%Y-%m-%dT%H:%M:%S%z"), "method": method, "path": path, "status": resp.status_code,
                    "request_id": resp.headers.get("X-Request-ID"), "dataset_id": resp.headers.get("X-Dataset-Id"),
                    "dataset_version": resp.headers.get("X-Dataset-Version"), "elapsed_s": round(elapsed, 3), "attempt": attempt,
                })
                if resp.status_code < 400:
                    return resp
                retryable = resp.status_code in (429, 500, 502, 503, 504)
                if resp.status_code == 429:
                    ra = resp.headers.get("Retry-After")
                    delay = max(delay, float(ra) if ra and ra.replace(".", "", 1).isdigit() else 10.0)
                    retry = True  # a rate-limited call never executed server-side: always safe to retry
                if not (retry and retryable) or attempt >= self.max_attempts:
                    try:
                        err = resp.json().get("error", {})
                    except Exception:
                        err = {"code": "http_error", "message": resp.text[:300]}
                    raise FinDataError(resp.status_code, err.get("code", "http_error"), err.get("message", ""), err.get("request_id"))
            except (httpx.TransportError, httpx.TimeoutException) as exc:
                self._log({"ts": time.strftime("%Y-%m-%dT%H:%M:%S%z"), "method": method, "path": path, "status": None,
                           "error": _redact(type(exc).__name__), "attempt": attempt})
                if not retry or attempt >= self.max_attempts:
                    raise RuntimeError(f"FinData transport failure after {attempt} attempts on {path}: {_redact(str(exc))}") from None
            time.sleep(delay + random.uniform(0, delay / 2))
            delay = min(delay * 2, 60.0)

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

    def post(self, path: str, body: dict, retry: bool = False) -> Any:
        return self._request("POST", path, json_body=body, retry=retry).json()

    # --------------------------------------------------------------- catalog
    def health(self) -> dict:
        return self.get("/health")

    def openapi(self) -> dict:
        return self.get("/openapi.json")

    def datasets(self) -> list[dict]:
        return self.get("/catalog/datasets")["data"]

    def dataset(self, dataset_id: str) -> dict:
        return self.get(f"/catalog/datasets/{dataset_id}")

    def schema_bars(self) -> dict:
        return self.get("/catalog/schema/bars")

    # ----------------------------------------------------------- instruments
    def instrument(self, symbol: str) -> dict:
        return self.get(f"/instruments/{symbol}")

    def instruments_search(self, query: str | None = None, asset_class: str | None = None, status: str | None = None,
                           limit: int = 200) -> Iterator[dict]:
        cursor = None
        while True:
            page = self.get("/instruments", query=query, asset_class=asset_class, status=status, limit=limit, cursor=cursor)
            yield from page.get("data", [])
            cursor = page.get("next_cursor")
            if not cursor:
                return

    def coverage_symbols(self, dataset_id: str, min_sessions: int = 1, limit: int = 5000) -> Iterator[dict]:
        cursor = None
        while True:
            page = self.get("/coverage/symbols", dataset_id=dataset_id, min_sessions=min_sessions, limit=limit, cursor=cursor)
            yield from page.get("data", [])
            cursor = page.get("next_cursor")
            if not cursor:
                return

    def coverage(self, dataset_id: str, symbol: str, start: str, end: str, limit: int = 5000) -> list[dict]:
        return self.get("/coverage", dataset_id=dataset_id, symbol=symbol, start=start, end=end, limit=limit)["data"]

    # ---------------------------------------------------------- fundamentals
    def fundamentals(self, entity_id: str, as_of: str | None = None, metrics: list[str] | None = None,
                     period_type: str | None = "q", limit: int = 20000, include_superseded: bool = False) -> dict:
        return self.get(f"/fundamentals/{entity_id}", as_of=as_of, metrics=",".join(metrics) if metrics else None,
                        period_type=period_type, limit=limit, include_superseded=str(include_superseded).lower())

    def fundamentals_periods(self, entity_id: str) -> list[dict]:
        return self.get(f"/fundamentals/{entity_id}/periods")["periods"]

    # ----------------------------------------------------------------- query
    def query(self, sql: str, max_rows: int = 1000) -> tuple[list[str], list[list[Any]], dict]:
        d = self.post("/query", {"sql": sql, "max_rows": max_rows}, retry=True)
        return [c["name"] for c in d["columns"]], d["rows"], d.get("meta", {})

    # --------------------------------------------------------------- exports
    def export(self, dataset_id: str, symbols: list[str], start: str, end: str, dest: Path, adjustment: str = "raw",
               poll_s: float = 1.0, max_wait_s: float = 600.0) -> ExportResult:
        body = {"dataset_id": dataset_id, "symbols": symbols, "start": start, "end": end, "adjustment": adjustment, "format": "parquet"}
        r = self._request("POST", "/exports", json_body=body, retry=False)
        created = r.json()
        export_id, request_id = created["id"], r.headers.get("X-Request-ID")
        t0 = time.time()
        while True:
            st = self.get(f"/exports/{export_id}")
            if st["status"] == "ready":
                break
            if st["status"] in ("failed", "expired"):
                raise FinDataError(500, "export_" + st["status"], str(st.get("error")), request_id)
            if time.time() - t0 > max_wait_s:
                raise FinDataError(504, "export_timeout", f"export {export_id} not ready after {max_wait_s}s", request_id)
            time.sleep(poll_s)
        resp = self._request("GET", f"/exports/{export_id}/download", stream_to=dest)
        snap = st.get("snapshot") or {}
        return ExportResult(export_id=export_id, dataset_id=dataset_id, dataset_version=snap.get("version"), symbols=symbols,
                            start=start, end=end, adjustment=adjustment, row_count=int(st.get("row_count") or 0),
                            bytes=int(st.get("bytes") or 0), sha256=getattr(resp, "sha256", ""), path=str(dest),
                            request_id=request_id, exported_at=time.strftime("%Y-%m-%dT%H:%M:%S%z"))

    def close(self) -> None:
        self._client.close()


def export_result_dict(res: ExportResult) -> dict:
    return asdict(res)