#!/usr/bin/env python3
# Read Terminal's events feed and publish each event to a Hummingbot MQTT topic.
# The event JSON is published as Terminal sent it. This file does not place an
# order and it does not call a trading API.
# paho-mqtt is imported only when a live run publishes.

import argparse
import json
import os
import re
import sys
import tempfile
import time
import urllib.error
import urllib.parse
import urllib.request

LONG_POLL_SECONDS = 25
PAGE_LIMIT = 50
SEEN_CAP = 10000
MQTT_QOS = 1
MAX_RETRY_AFTER = 3600
USER_AGENT = "Terminal-Hummingbot-Bridge/1"
DEFAULT_API_URL = "https://terminal.tokensmarketcap.com"
DEFAULT_MQTT_HOST = "localhost"
DEFAULT_MQTT_PORT = 1883
DEFAULT_MQTT_TOPIC = "hbot/{instance_id}/external/event/terminal"
PAHO_MESSAGE = "paho-mqtt is not installed. Install it with: pip install paho-mqtt"
KEY_MESSAGE = "Set TERMINAL_API_KEY to your Terminal API key."
KEY_CHARS_MESSAGE = (
    "TERMINAL_API_KEY has a character an API key never has, such as a space or a line break. "
    "Copy the key again."
)
CURSOR_PATTERN = re.compile(r"c1_[A-Za-z0-9_-]{1,64}", re.ASCII)
CURSOR_MESSAGE = "The cursor file does not contain a Terminal cursor. Delete it, or pass another --cursor-file."
MAX_BACKOFF = 60
FAST_EMPTY_SECONDS = 2
TOPIC_MESSAGE = (
    "Set MQTT_TOPIC to your bot's external-event topic, "
    "for example hbot/<instance id>/external/event/terminal. "
    "The instance id is the instance_id in the bot's conf_client.yml. "
    "The default still has {instance_id} in it, so a live run will not publish until you set MQTT_TOPIC."
)


class _NoRedirect(urllib.request.HTTPRedirectHandler):
    def redirect_request(self, req, fp, code, msg, headers, newurl):
        return None


def default_cursor_path():
    return os.path.join(os.path.expanduser("~"), ".local", "share", "terminal", "hummingbot-cursor")


def header_get(headers, name):
    if not headers:
        return None
    getter = getattr(headers, "get", None)
    if getter is not None:
        value = getter(name)
        if value is None:
            value = getter(name.lower())
        if value is not None:
            return value
    try:
        pairs = headers.items()
    except AttributeError:
        return None
    for key, value in pairs:
        if str(key).lower() == name.lower():
            return value
    return None


def retry_after_seconds(headers, default=1):
    raw = header_get(headers, "Retry-After")
    if raw is None:
        return default
    if isinstance(raw, list):
        raw = raw[0] if raw else ""
    text = str(raw).strip()
    if not text:
        return default
    if not text.isdigit():
        return default
    seconds = int(text)
    if seconds < 1:
        return 1
    if seconds > MAX_RETRY_AFTER:
        return MAX_RETRY_AFTER
    return seconds


def events_collection_url(base):
    text = (base or "").strip().rstrip("/")
    if not text:
        text = DEFAULT_API_URL
    if text.endswith("/v1/events"):
        return text
    if text.endswith("/v1"):
        return text + "/events"
    return text + "/v1/events"


def check_base_url(base):
    text = (base or "").strip()
    if not text:
        text = DEFAULT_API_URL
    parts = urllib.parse.urlsplit(text)
    if parts.scheme not in ("https", "http") or not parts.netloc:
        return "TERMINAL_API_URL must be an http or https origin, for example https://terminal.tokensmarketcap.com."
    if parts.username or parts.password or "@" in parts.netloc:
        return "TERMINAL_API_URL must not contain a key or a password."
    if "tmk_" in text:
        return "TERMINAL_API_URL must not contain an API key."
    return None


def build_events_url(base, cursor, wait, start="latest", limit=PAGE_LIMIT):
    params = {"wait": str(int(wait)), "limit": str(int(limit))}
    if cursor:
        params["cursor"] = cursor
    else:
        params["start"] = start
    return events_collection_url(base) + "?" + urllib.parse.urlencode(params)


