#!/usr/bin/env python3
# Long-poll Terminal's events feed and keep recent alerts per coin.
# A Freqtrade strategy imports TerminalFeed and reads recent(). This file does not place an order.
# The API key is read from the environment and sent only as an Authorization header.

import argparse
import copy
import json
import os
import sys
import threading
import time
import urllib.error
import urllib.parse
import urllib.request
from datetime import datetime, timezone

DEFAULT_API_URL = "https://terminal.tokensmarketcap.com"
PER_SYMBOL = 50
SEEN_CAP = 10000
WAIT_SECONDS = 25
REQUEST_TIMEOUT = 40
BODY_LIMIT = 1000000
MAX_BACKOFF = 60
MAX_RETRY_AFTER = 120
FAST_EMPTY_SECONDS = 1
FAST_EMPTY_THRESHOLD = 2
CURSOR_LIMIT = 512
USER_AGENT = "Terminal-Freqtrade/1"


class _Unset:
    pass


UNSET = _Unset()


class _RefuseRedirect(urllib.request.HTTPRedirectHandler):
    def redirect_request(self, req, fp, code, msg, headers, newurl):
        raise urllib.error.HTTPError(req.full_url, code, "redirect refused", headers, fp)


def symbol_from_pair(pair):
    """Coin symbol from a Freqtrade pair such as SOL/USDT or SOL/USDT:USDT."""
    if not isinstance(pair, str):
        return ""
    text = pair.strip()
    if not text:
        return ""
    text = text.split(":", 1)[0]
    text = text.split("/", 1)[0]
    return text.strip().upper()


def events_url(api_url, cursor):
    """URL for one events request. The API key is never added here.

    TERMINAL_API_URL may be the origin (https://terminal.tokensmarketcap.com),
    the API base (.../v1) or the feed itself (.../v1/events)."""
    base = str(api_url).rstrip("/")
    if base.endswith("/v1/events"):
        base = base[: -len("/events")]
    if not base.endswith("/v1"):
        base += "/v1"
    query = {"wait": str(WAIT_SECONDS)}
    if cursor:
        query["cursor"] = cursor
    else:
        query["start"] = "latest"
    return base + "/events?" + urllib.parse.urlencode(query)


def parse_time(value):
    """Unix seconds for a Terminal time string, or None."""
    if not isinstance(value, str) or len(value) < 20:
        return None
    text = value[:-1] + "+00:00" if value.endswith("Z") else value
    try:
        parsed = datetime.fromisoformat(text)
    except ValueError:
        return None
    if parsed.tzinfo is None:
        parsed = parsed.replace(tzinfo=timezone.utc)
    return parsed.timestamp()


def parse_retry_after(value):
    """Whole seconds from a Retry-After header, capped, or None when absent."""
    if value is None:
        return None
    text = str(value).strip()
    if not text:
        return None
    try:
        number = float(text)
    except ValueError:
        return None
    if number != number or number in (float("inf"), float("-inf")):
        return None
    if number <= 0:
        seconds = 1
    else:
        seconds = int(number) if number.is_integer() else int(number) + 1
    if seconds < 1:
        seconds = 1
    if seconds > MAX_RETRY_AFTER:
        seconds = MAX_RETRY_AFTER
    return seconds


def as_page(value):
    """A feed answer, or a single event wrapped as a one-event answer."""
    if not isinstance(value, dict):
        return None
    if isinstance(value.get("events"), list):
        return value
    if value.get("schema") == "terminal.alert.v1" or isinstance(value.get("asset"), dict):
        return {"events": [value], "has_more": False, "gap": False}
    return None


def valid_cursor(cursor, api_key=""):
    """A cursor is a short opaque token. It must not contain the API key or whitespace."""
    if not isinstance(cursor, str) or not cursor or len(cursor) > CURSOR_LIMIT:
        return False
    if any(char.isspace() for char in cursor):
        return False
    if api_key and api_key in cursor:
        return False
    return True


CURSOR_UNREADABLE = (
    "The cursor file could not be read, so the feed starts after the newest alert. "
    "Alerts recorded while the strategy was stopped are not read."
)


