from __future__ import annotations

import json
import time
import urllib.error
import urllib.parse
import urllib.request
from dataclasses import dataclass, field
from datetime import datetime, timezone
from decimal import Decimal, InvalidOperation
from sqlite3 import Connection
from typing import Protocol, Sequence

from jarvis_finance.imports.common import stable_id, utc_now
from jarvis_finance.quality.alerts import create_alert
from jarvis_finance.quality.data_quality import resolve_fixed_crypto_price_alerts


@dataclass(frozen=True)
class PriceQuote:
    coingecko_id: str
    currency: str
    price: Decimal | None
    provider: str = "CoinGecko"
    provider_timestamp: str | None = None
    quality_status: str = "fresh"
    error_message: str | None = None


@dataclass
class PriceRefreshResult:
    currency: str
    total_assets: int = 0
    updated_count: int = 0
    skipped_count: int = 0
    cached_count: int = 0
    stale_count: int = 0
    warning_count: int = 0
    error_count: int = 0
    missing_local_price_count: int = 0
    dry_run: bool = False
    written_price_ids: list[str] = field(default_factory=list)
    warnings: list[str] = field(default_factory=list)
    errors: list[str] = field(default_factory=list)

    @property
    def success_count(self) -> int:  # backward-compatible name used by older tests/callers
        return self.updated_count


class MarketDataProvider(Protocol):
    def get_crypto_price(self, coingecko_id: str, currency: str = "CHF") -> PriceQuote: ...


class BatchMarketDataProvider(MarketDataProvider, Protocol):
    def get_crypto_prices(self, coingecko_ids: Sequence[str], currency: str = "CHF") -> dict[str, PriceQuote]: ...


class ProviderRateLimitError(RuntimeError):
    pass


class ProviderFetchError(RuntimeError):
    pass


class CoinGeckoClient:
    provider_key = "coingecko"

    def __init__(self, *, base_url: str = "https://api.coingecko.com/api/v3", max_retries: int = 2, initial_backoff_seconds: float = 1.0, max_backoff_seconds: float = 8.0, opener=None, sleeper=time.sleep) -> None:
        self.base_url = base_url.rstrip("/")
        self.max_retries = max_retries
        self.initial_backoff_seconds = initial_backoff_seconds
        self.max_backoff_seconds = max_backoff_seconds
        self.opener = opener or urllib.request.urlopen
        self.sleeper = sleeper

    def get_crypto_price(self, coingecko_id: str, currency: str = "CHF") -> PriceQuote:
        return self.get_crypto_prices([coingecko_id], currency).get(
            coingecko_id,
            PriceQuote(coingecko_id, currency.upper(), None, quality_status="missing", error_message="price missing"),
        )

    def get_crypto_prices(self, coingecko_ids: Sequence[str], currency: str = "CHF") -> dict[str, PriceQuote]:
        ids = [cid for cid in dict.fromkeys(coingecko_ids) if cid]
        cur = currency.lower()
        if not ids:
            return {}
        query = urllib.parse.urlencode({"ids": ",".join(ids), "vs_currencies": cur, "include_last_updated_at": "true"})
        url = f"{self.base_url}/simple/price?{query}"
        delay = self.initial_backoff_seconds
        last_error: str | None = None
        for attempt in range(self.max_retries + 1):
            try:
                with self.opener(url, timeout=20) as resp:
                    payload = json.loads(resp.read().decode("utf-8"))
                return {coingecko_id: self._quote_from_payload(coingecko_id, currency, payload.get(coingecko_id) or {}) for coingecko_id in ids}
            except urllib.error.HTTPError as exc:
                last_error = f"HTTP {exc.code}"
                if exc.code == 429 and attempt < self.max_retries:
                    self.sleeper(min(delay, self.max_backoff_seconds))
                    delay *= 2
                    continue
                status = "stale" if exc.code == 429 else "error"
                return {cid: PriceQuote(cid, currency.upper(), None, quality_status=status, error_message=last_error) for cid in ids}
            except Exception as exc:  # network skeleton must not crash dashboard jobs
                last_error = str(exc)
                if attempt < self.max_retries:
                    self.sleeper(min(delay, self.max_backoff_seconds))
                    delay *= 2
                    continue
                return {cid: PriceQuote(cid, currency.upper(), None, quality_status="error", error_message=last_error) for cid in ids}
        return {cid: PriceQuote(cid, currency.upper(), None, quality_status="error", error_message=last_error) for cid in ids}

    @staticmethod
    def _quote_from_payload(coingecko_id: str, currency: str, data: dict) -> PriceQuote:
        cur = currency.lower()
        value = data.get(cur)
        if value is None:
            return PriceQuote(coingecko_id, currency.upper(), None, quality_status="missing", error_message="price missing")
        try:
            price = Decimal(str(value))
        except InvalidOperation:
            return PriceQuote(coingecko_id, currency.upper(), None, quality_status="error", error_message="invalid price")
        ts = data.get("last_updated_at")
        provider_ts = None
        if ts:
            try:
                provider_ts = datetime.fromtimestamp(int(ts), tz=timezone.utc).isoformat()
            except (TypeError, ValueError):
                provider_ts = str(ts)
        return PriceQuote(coingecko_id, currency.upper(), price, provider_timestamp=provider_ts)