def request_for(url, key):
    request = urllib.request.Request(url, method="GET")
    request.add_header("Accept", "application/json")
    request.add_header("User-Agent", USER_AGENT)
    request.add_header("Authorization", "Bearer " + key)
    return request


def http_get(url, key, timeout):
    opener = urllib.request.build_opener(_NoRedirect)
    request = request_for(url, key)
    try:
        with opener.open(request, timeout=timeout) as response:
            return response.status, response.headers, response.read()
    except urllib.error.HTTPError as error:
        body = error.read()
        return error.code, error.headers, body


def interpret_response(status, headers, raw):
    if status == 429:
        return {"action": "retry", "retry_after": retry_after_seconds(headers, default=1)}
    if 300 <= status < 400:
        return {
            "action": "stop",
            "message": "The events feed redirected. This script does not follow redirects. Check TERMINAL_API_URL.",
        }
    if status == 408 or 500 <= status <= 599:
        return {"action": "retry", "retry_after": retry_after_seconds(headers, default=5)}
    if status != 200:
        return {"action": "stop", "message": stop_message(status, raw)}
    try:
        body = json.loads(raw.decode("utf-8"))
    except (UnicodeDecodeError, json.JSONDecodeError):
        return {"action": "retry", "retry_after": 5}
    if not isinstance(body, dict) or not isinstance(body.get("events"), list):
        return {"action": "retry", "retry_after": 5}
    cursor = body.get("next_cursor")
    if not isinstance(cursor, str) or not cursor:
        cursor = None
    return {
        "action": "page",
        "events": body["events"],
        "cursor": cursor,
        "has_more": body.get("has_more") is True,
        "gap": body.get("gap") is True,
    }


def stop_message(status, raw):
    code = error_code(raw)
    if status == 401:
        text = "The API key was refused."
    elif status == 403:
        text = "This API key cannot read events."
    elif status == 400:
        text = "The events feed rejected the request. The cursor file may be wrong."
    else:
        text = f"The events feed answered {status}."
    if code:
        return f"{text} ({code})"
    return text


def error_code(raw):
    try:
        body = json.loads(raw.decode("utf-8"))
    except (UnicodeDecodeError, json.JSONDecodeError, AttributeError):
        return None
    if not isinstance(body, dict):
        return None
    error = body.get("error")
    if not isinstance(error, dict):
        return None
    code = error.get("code")
    if isinstance(code, str) and code and len(code) <= 80 and " " not in code:
        return code
    return None


def payload_of(event):
    return json.dumps(event, separators=(",", ":"), ensure_ascii=False)


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


def deliver(events, seen, emit):
    for event in events:
        if not isinstance(event, dict):
            print("Skipped an event that is not an object.", file=sys.stderr)
            continue
        event_id = event.get("id")
        if not isinstance(event_id, str) or not event_id:
            print("Skipped an event with no id.", file=sys.stderr)
            continue
        if event_id in seen:
            print(f"dropped duplicate {event_id}", file=sys.stderr)
            continue
        if not emit(event):
            return False
        remember(seen, event_id)
    return True


def events_from_record(record):
    if not isinstance(record, dict):
        return [], None
    if isinstance(record.get("events"), list):
        cursor = record.get("next_cursor")
        if not isinstance(cursor, str) or not cursor:
            cursor = None
        return record["events"], cursor
    if isinstance(record.get("id"), str) and record.get("id"):
        return [record], None
    return [], None


def load_cursor(path):
    cursor, _problem = read_cursor(path)
    return cursor


def read_cursor(path):
    # Returns (cursor, problem). Only a c1_ cursor is sent: anything else in the
    # file (another file passed by mistake) never reaches a query string.
    try:
        handle = open(path, encoding="utf-8")
    except FileNotFoundError:
        return None, None
    except (OSError, ValueError):
        return None, "Could not read the cursor file."
    with handle:
        try:
            lines = handle.readlines()
        except (OSError, ValueError):
            return None, "Could not read the cursor file."
    for line in lines:
        cursor = line.strip()
        if not cursor:
            continue
        if CURSOR_PATTERN.fullmatch(cursor):
            return cursor, None
        return None, CURSOR_MESSAGE
    return None, None