def _load_cursor(path):
    """(cursor, problem). A missing file is not a problem: the feed starts after the newest alert."""
    try:
        if path and os.path.getsize(path) > 4096:
            return None, CURSOR_UNREADABLE
        with open(path, encoding="utf-8") as handle:
            data = json.load(handle)
    except FileNotFoundError:
        return None, None
    except (OSError, ValueError, TypeError):
        return None, CURSOR_UNREADABLE
    if isinstance(data, dict):
        cursor = data.get("cursor")
    else:
        cursor = data
    if valid_cursor(cursor):
        return cursor, None
    return None, CURSOR_UNREADABLE


def _configuration_error(api_key, api_url):
    if not api_key:
        return "Set TERMINAL_API_KEY to your API key."
    # A key is printable ASCII with no spaces. A line break left in by a copy
    # would make every request fail.
    if any(ord(char) < 0x21 or ord(char) > 0x7E for char in api_key):
        return "TERMINAL_API_KEY has a character an API key never has, such as a space or a line break. Copy the key again."
    if api_key in api_url:
        return "TERMINAL_API_URL must not contain the API key."
    if "?" in api_url or "#" in api_url:
        return "TERMINAL_API_URL must be the API base, without a query string."
    parsed = urllib.parse.urlparse(api_url)
    if parsed.scheme not in ("https", "http") or not parsed.netloc:
        return "TERMINAL_API_URL must be an http or https URL."
    return None


