#!/usr/bin/env python3
"""Polymarket BTC 图表本地服务。

一个进程做三件事：
  1. 静态文件服务（charts 目录，替代 python -m http.server）
  2. 后台常驻订阅 Chainlink RTDS（crypto_prices_chainlink / btc/usd），秒级落盘
  3. 提供 /api/cl?hours=N 给页面读历史，打开即有完整 TWAP

用法:
  python serve.py                 # 默认 8900，代理 127.0.0.1:10808
  python serve.py --port 8900 --proxy 127.0.0.1:10808
  python serve.py --no-proxy      # 境外机器直连
  python serve.py --no-recorder   # 采集器由独立进程/单元监管（生产用）

数据文件: data/chainlink_btc_YYYYMMDD.ndjson（每行 {"t": 秒, "v": 价格}）
"""
from __future__ import annotations

import argparse
import atexit
import hashlib
import json
import os
import re
import shutil
import subprocess
import threading
import time
import urllib.error
import urllib.request
from datetime import datetime, timezone
from http.server import SimpleHTTPRequestHandler, ThreadingHTTPServer
from urllib.parse import parse_qs, quote, urlparse

ROOT = os.path.dirname(os.path.abspath(__file__))
DATA_DIR = os.path.join(ROOT, "data")

RTDS_URL = "wss://ws-live-data.polymarket.com"
CL_SYMBOL = "btc/usd"
SUBSCRIBE = {
    "action": "subscribe",
    "subscriptions": [{
        "topic": "crypto_prices_chainlink",
        "type": "update",
        "filters": json.dumps({"symbol": CL_SYMBOL}),
    }],
}

PROXY: tuple[str | None, int | None] = (None, None)
_write_lock = threading.Lock()

# ---------- 上游代理缓存（Polymarket 成交 / 市场元数据） ----------
# 分级依据是「数据可变性」，不是统一 TTL：
#   · 已结束市场的逐笔成交、已结算市场的 gamma 元数据 ⇒ 不可变 ⇒ 永久缓存
#   · 未结束市场 ⇒ 必须新鲜 ⇒ 不缓存（或极短 TTL），直通上游
#   · Chainlink / OKX 两地是**追加型原始数据**，落盘在 data/（不是 cache/）
CACHE_DIR = os.path.join(ROOT, "cache")
CACHE_MAX_AGE_D = 7.0     # 缓存最长保留天数（只清 cache/，绝不碰 data/）
UP_TRADES = "https://data-api.polymarket.com/v2/trades"
UP_GAMMA = "https://gamma-api.polymarket.com/events"
CLOSED_GRACE_S = 120      # 市场结束后再等 120s 才视为"成交已定型"
META_LIVE_TTL_S = 20      # 未结算市场元数据的短缓存
_cache_lock = threading.Lock()


def _opener():
    if PROXY[0]:
        px = f"http://{PROXY[0]}:{PROXY[1]}"
        return urllib.request.build_opener(
            urllib.request.ProxyHandler({"http": px, "https": px})
        )
    return urllib.request.build_opener()


def fetch_upstream(url: str, timeout: float = 20.0) -> tuple[int, bytes]:
    req = urllib.request.Request(url, headers={"User-Agent": "polymarket-5m-chart/1.0"})
    try:
        with _opener().open(req, timeout=timeout) as r:
            return int(r.status), r.read()
    except urllib.error.HTTPError as e:
        return int(e.code), e.read()
    except Exception as e:  # 超时 / DNS / 连接重置
        log("upstream error:", repr(e)[:120])
        return 0, b""


def _cache_path(kind: str, key: str) -> str:
    d = os.path.join(CACHE_DIR, kind)
    name = hashlib.sha1(key.encode("utf-8")).hexdigest()[:28] + ".json"
    return os.path.join(d, name)


def cache_read(kind: str, key: str, ttl: float | None = None) -> bytes | None:
    p = _cache_path(kind, key)
    try:
        if ttl is not None and time.time() - os.path.getmtime(p) > ttl:
            return None
        with open(p, "rb") as f:
            return f.read()
    except OSError:
        return None


def cache_write(kind: str, key: str, body: bytes) -> None:
    p = _cache_path(kind, key)
    try:
        os.makedirs(os.path.dirname(p), exist_ok=True)
        tmp = p + ".tmp"
        with _cache_lock:
            with open(tmp, "wb") as f:
                f.write(body)
            os.replace(tmp, p)
    except OSError as e:
        log("cache write fail:", repr(e)[:80])