def save_cursor(path, cursor):
    if not isinstance(cursor, str) or not cursor or "\n" in cursor or "\r" in cursor:
        return
    directory = os.path.dirname(os.path.abspath(path)) or "."
    os.makedirs(directory, exist_ok=True)
    fd, tmp = tempfile.mkstemp(prefix=".cursor-", dir=directory)
    try:
        with os.fdopen(fd, "w", encoding="utf-8") as handle:
            handle.write(cursor)
            handle.write("\n")
        os.replace(tmp, path)
    except Exception:
        try:
            os.unlink(tmp)
        except OSError:
            pass
        raise


def dry_run(path, cursor_path):
    seen = {}
    try:
        handle = open(path, encoding="utf-8")
    except OSError:
        print(f"Could not read {path}.", file=sys.stderr)
        return 1
    with handle:
        for raw in handle:
            line = raw.strip()
            if not line:
                continue
            try:
                record = json.loads(line)
            except json.JSONDecodeError:
                print("Skipped a line that is not JSON.", file=sys.stderr)
                continue
            events, cursor = events_from_record(record)
            if not deliver(events, seen, _print_event):
                return 1
            if cursor:
                save_cursor(cursor_path, cursor)
    return 0


def _print_event(event):
    print(payload_of(event))
    return True


def load_mqtt():
    try:
        import paho.mqtt.client as mqtt
    except ImportError:
        return None
    return mqtt


def make_client(mqtt):
    client_id = f"terminal-bridge-{os.getpid()}"
    if hasattr(mqtt, "CallbackAPIVersion"):
        return mqtt.Client(mqtt.CallbackAPIVersion.VERSION2, client_id=client_id)
    return mqtt.Client(client_id=client_id)


def mqtt_publish(client, topic, event):
    payload = payload_of(event)
    try:
        if not client.is_connected():
            client.reconnect()
        info = client.publish(topic, payload, qos=MQTT_QOS)
        info.wait_for_publish(timeout=30)
    except Exception:
        return False
    return bool(info.is_published())


def parse_port(raw):
    if raw is None or str(raw).strip() == "":
        return DEFAULT_MQTT_PORT, None
    text = str(raw).strip()
    if not text.isdigit():
        return None, "MQTT_PORT must be a whole number from 1 to 65535."
    port = int(text)
    if port < 1 or port > 65535:
        return None, "MQTT_PORT must be a whole number from 1 to 65535."
    return port, None


def prepare(config):
    if not config["key"]:
        return KEY_MESSAGE
    # A key is printable ASCII with no spaces. Anything else would make the
    # HTTP library raise an error that quotes the whole header.
    if any(ord(char) < 0x21 or ord(char) > 0x7E for char in config["key"]):
        return KEY_CHARS_MESSAGE
    base_error = check_base_url(config["base_url"])
    if base_error:
        return base_error
    if config["port_error"]:
        return config["port_error"]
    topic = config["topic"]
    if not topic or "{instance_id}" in topic or "{namespace}" in topic:
        return TOPIC_MESSAGE
    if "+" in topic or "#" in topic:
        return "MQTT_TOPIC must be a topic name, not a filter. Wildcards + and # are for subscribers."
    host = config["host"]
    if not host or "://" in host or "/" in host or "@" in host:
        return "MQTT_HOST must be a hostname, for example localhost."
    start = config["start"]
    if start not in ("latest", "earliest"):
        return "TERMINAL_START must be latest or earliest."
    return None


def backoff(failures):
    # 5, 10, 20, 40, then 60 seconds while the same trouble lasts.
    return min(MAX_BACKOFF, 5 * 2 ** max(0, failures - 1))


