from __future__ import annotations

import json
import threading
from datetime import datetime, timezone
from decimal import Decimal, InvalidOperation
from typing import Any, Callable
from urllib.parse import urlparse

import websocket


PriceCallback = Callable[[str, str, datetime], None]
SymbolProvider = Callable[[], list[str]]


def clean_price(value: object) -> str | None:
    try:
        price = Decimal(str(value))
    except (InvalidOperation, TypeError, ValueError):
        return None
    if not price.is_finite() or price <= 0:
        return None
    return format(price.normalize(), "f")


def parse_proxy(proxy: str | None) -> dict[str, Any]:
    """Turn a proxy URL into websocket-client ``create_connection`` kwargs.

    Returns an empty dict when no proxy is set or the URL is unusable, so
    the stream keeps its direct-connection behaviour by default.
    """
    text = str(proxy or "").strip()
    if not text:
        return {}
    parsed = urlparse(text)
    scheme = parsed.scheme.lower()
    if scheme not in ("http", "https", "socks4", "socks5", "socks5h"):
        return {}
    if not parsed.hostname or not parsed.port:
        return {}
    kwargs: dict[str, Any] = {
        "proxy_type": "http" if scheme in ("http", "https") else scheme.rstrip("h"),
        "http_proxy_host": parsed.hostname,
        "http_proxy_port": parsed.port,
    }
    if parsed.username:
        kwargs["http_proxy_auth"] = (parsed.username, parsed.password or "")
    return kwargs


class RealtimePriceStream:
    """Public Binance Spot prices for custom monitors; never authenticates or trades."""

    endpoint = "wss://stream.binance.com:9443/stream?streams="

    def __init__(
        self,
        symbols: SymbolProvider,
        on_price: PriceCallback | None = None,
        *,
        reconnect_seconds: int = 3,
        proxy: str | None = None,
    ) -> None:
        self._symbols = symbols
        self._on_price = on_price
        self._reconnect_seconds = max(1, int(reconnect_seconds))
        self._proxy = str(proxy).strip() if proxy and str(proxy).strip() else None
        self._stop = threading.Event()
        self._thread: threading.Thread | None = None
        self._lock = threading.Lock()
        self._socket: websocket.WebSocket | None = None
        self._prices: dict[str, dict[str, str]] = {}
        self._connected = False
        self._last_error: str | None = None
        self._active_symbols: tuple[str, ...] = ()

    def set_proxy(self, proxy: str | None) -> None:
        """Change the proxy for future reconnects (applied on next loop)."""
        with self._lock:
            self._proxy = str(proxy).strip() if proxy and str(proxy).strip() else None

    @property
    def proxy(self) -> str | None:
        with self._lock:
            return self._proxy

    def start(self) -> None:
        if self._thread is not None and self._thread.is_alive():
            return
        self._stop.clear()
        self._thread = threading.Thread(
            target=self._run,
            name="binance-public-price-stream",
            daemon=True,
        )
        self._thread.start()

    def stop(self) -> None:
        self._stop.set()
        socket = self._socket
        if socket is not None:
            try:
                socket.close()
            except Exception:
                pass
        if self._thread is not None:
            self._thread.join(timeout=5)

    def snapshot(self) -> dict[str, Any]:
        with self._lock:
            return {
                "connected": self._connected,
                "active_symbols": list(self._active_symbols),
                "prices": {symbol: dict(item) for symbol, item in self._prices.items()},
                "last_error": self._last_error,
                "thread_alive": self._thread is not None and self._thread.is_alive(),
                "proxy_active": self._proxy is not None,
            }

    def ingest_message(self, raw: str) -> bool:
        try:
            message = json.loads(raw)
        except (TypeError, ValueError):
            return False
        data = message.get("data", message) if isinstance(message, dict) else None
        if not isinstance(data, dict):
            return False
        symbol = str(data.get("s", "")).strip().upper()
        price = clean_price(data.get("c"))
        if not symbol or price is None:
            return False
        observed = datetime.now(timezone.utc)
        with self._lock:
            self._prices[symbol] = {
                "price": price,
                "updated_at": observed.isoformat(),
            }
            self._last_error = None
        if self._on_price is not None:
            self._on_price(symbol, price, observed)
        return True

    def _current_symbols(self) -> tuple[str, ...]:
        return tuple(
            sorted(
                {
                    str(symbol).strip().upper()
                    for symbol in self._symbols()
                    if str(symbol).strip().upper().endswith("USDT")
                }
            )
        )

    def _set_connection(self, connected: bool, symbols: tuple[str, ...]) -> None:
        with self._lock:
            self._connected = connected
            self._active_symbols = symbols

    def _run(self) -> None:
        while not self._stop.is_set():
            symbols = self._current_symbols()
            if not symbols:
                self._set_connection(False, ())
                self._stop.wait(1)
                continue
            streams = "/".join(f"{symbol.lower()}@miniTicker" for symbol in symbols)
            try:
                with self._lock:
                    proxy_kwargs = parse_proxy(self._proxy)
                socket = websocket.create_connection(
                    f"{self.endpoint}{streams}",
                    timeout=10,
                    enable_multithread=True,
                    **proxy_kwargs,
                )
                socket.settimeout(5)
                self._socket = socket
                self._set_connection(True, symbols)
                while not self._stop.is_set():
                    if self._current_symbols() != symbols:
                        break
                    try:
                        raw = socket.recv()
                    except websocket.WebSocketTimeoutException:
                        continue
                    if not raw:
                        break
                    self.ingest_message(str(raw))
            except Exception as exc:
                with self._lock:
                    self._last_error = type(exc).__name__
            finally:
                self._set_connection(False, symbols)
                socket = self._socket
                self._socket = None
                if socket is not None:
                    try:
                        socket.close()
                    except Exception:
                        pass
            self._stop.wait(self._reconnect_seconds)