"""Typed, redacting client for the FinData OS API (contract: docs/DATA_CARD.md §connector).

Rules: bearer token from env/secret file only (never logged, never in URLs); bounded exponential backoff with jitter on
429/5xx; every call appends a provenance record (method, path, status, X-Request-ID, X-Dataset-Id, X-Dataset-Version,
elapsed) to a JSONL log; exports are validated against the published bars schema and hashed on disk.
"""
from __future__ import annotations

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

import httpx
import pyarrow as pa
import pyarrow.ipc as ipc
import pyarrow.parquet as pq

_TOKEN_RE = re.compile(r"fd_[A-Za-z0-9_.\-]+")
DEFAULT_BASE = "https://findata.spboucher.ai/api/v1"
BARS_1M_MAX_DAYS = 31
EXPORT_MAX_SYMBOLS = 500


def redact(text: str) -> str:
    return _TOKEN_RE.sub("fd_<REDACTED>", str(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()
    candidates = [os.environ.get("FINDATA_API_KEY_FILE"), "~/findata-secrets/api-key-wp41.txt",
                  "~/findata-secrets/api-key-claude-code.txt",
                  "~/Desktop/Cluster/secrets/findata/api-key-claude-code.txt"]
    for c in candidates:
        if not c:
            continue
        p = Path(os.path.expanduser(c))
        if p.exists():
            return p.read_text().strip()
    raise FileNotFoundError("FinData API key not found (FINDATA_API_KEY or FINDATA_API_KEY_FILE)")


def base_url() -> str:
    return os.environ.get("FINDATA_BASE_URL", DEFAULT_BASE).rstrip("/")


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

    @classmethod
    def wait(cls) -> None:
        rate = float(os.environ.get("WP41_RATE", "5"))
        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()


@dataclass
class Provenance:
    ts: str
    method: str
    path: str
    status: int
    request_id: str | None
    dataset_id: str | None
    dataset_version: str | None
    elapsed_ms: int
    note: str | None = None


@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
    retrieved_at: str
    revision_ids: list[int]


def sha256_file(path: str | Path) -> str:
    h = hashlib.sha256()
    with open(path, "rb") as fh:
        for chunk in iter(lambda: fh.read(1 << 20), b""):
            h.update(chunk)
    return h.hexdigest()


class FinDataClient:
    def __init__(self, base: str | None = None, api_key: str | None = None, log_path: str | Path | None = None,
                 timeout: float = 180.0, max_retries: int = 6):
        self.base = (base or base_url()).rstrip("/")
        self._key = api_key or load_api_key()
        self.log_path = Path(log_path) if log_path else None
        self.max_retries = max_retries
        self._client = httpx.Client(base_url=self.base, timeout=timeout,
                                    headers={"Authorization": f"Bearer {self._key}", "User-Agent": "wp41-chronos-x/0.1"})
        self._lock = threading.Lock()

    # ---------------------------------------------------------------- low level
    def _log(self, rec: Provenance) -> None:
        if not self.log_path:
            return
        self.log_path.parent.mkdir(parents=True, exist_ok=True)
        with self._lock, open(self.log_path, "a") as fh:
            fh.write(json.dumps(asdict(rec)) + "\n")

    def request(self, method: str, path: str, *, params: dict | None = None, json_body: dict | None = None,
                headers: dict | None = None, stream_to: Path | None = None, idempotent: bool = True) -> httpx.Response:
        attempt = 0
        while True:
            _RateLimiter.wait()
            t0 = time.monotonic()
            try:
                if stream_to is not None:
                    with self._client.stream(method, path, params=params, headers=headers) as r:
                        if r.status_code < 400:
                            stream_to.parent.mkdir(parents=True, exist_ok=True)
                            with open(stream_to, "wb") as fh:
                                for chunk in r.iter_bytes(1 << 20):
                                    fh.write(chunk)
                        else:
                            r.read()
                        resp = r
                else:
                    resp = self._client.request(method, path, params=params, json=json_body, headers=headers)
            except (httpx.TransportError,) as exc:
                attempt += 1
                if attempt > self.max_retries or not idempotent:
                    raise FinDataError(0, "transport", redact(str(exc)))
                time.sleep(min(30, 0.5 * 2 ** attempt) + random.random())
                continue
            elapsed = int((time.monotonic() - t0) * 1000)
            self._log(Provenance(datetime.now(timezone.utc).isoformat(), method, redact(path), resp.status_code,
                                 resp.headers.get("x-request-id"), resp.headers.get("x-dataset-id"),
                                 resp.headers.get("x-dataset-version"), elapsed))
            if resp.status_code in (429, 500, 502, 503, 504) and attempt < self.max_retries and idempotent:
                attempt += 1
                ra = resp.headers.get("retry-after")
                time.sleep(float(ra) if ra and ra.isdigit() else min(60, 0.5 * 2 ** attempt) + random.random())
                continue
            if resp.status_code >= 400:
                try:
                    err = resp.json().get("error", {})
                except Exception:
                    err = {}
                raise FinDataError(resp.status_code, err.get("code", "http_error"),
                                   err.get("message", resp.text[:300]), err.get("request_id"))
            return resp

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

    def post_json(self, path: str, body: dict, idempotent: bool = False) -> dict:
        return self.request("POST", path, json_body=body, idempotent=idempotent).json()

    # ---------------------------------------------------------------- catalog & reference
    def health(self) -> dict:
        return self.get_json("/health")

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

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

    def bars_schema(self) -> dict:
        return self.get_json("/catalog/schema/bars")

    def openapi(self) -> dict:
        return self.request("GET", "/openapi.json").json()

    def instruments(self, asset_class: str | None = None, status: str | None = None, query: str = "",
                    limit: int = 500) -> Iterator[dict]:
        cursor = None
        while True:
            d = self.get_json("/instruments", query=query or None, asset_class=asset_class, status=status, limit=limit,
                              cursor=cursor)
            yield from d["data"]
            cursor = d.get("next_cursor")
            if not cursor or not d["data"]:
                break

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

    def sessions(self, symbol: str, start: str, end: str, interval: str = "1m") -> list[dict]:
        return self.get_json("/market/sessions", symbol=symbol, start=start, end=end, interval=interval)["data"]

    def coverage(self, symbol: str, start: str, end: str, dataset_id: str = "market_bars.stock.1m.raw",
                 limit: int = 5000) -> list[dict]:
        return self.get_json("/coverage", symbol=symbol, start=start, end=end, dataset_id=dataset_id, limit=limit)["data"]

    def anomalies(self, symbol: str, dataset_id: str = "market_bars.stock.1m.raw", limit: int = 5000) -> list[dict]:
        return self.get_json("/quality/anomalies", symbol=symbol, dataset_id=dataset_id, limit=limit)["data"]

    def sql(self, sql: str, max_rows: int = 100_000) -> tuple[list[str], list[dict], dict]:
        d = self.post_json("/query", {"sql": sql, "max_rows": max_rows}, idempotent=True)
        return [c["name"] for c in d["columns"]], d["rows"], d["meta"]

    # ---------------------------------------------------------------- bars (small, paginated, Arrow transport)
    def bars(self, symbol: str, start: str, end: str, interval: str = "1m", adjustment: str = "raw",
             limit: int = 10_000) -> pa.Table:
        """All pages for one symbol/range (1m ≤ 31 days per call enforced by the server)."""
        tables, cursor = [], None
        while True:
            params = {"symbol": symbol, "interval": interval, "start": start, "end": end, "adjustment": adjustment,
                      "limit": limit, "format": "arrow"}
            if cursor:
                params["cursor"] = cursor
            r = self.request("GET", "/market/bars", params=params)
            with ipc.open_stream(r.content) as reader:
                tables.append(reader.read_all())
            cursor = r.headers.get("x-next-cursor")
            if not cursor:
                break
        if not tables:
            return pa.table({})
        return pa.concat_tables(tables)

    # ---------------------------------------------------------------- exports (large, async, pinned version)
    def export(self, symbols: list[str], start: str, end: str, dest: Path, dataset_id: str = "market_bars.stock.1m.raw",
               adjustment: str = "raw", poll: float = 2.0, max_wait: float = 3600.0) -> ExportResult:
        if len(symbols) > EXPORT_MAX_SYMBOLS:
            raise ValueError(f"max {EXPORT_MAX_SYMBOLS} symbols per export")
        body = {"dataset_id": dataset_id, "symbols": symbols, "start": start, "end": end, "adjustment": adjustment,
                "format": "parquet"}
        r = self.request("POST", "/exports", json_body=body, idempotent=False)
        created = r.json()
        eid = created["id"]
        t0 = time.monotonic()
        while True:
            st = self.get_json(f"/exports/{eid}")
            if st["status"] == "ready":
                break
            if st["status"] in ("failed", "expired", "cancelled"):
                raise FinDataError(500, "export_" + st["status"], str(st.get("error")), r.headers.get("x-request-id"))
            if time.monotonic() - t0 > max_wait:
                raise FinDataError(504, "export_timeout", f"export {eid} not ready after {max_wait}s")
            time.sleep(poll)
        tmp = dest.with_suffix(dest.suffix + ".part")
        rd = self.request("GET", f"/exports/{eid}/download", stream_to=tmp)
        meta = pq.read_metadata(tmp)
        tbl_schema = pq.read_schema(tmp)
        validate_bars_schema(tbl_schema)
        rev = pq.read_table(tmp, columns=["revision_id"]).column("revision_id").unique().to_pylist()
        tmp.replace(dest)
        snap = st.get("snapshot") or {}
        return ExportResult(eid, dataset_id, snap.get("version"), symbols, start, end, adjustment,
                            int(st.get("row_count") or meta.num_rows), int(st.get("bytes") or dest.stat().st_size),
                            sha256_file(dest), str(dest), rd.headers.get("x-request-id") or r.headers.get("x-request-id"),
                            datetime.now(timezone.utc).isoformat(), sorted(int(x) for x in rev))


BARS_REQUIRED = {"instrument_id", "symbol_as_reported", "asset_class", "interval", "timestamp_utc", "session_date",
                 "open", "high", "low", "close", "volume", "adjustment_type", "quality_flags", "revision_id"}


def validate_bars_schema(schema: pa.Schema) -> None:
    missing = BARS_REQUIRED - set(schema.names)
    if missing:
        raise ValueError(f"bars schema missing columns: {sorted(missing)}")
    ts = schema.field("timestamp_utc").type
    if not pa.types.is_timestamp(ts) or ts.tz is None:
        raise ValueError(f"timestamp_utc must be tz-aware timestamp, got {ts}")
    for c in ("open", "high", "low", "close", "volume"):
        if not pa.types.is_floating(schema.field(c).type):
            raise ValueError(f"{c} must be float, got {schema.field(c).type}")