def poll_loop(config, client, fetch, sleep, publish=None, clock=time.monotonic):
    if publish is None:
        publish = mqtt_publish
    seen = {}
    cursor, problem = read_cursor(config["cursor_path"])
    if problem:
        print(problem, file=sys.stderr)
        return 1
    if cursor:
        print("Resuming from the cursor file.", file=sys.stderr)
    else:
        print("Starting from alerts recorded after this moment.", file=sys.stderr)
    print(
        f"Publishing events to {config['topic']} on {config['host']}:{config['port']}.",
        file=sys.stderr,
    )
    wait = LONG_POLL_SECONDS
    failures = 0
    while True:
        url = build_events_url(config["base_url"], cursor, wait, start=config["start"])
        started = clock()
        try:
            status, headers, raw = fetch(url, config["key"], wait + 20)
        except Exception as error:
            # Only the error's type: a message can quote the request, key included.
            failures += 1
            delay = backoff(failures)
            print(f"The events feed did not answer ({type(error).__name__}). Waiting {delay} seconds.", file=sys.stderr)
            sleep(delay)
            continue
        result = interpret_response(status, headers, raw)
        if result["action"] == "retry":
            failures += 1
            delay = result["retry_after"] if status == 429 else max(result["retry_after"], backoff(failures))
            print(f"Waiting {delay} seconds before asking again.", file=sys.stderr)
            sleep(delay)
            continue
        if result["action"] == "stop":
            print(result["message"], file=sys.stderr)
            return 1
        failures = 0
        # Keep the page while the broker is down: asking for it again would
        # count a call each time (an answer with events is a call).
        broker_failures = 0
        while not deliver(result["events"], seen, lambda event: publish(client, config["topic"], event)):
            broker_failures += 1
            delay = backoff(broker_failures)
            print(f"The broker did not take the event. Trying again in {delay} seconds.", file=sys.stderr)
            sleep(delay)
        if result["cursor"]:
            save_cursor(config["cursor_path"], result["cursor"])
            cursor = result["cursor"]
        if result["gap"]:
            print(
                "An alert after the cursor is older than 7 days and is no longer in the feed. "
                "Read GET /v1/rules/<rule id>/record for that rule.",
                file=sys.stderr,
            )
        if result["has_more"]:
            wait = 0
            continue
        wait = LONG_POLL_SECONDS
        if not result["events"] and clock() - started < FAST_EMPTY_SECONDS:
            # A long-poll that came back empty at once (a server shutting down)
            # would otherwise be asked again in a tight loop.
            sleep(1)


def config_from_env(cursor_path):
    port, port_error = parse_port(os.environ.get("MQTT_PORT"))
    start = os.environ.get("TERMINAL_START", "latest").strip() or "latest"
    return {
        "key": os.environ.get("TERMINAL_API_KEY", "").strip(),
        "base_url": os.environ.get("TERMINAL_API_URL", DEFAULT_API_URL).strip() or DEFAULT_API_URL,
        "host": os.environ.get("MQTT_HOST", DEFAULT_MQTT_HOST).strip() or DEFAULT_MQTT_HOST,
        "port": port if port is not None else 0,
        "port_error": port_error,
        "topic": os.environ.get("MQTT_TOPIC", DEFAULT_MQTT_TOPIC).strip() or DEFAULT_MQTT_TOPIC,
        "cursor_path": cursor_path,
        "username": os.environ.get("MQTT_USERNAME", ""),
        "password": os.environ.get("MQTT_PASSWORD", ""),
        "start": start,
    }


def run_bridge(config, fetch=http_get, sleep=time.sleep):
    problem = prepare(config)
    if problem:
        print(problem, file=sys.stderr)
        return 1
    mqtt = load_mqtt()
    if mqtt is None:
        print(PAHO_MESSAGE, file=sys.stderr)
        return 1
    client = make_client(mqtt)
    if config["username"] or config["password"]:
        client.username_pw_set(config["username"], config["password"] or None)
    try:
        client.connect(config["host"], config["port"], keepalive=60)
    except Exception:
        print(
            f"Could not connect to the MQTT broker at {config['host']}:{config['port']}.",
            file=sys.stderr,
        )
        return 1
    client.loop_start()
    try:
        return poll_loop(config, client, fetch, sleep)
    finally:
        client.loop_stop()
        try:
            client.disconnect()
        except Exception:
            pass


def main(argv=None):
    parser = argparse.ArgumentParser(
        description="Publish Terminal alert events to a Hummingbot MQTT topic."
    )
    parser.add_argument(
        "--dry-run",
        metavar="FILE",
        help="Read a JSONL file and print each event that would be published. No network and no MQTT.",
    )
    parser.add_argument(
        "--cursor-file",
        metavar="FILE",
        help="Where to store the feed cursor. Overrides TERMINAL_CURSOR_FILE.",
    )
    args = parser.parse_args(argv)
    cursor_path = args.cursor_file or os.environ.get("TERMINAL_CURSOR_FILE") or default_cursor_path()
    if args.dry_run:
        return dry_run(args.dry_run, cursor_path)
    try:
        return run_bridge(config_from_env(cursor_path))
    except KeyboardInterrupt:
        return 0


if __name__ == "__main__":
    sys.exit(main())