class TerminalFeed:
    """Background reader for GET /v1/events.

    recent(symbol, minutes) returns alerts kept for that coin whose fired_at
    is within the last `minutes` minutes. The key never goes in a URL or a log line.
    """

    def __init__(self, api_key=None, api_url=None, cursor_path=UNSET, per_symbol=PER_SYMBOL, log=None):
        if api_key is None:
            api_key = os.environ.get("TERMINAL_API_KEY", "")
        self._api_key = api_key.strip() if isinstance(api_key, str) else ""
        if api_url is None:
            api_url = os.environ.get("TERMINAL_API_URL", "")
        if not isinstance(api_url, str) or not api_url.strip():
            api_url = DEFAULT_API_URL
        self._api_url = api_url.strip()
        if cursor_path is UNSET:
            configured = os.environ.get("TERMINAL_CURSOR_PATH", "").strip()
            cursor_path = configured or os.path.join(os.getcwd(), "terminal-events.cursor")
        self._cursor_path = cursor_path or None
        try:
            count = int(per_symbol)
        except (TypeError, ValueError):
            count = PER_SYMBOL
        self._per_symbol = count if count > 0 else PER_SYMBOL
        self._log_fn = log
        self._lock = threading.Lock()
        self._stop = threading.Event()
        self._thread = None
        self._seen = {}
        self._by_symbol = {}
        self._cursor = None
        self._cursor_refused = False
        loaded, problem = _load_cursor(self._cursor_path) if self._cursor_path else (None, None)
        if problem:
            self._log(problem)
        if loaded and not valid_cursor(loaded, self._api_key):
            self._cursor_refused = True
            self._log("The saved cursor contains the API key. Delete the cursor file.")
        else:
            self._cursor = loaded
        self._opener = urllib.request.build_opener(_RefuseRedirect())

    @property
    def cursor(self):
        with self._lock:
            return self._cursor

    @property
    def api_url(self):
        return self._api_url

    def _log(self, message):
        text = self._redact(str(message))
        if self._log_fn is not None:
            self._log_fn(text)
            return
        print(text, file=sys.stderr)

    def _redact(self, text):
        key = self._api_key
        if not key:
            return text
        redacted = text.replace(key, "[redacted]")
        quoted = urllib.parse.quote(key, safe="")
        if quoted and quoted != key:
            redacted = redacted.replace(quoted, "[redacted]")
        return redacted

    def _usable_cursor(self):
        with self._lock:
            cursor = self._cursor
        if valid_cursor(cursor, self._api_key):
            return cursor
        if isinstance(cursor, str) and self._api_key and self._api_key in cursor and not self._cursor_refused:
            self._cursor_refused = True
            self._log("The saved cursor contains the API key. Delete the cursor file.")
        return None

    def _persist_cursor(self):
        path = self._cursor_path
        cursor = self._cursor
        if not path or not isinstance(cursor, str) or not cursor:
            return
        if self._api_key and self._api_key in cursor:
            return
        temporary = path + ".tmp"
        try:
            directory = os.path.dirname(path)
            if directory:
                os.makedirs(directory, exist_ok=True)
            with open(temporary, "w", encoding="utf-8") as handle:
                json.dump({"cursor": cursor}, handle)
                handle.write("\n")
                handle.flush()
                os.fsync(handle.fileno())
            os.replace(temporary, path)
        except OSError:
            self._log("could not save the cursor file")

    def _remember(self, event_id):
        seen = self._seen
        seen.pop(event_id, None)
        seen[event_id] = True
        while len(seen) > SEEN_CAP:
            oldest = next(iter(seen))
            del seen[oldest]

    def _take(self, event):
        """Store one event. The caller has the lock. Returns a kind and a payload."""
        if not isinstance(event, dict):
            return "skipped", None
        event_id = event.get("id")
        if not isinstance(event_id, str) or not event_id or len(event_id) > 200:
            return "skipped", None
        if event_id in self._seen:
            self._remember(event_id)
            return "duplicate", event_id
        asset = event.get("asset")
        symbol = asset.get("symbol") if isinstance(asset, dict) else None
        if not isinstance(symbol, str) or not symbol.strip():
            return "skipped", event_id
        self._remember(event_id)
        key = symbol.strip().upper()
        bucket = self._by_symbol.setdefault(key, [])
        bucket.append(event)
        extra = len(bucket) - self._per_symbol
        if extra > 0:
            del bucket[0:extra]
        return "accepted", event

    def apply_page(self, page):
        """Ingest one feed answer. Saves next_cursor when the answer includes one."""
        if not isinstance(page, dict):
            return {"outcomes": [], "accepted": 0, "dropped": 0, "gap": False, "cursor": self.cursor}
        events = page.get("events")
        if not isinstance(events, list):
            events = []
        outcomes = []
        accepted = 0
        dropped = 0
        persist = False
        with self._lock:
            for event in events:
                kind, payload = self._take(event)
                outcomes.append((kind, payload))
                if kind == "accepted":
                    accepted += 1
                elif kind == "duplicate":
                    dropped += 1
            cursor = page.get("next_cursor")
            if valid_cursor(cursor, self._api_key):
                self._cursor = cursor
                persist = True
            gap = page.get("gap") is True
            saved = self._cursor
        if persist:
            self._persist_cursor()
        return {
            "outcomes": outcomes,
            "accepted": accepted,
            "dropped": dropped,
            "gap": gap,
            "cursor": saved,
        }

    def replay(self, path):
        """Read a recorded file of feed answers. Does not call the API."""
        with open(path, encoding="utf-8") as handle:
            for raw in handle:
                line = raw.strip()
                if not line:
                    continue
                record = json.loads(line)
                page = as_page(record)
                if page is None:
                    continue
                self.apply_page(page)
        return self

    def recent(self, symbol, minutes, now=None):
        """Alerts for `symbol` with fired_at at or after now minus `minutes` minutes."""
        key = symbol.strip().upper() if isinstance(symbol, str) else ""
        if not key:
            return []
        try:
            window = float(minutes)
        except (TypeError, ValueError):
            return []
        if window < 0 or window != window or window == float("inf"):
            return []
        moment = time.time() if now is None else float(now)
        if moment != moment or moment in (float("inf"), float("-inf")):
            return []
        cutoff = moment - window * 60
        with self._lock:
            rows = list(self._by_symbol.get(key, ()))
        kept = []
        for event in rows:
            fired = parse_time(event.get("fired_at"))
            if fired is None or fired < cutoff:
                continue
            kept.append(copy.deepcopy(event))
        kept.sort(key=lambda event: parse_time(event.get("fired_at")) or 0)
        return kept

    def start(self):
        """Start the long-poll thread. A second call does nothing while it is running."""
        with self._lock:
            if self._thread is not None and self._thread.is_alive():
                return
            self._stop.clear()
            self._thread = threading.Thread(target=self._run, name="terminal-feed", daemon=True)
            self._thread.start()

    def stop(self, timeout=5):
        """Ask the thread to finish and wait up to `timeout` seconds."""
        self._stop.set()
        thread = self._thread
        if thread is not None and thread is not threading.current_thread():
            thread.join(timeout)

    def _fetch(self):
        cursor = self._usable_cursor()
        url = events_url(self._api_url, cursor)
        if self._api_key and self._api_key in url:
            raise RuntimeError("refusing a URL that contains the API key")
        request = urllib.request.Request(url, method="GET")
        request.add_header("Authorization", "Bearer " + self._api_key)
        request.add_header("Accept", "application/json")
        request.add_header("User-Agent", USER_AGENT)
        try:
            response = self._opener.open(request, timeout=REQUEST_TIMEOUT)
        except urllib.error.HTTPError as error:
            try:
                return error.code, error.headers, error.read(BODY_LIMIT + 1)
            finally:
                error.close()
        with response:
            status = getattr(response, "status", None)
            if status is None:
                status = response.getcode()
            return status, response.headers, response.read(BODY_LIMIT + 1)

    def _decode_page(self, raw):
        if not isinstance(raw, (bytes, bytearray)) or len(raw) > BODY_LIMIT:
            return None
        try:
            page = json.loads(raw.decode("utf-8"))
        except (UnicodeDecodeError, json.JSONDecodeError):
            return None
        if not isinstance(page, dict) or not isinstance(page.get("events"), list):
            return None
        return page

    def _error_code(self, raw):
        page = None
        if isinstance(raw, (bytes, bytearray)) and len(raw) <= BODY_LIMIT:
            try:
                page = json.loads(raw.decode("utf-8"))
            except (UnicodeDecodeError, json.JSONDecodeError):
                page = None
        if isinstance(page, dict):
            error = page.get("error")
            if isinstance(error, dict) and isinstance(error.get("code"), str):
                return error["code"]
        return "none"

    def _run(self):
        problem = _configuration_error(self._api_key, self._api_url)
        if problem:
            self._log(problem)
            return
        self._log("reading the events feed")
        backoff = 1
        while not self._stop.is_set():
            started = time.monotonic()
            try:
                status, headers, raw = self._fetch()
            except Exception as error:
                self._log(f"events request failed ({error.__class__.__name__}), waiting {backoff}s")
                if self._stop.wait(backoff):
                    return
                backoff = min(backoff * 2, MAX_BACKOFF)
                continue
            elapsed = time.monotonic() - started
            if status == 200:
                page = self._decode_page(raw)
                if page is None:
                    self._log(f"events response was not a JSON object, waiting {backoff}s")
                    if self._stop.wait(backoff):
                        return
                    backoff = min(backoff * 2, MAX_BACKOFF)
                    continue
                summary = self.apply_page(page)
                if summary["accepted"] or summary["dropped"] or summary["gap"]:
                    gap = "true" if summary["gap"] else "false"
                    self._log(
                        f"events accepted {summary['accepted']} dropped {summary['dropped']} gap {gap}"
                    )
                backoff = 1
                if page.get("has_more") is True:
                    continue
                # A normal long-poll stays open on the server. A fast answer with nothing new
                # would otherwise spin. Wait briefly, then ask again.
                if summary["accepted"] == 0 and elapsed < FAST_EMPTY_THRESHOLD:
                    if self._stop.wait(FAST_EMPTY_SECONDS):
                        return
                continue
            code = self._error_code(raw)
            # These do not get better by asking again: stop and say why.
            if status in (401, 403):
                self._log(f"The API key was refused ({status} {code}). The feed stopped. Check TERMINAL_API_KEY and its read scope.")
                return
            if status == 400 and self._cursor_refused_by_server(raw):
                self._log("The events feed refused the saved cursor. The feed stopped. Delete the cursor file and start again.")
                return
            if status in (400, 404):
                self._log(f"The events feed answered {status} ({code}). The feed stopped. Check TERMINAL_API_URL.")
                return
            retry = parse_retry_after(headers.get("Retry-After") if headers is not None else None)
            if retry is None:
                wait = backoff
                backoff = min(backoff * 2, MAX_BACKOFF)
            else:
                wait = retry
                backoff = 1
            self._log(f"events request status {status} code {code}, waiting {wait}s")
            if self._stop.wait(wait):
                return

    def _cursor_refused_by_server(self, raw):
        if not isinstance(raw, (bytes, bytearray)) or len(raw) > BODY_LIMIT:
            return False
        try:
            page = json.loads(raw.decode("utf-8"))
        except (UnicodeDecodeError, json.JSONDecodeError):
            return False
        error = page.get("error") if isinstance(page, dict) else None
        fields = error.get("fields") if isinstance(error, dict) else None
        return isinstance(fields, dict) and "cursor" in fields

    @property
    def running(self):
        thread = self._thread
        return thread is not None and thread.is_alive()