def prune_cache(max_age_d: float = CACHE_MAX_AGE_D) -> int:
    """删除超龄缓存文件。cache/ 是派生数据，与 data/ 的原始数据严格分开。"""
    if not os.path.isdir(CACHE_DIR):
        return 0
    cutoff = time.time() - max_age_d * 86400
    n = 0
    for kind in os.listdir(CACHE_DIR):
        kd = os.path.join(CACHE_DIR, kind)
        if not os.path.isdir(kd):
            continue
        for fn in os.listdir(kd):
            if not fn.endswith(".json"):
                continue
            p = os.path.join(kd, fn)
            try:
                if os.path.getmtime(p) < cutoff:
                    os.remove(p)
                    n += 1
            except OSError:
                pass
    return n


def log(*a):
    print(datetime.now().strftime("%H:%M:%S"), *a, flush=True)


def day_path(ts: int) -> str:
    d = datetime.fromtimestamp(ts, tz=timezone.utc).strftime("%Y%m%d")
    return os.path.join(DATA_DIR, f"chainlink_btc_{d}.ndjson")


def okx_day_path(ts: int) -> str:
    d = datetime.fromtimestamp(ts, tz=timezone.utc).strftime("%Y%m%d")
    return os.path.join(DATA_DIR, f"okx_btc_{d}.ndjson")


def load_okx(since_s: int, until_s: int) -> list[list[float]]:
    """读 OKX 成交历史（recorder.js 落盘，毫秒 ts），按 (t) 去重升序。"""
    out: dict[int, float] = {}
    day = since_s - since_s % 86400
    while day <= until_s:
        path = okx_day_path(day + 1)
        if os.path.exists(path):
            try:
                with open(path, encoding="utf-8") as f:
                    for line in f:
                        line = line.strip()
                        if not line:
                            continue
                        try:
                            r = json.loads(line)
                            t = int(r["t"])
                        except Exception:
                            continue
                        if since_s * 1000 <= t <= until_s * 1000:
                            out[t] = float(r["p"])
            except OSError:
                pass
        day += 86400
    return [[t, out[t]] for t in sorted(out)]


def append_rows(rows: list[tuple[int, float]]) -> None:
    """按天追加写；同一秒重复写入由读端去重。"""
    os.makedirs(DATA_DIR, exist_ok=True)
    with _write_lock:
        buckets: dict[str, list[str]] = {}
        for t, v in rows:
            line = '{"t":%d,"v":%s}' % (t, repr(round(v, 6)))
            buckets.setdefault(day_path(t), []).append(line)
        for path, lines in buckets.items():
            with open(path, "a", encoding="utf-8") as f:
                f.write("\n".join(lines) + "\n")


def start_recorder(proxy) -> subprocess.Popen | None:
    """拉起 Node 采集器（Python websocket-client 只能收到首包，实际不推送）。"""
    node = shutil.which("node")
    if not node:
        log("!! node not found, recorder disabled (page will rely on live WS only)")
        return None
    env = dict(os.environ)
    if proxy and proxy[0]:
        env["NODE_USE_ENV_PROXY"] = "1"
        env["https_proxy"] = f"http://{proxy[0]}:{proxy[1]}"
        env["http_proxy"] = env["https_proxy"]
    p = subprocess.Popen([node, os.path.join(ROOT, "recorder.js")], env=env, cwd=ROOT)
    log(f"recorder started (pid {p.pid})")
    atexit.register(lambda: p.terminate())
    return p