def _parse_dt(value: str | None) -> datetime | None:
    if not value:
        return None
    try:
        dt = datetime.fromisoformat(value.replace("Z", "+00:00"))
    except ValueError:
        return None
    return dt if dt.tzinfo else dt.replace(tzinfo=timezone.utc)


def is_price_stale(timestamp: str | None, *, max_age_seconds: int = 86_400) -> bool:
    dt = _parse_dt(timestamp)
    if dt is None:
        return True
    return (datetime.now(timezone.utc) - dt).total_seconds() > max_age_seconds


def latest_crypto_price_row(conn: Connection, asset_id: str, currency: str = "CHF"):
    return conn.execute(
        """
        SELECT * FROM crypto_prices
        WHERE asset_id=? AND price_currency=?
        ORDER BY COALESCE(provider_timestamp, fetched_at, '') DESC, fetched_at DESC
        LIMIT 1
        """,
        (asset_id, currency.upper()),
    ).fetchone()


def store_crypto_price(conn: Connection, *, asset_id: str, quote: PriceQuote) -> str:
    now = utc_now()
    price_id = stable_id("cryptoprice", asset_id, quote.coingecko_id, quote.currency, quote.provider, now)
    conn.execute(
        """
        INSERT INTO crypto_prices(crypto_price_id, asset_id, coingecko_id, price_currency, price, provider, provider_timestamp, fetched_at, quality_status, error_message)
        VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
        """,
        (price_id, asset_id, quote.coingecko_id, quote.currency.upper(), format(quote.price, "f") if quote.price is not None else "", quote.provider, quote.provider_timestamp, now, quote.quality_status, quote.error_message),
    )
    if quote.quality_status in {"missing", "stale", "error", "conflict"}:
        create_alert(conn, priority="warnung", category="market_data", entity_type="crypto_asset", entity_id=asset_id, rule_id=f"crypto_price_{quote.quality_status}", message=f"Crypto price quality is {quote.quality_status}.", evidence={"coingecko_id": quote.coingecko_id, "currency": quote.currency, "error": quote.error_message})
    conn.commit()
    return price_id


def _create_missing_local_price_alert(conn: Connection, *, asset_id: str, symbol: str | None, coingecko_id: str | None, currency: str) -> None:
    create_alert(
        conn,
        priority="warnung",
        category="market_data",
        entity_type="crypto_asset",
        entity_id=asset_id,
        rule_id="crypto_price_missing_local",
        message="Crypto asset has no fresh local price after refresh.",
        evidence={"symbol": symbol, "coingecko_id": coingecko_id, "currency": currency},
        fingerprint=f"crypto_price_missing_local:{currency}",
    )


def _has_fresh_local_price(conn: Connection, asset_id: str, currency: str, max_age_seconds: int) -> bool:
    latest = latest_crypto_price_row(conn, asset_id, currency)
    return bool(latest and latest["quality_status"] == "fresh" and not is_price_stale(latest["fetched_at"], max_age_seconds=max_age_seconds))


def _filter_assets(conn: Connection, *, currency: str, max_age_seconds: int, only_missing: bool, only_stale: bool, only_symbol: str | None, limit: int | None):
    assets = conn.execute("SELECT asset_id, coingecko_id, symbol FROM crypto_assets WHERE is_active=1 ORDER BY symbol").fetchall()
    filtered = []
    wanted_symbol = only_symbol.upper() if only_symbol else None
    for asset in assets:
        if wanted_symbol and (asset["symbol"] or "").upper() != wanted_symbol:
            continue
        latest = latest_crypto_price_row(conn, asset["asset_id"], currency)
        has_any = latest is not None and latest["price"] not in (None, "") and latest["quality_status"] == "fresh"
        is_stale = bool(latest and latest["quality_status"] == "fresh" and is_price_stale(latest["fetched_at"], max_age_seconds=max_age_seconds))
        if only_missing and has_any:
            continue
        if only_stale and not is_stale:
            continue
        filtered.append(asset)
        if limit is not None and len(filtered) >= limit:
            break
    return assets, filtered