_SHARED = {"feed": None}
_SHARED_LOCK = threading.Lock()


def shared_feed(log=None):
    """One running feed per process. Freqtrade builds the strategy again on
    /reload_config; this module stays loaded, so the new strategy gets the same
    feed instead of a second thread writing the same cursor file."""
    with _SHARED_LOCK:
        feed = _SHARED["feed"]
        if feed is None or not feed.running:
            feed = TerminalFeed(log=log)
            feed.start()
            _SHARED["feed"] = feed
        return feed


def _print_summary(feed, summary):
    def emit(line):
        print(feed._redact(line))

    for kind, payload in summary["outcomes"]:
        if kind == "accepted" and isinstance(payload, dict):
            asset = payload.get("asset")
            symbol = asset.get("symbol") if isinstance(asset, dict) else ""
            symbol = symbol.strip().upper() if isinstance(symbol, str) else ""
            event_type = payload.get("type") if isinstance(payload.get("type"), str) else ""
            emit(f"event {payload.get('id')} {symbol} {event_type}".rstrip())
        elif kind == "duplicate":
            emit(f"dropped_duplicate {payload}")
        else:
            emit("skipped")
    if summary["gap"]:
        emit("gap true")


def replay_file(path):
    """Replay a recorded file and print one line per event. Does not call the API."""
    configured = os.environ.get("TERMINAL_CURSOR_PATH", "").strip()
    feed = TerminalFeed(cursor_path=configured or None)
    try:
        handle = open(path, encoding="utf-8")
    except OSError:
        feed._log("could not read the events file")
        return 1
    with handle:
        for raw in handle:
            line = raw.strip()
            if not line:
                continue
            try:
                record = json.loads(line)
            except json.JSONDecodeError:
                print("refused_line")
                continue
            page = as_page(record)
            if page is None:
                print("refused_line")
                continue
            _print_summary(feed, feed.apply_page(page))
    if feed.cursor:
        print(feed._redact(f"cursor {feed.cursor}"))
    return 0


def main(argv):
    parser = argparse.ArgumentParser(description="Read Terminal alerts for a Freqtrade strategy.")
    parser.add_argument(
        "--dry-run",
        metavar="FILE",
        help="Replay a file of recorded feed answers instead of calling the API",
    )
    args = parser.parse_args(argv)
    if args.dry_run:
        return replay_file(args.dry_run)
    feed = TerminalFeed()
    problem = _configuration_error(feed._api_key, feed._api_url)
    if problem:
        feed._log(problem)
        return 1
    feed.start()
    try:
        while feed.running:
            time.sleep(0.5)
    except KeyboardInterrupt:
        feed.stop()
        return 0
    feed.stop()
    return 1


if __name__ == "__main__":
    sys.exit(main(sys.argv[1:]))