def recorder_loop() -> None:
    import websocket  # websocket-client

    backoff = 3
    while True:
        try:
            ws = websocket.WebSocket()
            kw = {"timeout": 40}
            if PROXY[0]:
                kw.update(http_proxy_host=PROXY[0], http_proxy_port=PROXY[1])
            ws.connect(RTDS_URL, **kw)
            ws.send(json.dumps(SUBSCRIBE))
            log("RTDS connected")
            backoff = 3
            pending: list[tuple[int, float]] = []
            last_flush = time.time()
            while True:
                raw = ws.recv()
                if not raw:
                    continue
                try:
                    msg = json.loads(raw)
                except Exception:
                    continue
                if msg.get("topic") != "crypto_prices_chainlink":
                    continue
                p = msg.get("payload") or {}
                rows = p.get("data") if isinstance(p.get("data"), list) else None
                if rows is None and isinstance(p.get("value"), (int, float)) and p.get("timestamp"):
                    rows = [p]
                if not rows:
                    continue
                for r in rows:
                    try:
                        pending.append((int(int(r["timestamp"]) // 1000), float(r["value"])))
                    except Exception:
                        pass
                if len(pending) >= 30 or (time.time() - last_flush) > 5:
                    if pending:
                        append_rows(pending)
                        pending.clear()
                    last_flush = time.time()
        except Exception as e:
            log("RTDS error:", repr(e)[:120], f"retry in {backoff}s")
            time.sleep(backoff)
            backoff = min(backoff * 2, 30)


def load_recent(hours: float) -> list[list[float]]:
    """读最近 N 小时的秒级数据，按秒去重升序。"""
    now = int(time.time())
    start = now - int(hours * 3600)
    out: dict[int, float] = {}
    day = start - start % 86400
    while day <= now:
        path = day_path(day + 1)
        if os.path.exists(path):
            try:
                with open(path, encoding="utf-8") as f:
                    for line in f:
                        line = line.strip()
                        if not line:
                            continue
                        try:
                            r = json.loads(line)
                        except Exception:
                            continue
                        t = int(r["t"])
                        if t >= start:
                            out[t] = float(r["v"])
            except OSError:
                pass
        day += 86400
    return [[t, out[t]] for t in sorted(out)]


class Handler(SimpleHTTPRequestHandler):
    def __init__(self, *a, **kw):
        super().__init__(*a, directory=ROOT, **kw)

    def end_headers(self):
        self.send_header("Access-Control-Allow-Origin", "*")
        self.send_header("Cache-Control", "no-store")
        super().end_headers()

    def _send_json(self, body: bytes, status: int = 200) -> None:
        self.send_response(status)
        self.send_header("Content-Type", "application/json")
        self.send_header("Content-Length", str(len(body)))
        self.end_headers()
        self.wfile.write(body)

    def serve_trades(self, query: str) -> bytes | None:
        """逐笔成交：已定型（end < now-120s）的窗口永久缓存，当前窗口直通。"""
        q = parse_qs(query)
        try:
            end = int(float(q.get("end", [0])[0]))
        except ValueError:
            end = 0
        settled = end and end <= int(time.time()) - CLOSED_GRACE_S
        url = UP_TRADES + ("?" + query if query else "")
        if settled:
            hit = cache_read("trades_final", url)
            if hit is not None:
                self._send_json(hit)
                return hit
        st, body = fetch_upstream(url)
        if st == 200 and settled:
            cache_write("trades_final", url, body)
        if st == 0:
            return None
        self._send_json(body, st)
        return body

    def serve_meta(self, query: str) -> bytes | None:
        """市场元数据：已结算永久缓存；未结算短 TTL（避免刷新时重复打 gamma）。"""
        q = parse_qs(query)
        slug = (q.get("slug") or [""])[0]
        if not re.fullmatch(r"[A-Za-z0-9\-]{1,80}", slug):
            self._send_json(b'{"error":"bad slug"}', 400)
            return b"{}"
        key = "slug:" + slug
        hit = cache_read("meta_final", key)
        if hit is not None:
            self._send_json(hit)
            return hit
        hit = cache_read("meta_live", key, META_LIVE_TTL_S)
        if hit is not None:
            self._send_json(hit)
            return hit
        st, body = fetch_upstream(f"{UP_GAMMA}?slug={quote(slug)}")
        if st == 0:
            return None
        if st == 200:
            compact = body.replace(b" ", b"")
            if b'"closed":true' in compact:
                cache_write("meta_final", key, body)
            else:
                cache_write("meta_live", key, body)
        self._send_json(body, st)
        return body

    def do_GET(self):
        u = urlparse(self.path)
        if u.path == "/api/cl":
            q = parse_qs(u.query)
            try:
                hours = float(q.get("hours", ["2"])[0])
            except ValueError:
                hours = 2.0
            hours = max(0.1, min(hours, 48.0))
            body = json.dumps(load_recent(hours), separators=(",", ":")).encode()
            self.send_response(200)
            self.send_header("Content-Type", "application/json")
            self.send_header("Content-Length", str(len(body)))
            self.end_headers()
            self.wfile.write(body)
            return
        if u.path == "/api/trades":
            body = self.serve_trades(u.query)
            if body is not None:
                return
            self.send_error(502, "upstream unavailable")
            return
        if u.path == "/api/meta":
            body = self.serve_meta(u.query)
            if body is not None:
                return
            self.send_error(502, "upstream unavailable")
            return
        if u.path == "/api/cache/prune":
            q = parse_qs(u.query)
            try:
                days = float(q.get("days", [CACHE_MAX_AGE_D])[0])
            except ValueError:
                days = CACHE_MAX_AGE_D
            days = max(0.1, min(days, 365.0))
            n = prune_cache(days)
            body = json.dumps({"removed": n, "max_age_days": days}).encode()
            self.send_response(200)
            self.send_header("Content-Type", "application/json")
            self.send_header("Content-Length", str(len(body)))
            self.end_headers()
            self.wfile.write(body)
            return
        if u.path == "/api/cache/status":
            payload = {"dir": CACHE_DIR, "kinds": {}}
            if os.path.isdir(CACHE_DIR):
                for k in sorted(os.listdir(CACHE_DIR)):
                    kd = os.path.join(CACHE_DIR, k)
                    if not os.path.isdir(kd):
                        continue
                    n = 0
                    size = 0
                    for fn in os.listdir(kd):
                        try:
                            size += os.path.getsize(os.path.join(kd, fn))
                            n += 1
                        except OSError:
                            pass
                    payload["kinds"][k] = {"files": n, "bytes": size}
            body = json.dumps(payload).encode()
            self.send_response(200)
            self.send_header("Content-Type", "application/json")
            self.send_header("Content-Length", str(len(body)))
            self.end_headers()
            self.wfile.write(body)
            return
        if u.path == "/api/okx":
            q = parse_qs(u.query)
            now_s = int(time.time())
            try:
                until = int(float(q.get("until", [now_s])[0]))
            except ValueError:
                until = now_s
            try:
                since = int(float(q.get("since", [until - 3600])[0]))
            except ValueError:
                since = until - 3600
            until = min(until, now_s)
            since = max(since, until - 6 * 3600)   # 单次最多 6 小时，防大包
            since = min(since, until)
            rows = load_okx(since, until)
            body = json.dumps(rows, separators=(",", ":")).encode()
            self.send_response(200)
            self.send_header("Content-Type", "application/json")
            self.send_header("Content-Length", str(len(body)))
            self.end_headers()
            self.wfile.write(body)
            return
        if u.path == "/api/cl/status":
            body = json.dumps({"ok": True, "files": sorted(os.listdir(DATA_DIR)) if os.path.isdir(DATA_DIR) else []}).encode()
            self.send_response(200)
            self.send_header("Content-Type", "application/json")
            self.send_header("Content-Length", str(len(body)))
            self.end_headers()
            self.wfile.write(body)
            return
        super().do_GET()

    def log_message(self, fmt, *args):
        # 404 走 send_error → log_error("code %d, message %s", code, msg)，
        # 此时 args[0] 是 HTTPStatus 枚举，非字符串；直接 in 判断会抛 TypeError，
        # 把错误响应本身打断（客户端拿到空回复，实测 curl exit 52）。
        msg = args[0] if args else ""
        if isinstance(msg, str) and "/api/cl" in msg:
            return
        super().log_message(fmt, *args)


def main() -> None:
    global PROXY
    ap = argparse.ArgumentParser()
    ap.add_argument("--port", type=int, default=8900)
    ap.add_argument("--host", default="127.0.0.1")
    ap.add_argument("--proxy", default=os.environ.get("HTTPS_PROXY") or "http://127.0.0.1:10808")
    ap.add_argument("--no-proxy", action="store_true")
    ap.add_argument("--no-recorder", action="store_true",
                    help="不自启 recorder.js（交给外部监管，如独立 systemd 单元）")
    args = ap.parse_args()

    if not args.no_proxy and args.proxy:
        p = urlparse(args.proxy if "://" in args.proxy else "http://" + args.proxy)
        PROXY = (p.hostname, p.port or 8080)

    os.makedirs(DATA_DIR, exist_ok=True)
    removed = prune_cache()      # 启动时清一次超龄缓存（只动 cache/）
    if removed:
        log(f"cache pruned: {removed} file(s) older than {CACHE_MAX_AGE_D:g}d")
    if not args.no_recorder:
        start_recorder(PROXY)

    srv = ThreadingHTTPServer((args.host, args.port), Handler)
    log(f"serving http://{args.host}:{args.port}/  (data dir: {DATA_DIR}, proxy: {PROXY})")
    srv.serve_forever()


if __name__ == "__main__":
    main()