def _provider_get_batch(provider: MarketDataProvider, coingecko_ids: Sequence[str], currency: str) -> dict[str, PriceQuote]:
    if hasattr(provider, "get_crypto_prices"):
        return provider.get_crypto_prices(coingecko_ids, currency)  # type: ignore[attr-defined]
    return {cid: provider.get_crypto_price(cid, currency) for cid in coingecko_ids}


def refresh_crypto_prices(
    conn: Connection,
    *,
    provider: MarketDataProvider,
    currency: str = "CHF",
    max_age_seconds: int = 3600,
    only_missing: bool = False,
    only_stale: bool = False,
    only_symbol: str | None = None,
    limit: int | None = None,
    dry_run: bool = False,
    sleep_seconds: float = 0.0,
    batch_size: int = 100,
) -> PriceRefreshResult:
    currency = currency.upper()
    result = PriceRefreshResult(currency=currency, dry_run=dry_run)
    all_assets, assets = _filter_assets(conn, currency=currency, max_age_seconds=max_age_seconds, only_missing=only_missing, only_stale=only_stale, only_symbol=only_symbol, limit=limit)
    result.total_assets = len(all_assets)
    if only_missing or only_stale or only_symbol or limit is not None:
        result.skipped_count += max(0, len(all_assets) - len(assets))

    request_assets = []
    for asset in assets:
        asset_id = asset["asset_id"]
        if not asset["coingecko_id"]:
            if not dry_run:
                create_alert(conn, priority="warnung", category="crypto", entity_type="crypto_asset", entity_id=asset_id, rule_id="missing_coingecko_id", message="Crypto asset has no CoinGecko ID; price refresh skipped.", evidence={"symbol": asset["symbol"]}, fingerprint="missing_coingecko_id")
                _create_missing_local_price_alert(conn, asset_id=asset_id, symbol=asset["symbol"], coingecko_id=None, currency=currency)
            result.skipped_count += 1
            result.warning_count += 1
            result.warnings.append(f"{asset['symbol']}: missing_coingecko_id")
            continue
        latest = latest_crypto_price_row(conn, asset_id, currency)
        if latest and latest["quality_status"] == "fresh" and not is_price_stale(latest["fetched_at"], max_age_seconds=max_age_seconds):
            result.cached_count += 1
            continue
        request_assets.append(asset)

    refreshed_fresh_asset_ids: set[str] = set()
    for start in range(0, len(request_assets), max(1, batch_size)):
        if start and sleep_seconds > 0:
            time.sleep(sleep_seconds)
        chunk = request_assets[start : start + max(1, batch_size)]
        ids = [asset["coingecko_id"] for asset in chunk]
        try:
            quotes = _provider_get_batch(provider, ids, currency)
        except Exception as exc:
            quotes = {cid: PriceQuote(cid, currency, None, quality_status="error", error_message=str(exc)) for cid in ids}
        for asset in chunk:
            quote = quotes.get(asset["coingecko_id"]) or PriceQuote(asset["coingecko_id"], currency, None, quality_status="missing", error_message="price missing")
            if not dry_run:
                price_id = store_crypto_price(conn, asset_id=asset["asset_id"], quote=quote)
                result.written_price_ids.append(price_id)
            if quote.price is not None and quote.quality_status == "fresh":
                result.updated_count += 1
                refreshed_fresh_asset_ids.add(asset["asset_id"])
                if not dry_run:
                    resolve_fixed_crypto_price_alerts(conn, currency=currency, max_age_seconds=max_age_seconds)
            elif quote.quality_status in {"missing", "stale"}:
                result.warning_count += 1
                result.warnings.append(f"{asset['symbol']}: {quote.quality_status}")
            else:
                result.error_count += 1
                result.errors.append(f"{asset['symbol']}: {quote.error_message or quote.quality_status}")
            if quote.quality_status == "stale":
                result.stale_count += 1
            if (quote.price is None or quote.quality_status != "fresh") and not dry_run:
                _create_missing_local_price_alert(conn, asset_id=asset["asset_id"], symbol=asset["symbol"], coingecko_id=asset["coingecko_id"], currency=currency)

    if not dry_run:
        for asset in assets:
            if asset["asset_id"] in refreshed_fresh_asset_ids:
                continue
            if not _has_fresh_local_price(conn, asset["asset_id"], currency, max_age_seconds):
                result.missing_local_price_count += 1
                _create_missing_local_price_alert(conn, asset_id=asset["asset_id"], symbol=asset["symbol"], coingecko_id=asset["coingecko_id"], currency=currency)
        conn.commit()
    else:
        # Dry-run reports current known gaps plus simulated failed quotes without mutating alerts/prices.
        for asset in assets:
            if not _has_fresh_local_price(conn, asset["asset_id"], currency, max_age_seconds):
                result.missing_local_price_count += 1
    return result