SimpleFarm/farm/client.py

"""远程农场客户端：让本地代码（TUI、脚本、机器人）像用本地引擎一样用服务器。

设计目标
--------
TUI 原本直接拿着 ``FarmEngine`` 对象，现在游戏跑在服务器进程里，客户端必须
换成"远程把手"。为了不把上千行渲染代码推倒重写，这里做了一个**接口同构**的
:class:`RemoteFarm`：

    本地引擎                     远程客户端
    ----------                   ----------
    engine.state.plots[0].is_ripe   state.plots[0].is_ripe
    engine.harvest(1)               harvest(1)
    engine.catalog()                catalog()
    engine.recent_log(5)            recent_log(5)

字段名和方法名都对齐，所以 :mod:`farm.tui` 不需要知道自己在跟谁说话。

读写分工
--------
* **写操作走 HTTP**（POST /api/actions）：语义清晰、可重试、失败有明确错误码。
* **读数走 SSE**（GET /api/stream）：服务器在状态变化时顺着一条 HTTP 长连接推过来，
  界面不用轮询；上行下行都是普通 HTTP，不需要任何第三方库。
* SSE 不可用时自动降级成"HTTP 轮询 + 写操作"，功能不缺，只是延迟高一点。

请求库用的是标准库 ``urllib``，所以即使机器上没装任何第三方包，TUI 也能跑。
"""

from __future__ import annotations

import contextlib
import json
import socket
import threading
import time
import urllib.error
import urllib.parse
import urllib.request
from collections import deque
from collections.abc import Callable
from typing import Any

from . import config, net
from .engine import GameError

DEFAULT_ENDPOINT = "http://127.0.0.1:8765"
LOG_KEEP = 200
STREAM_RECONNECT_DELAY = 3.0  # SSE 断开后重连的间隔（秒）
STREAM_READ_TIMEOUT = 40.0  # 读 SSE 的 socket 超时，要比服务器心跳（15s）宽裕


def _underlying_socket(response: Any) -> Any:
    """从 ``HTTPResponse`` 里掏出底层 socket。

    用途只有一个：断开时能立刻中断正在阻塞的读取（``close()`` 会等读线程的锁，
    最长要等到下一次心跳，不可接受）。掏不到就返回 None，调用方会退化成 close()。
    """
    fp = getattr(response, "fp", None)
    raw = getattr(fp, "raw", None)  # http.client 把 socket 包成 SocketIO
    return getattr(raw, "_sock", None)


# --------------------------------------------------------------------- 状态视图
class PlotView:
    """一块地的只读视图（字段名与 :class:`farm.models.Plot` 对齐）。"""

    __slots__ = ("_d",)

    def __init__(self, data: dict[str, Any]) -> None:
        self._d = data

    @property
    def index(self) -> int:
        return int(self._d.get("index", 0))

    @property
    def position(self) -> int:
        return int(self._d.get("position", self.index + 1))

    @property
    def crop(self) -> str | None:
        return self._d.get("crop")

    @property
    def crop_name(self) -> str:
        return self._d.get("crop_name") or ""

    @property
    def growth(self) -> float:
        return float(self._d.get("growth", 0.0))

    @property
    def grow_days(self) -> int:
        return int(self._d.get("grow_days", 0))

    @property
    def days_left(self) -> int:
        return int(self._d.get("days_left", 0))

    @property
    def moisture(self) -> int:
        return int(self._d.get("moisture", 0))

    @property
    def is_ripe(self) -> bool:
        return bool(self._d.get("is_ripe"))

    @property
    def is_empty(self) -> bool:
        return bool(self._d.get("is_empty", self.crop is None))

    @property
    def stage(self) -> str:
        return self._d.get("stage", "")

    def to_dict(self) -> dict[str, Any]:
        return dict(self._d)


class AnimalView:
    """一只牲畜的只读视图（``aid`` 对齐本地模型的 ``.aid``）。"""

    __slots__ = ("_d",)

    def __init__(self, data: dict[str, Any]) -> None:
        self._d = data

    @property
    def aid(self) -> int:
        return int(self._d.get("id", 0))

    @property
    def kind(self) -> str:
        return self._d.get("kind", "")

    @property
    def name(self) -> str:
        return self._d.get("name", "")

    @property
    def product(self) -> str:
        return self._d.get("product", "")

    @property
    def fed(self) -> bool:
        return bool(self._d.get("fed"))

    @property
    def pending(self) -> int:
        return int(self._d.get("pending", 0))

    @property
    def cycle(self) -> int:
        return int(self._d.get("cycle", 1))

    @property
    def progress(self) -> int:
        return int(self._d.get("progress", 0))

    @property
    def starve(self) -> int:
        return int(self._d.get("starve", 0))

    @property
    def days_to_product(self) -> int:
        return int(self._d.get("days_to_product", 0))

    @property
    def is_sick(self) -> bool:
        return bool(self._d.get("is_sick"))

    def to_dict(self) -> dict[str, Any]:
        return dict(self._d)


class StateView:
    """整局游戏的只读视图。刻意保留 ``to_dict()``，方便直接落盘或转发。"""

    def __init__(self, data: dict[str, Any] | None = None) -> None:
        self._d: dict[str, Any] = {}
        # 空视图也要长得像"空农场"：还没收到第一份快照时（或服务器连不上时），
        # 上层会直接读 state.plots 渲染，不能甩出 AttributeError。
        self.plots: list[PlotView] = []
        self.animals: list[AnimalView] = []
        self.apply(data or {})

    def apply(self, data: dict[str, Any]) -> None:
        """用新数据刷新内部视图。

        服务器会发两种东西：**完整快照**（带 plots/animals）和**精简状态**
        （动作回执里的 state，只有几个标量）。前者整体替换，后者只做字段合并——
        否则一次播种就会把 TUI 的农田格子清空。
        """
        if not data:
            return
        if "plots" in data:
            merged = dict(data)
        else:
            merged = dict(self._d)
            merged.update(data)
        self._d = merged
        self.plots = [PlotView(p) for p in self._d.get("plots", []) if isinstance(p, dict)]
        self.animals = [AnimalView(a) for a in self._d.get("animals", []) if isinstance(a, dict)]

    # 常用标量字段
    @property
    def day(self) -> int:
        return int(self._d.get("day", 1))

    @property
    def gold(self) -> int:
        return int(self._d.get("gold", 0))

    @property
    def level(self) -> int:
        return int(self._d.get("level", 1))

    @property
    def exp(self) -> int:
        return int(self._d.get("exp", 0))

    @property
    def feed(self) -> int:
        return int(self._d.get("feed", 0))

    @property
    def market(self) -> dict[str, float]:
        return self._d.get("market", {})

    @property
    def weather(self) -> str:
        return self._d.get("weather", config.DEFAULT_WEATHER)

    @property
    def weather_name(self) -> str:
        return self._d.get("weather_name") or config.weather_info(self.weather)["name"]

    @property
    def forecast(self) -> str:
        return self._d.get("forecast", config.DEFAULT_WEATHER)

    @property
    def forecast_name(self) -> str:
        return self._d.get("forecast_name") or config.weather_info(self.forecast)["name"]

    @property
    def inventory(self) -> dict[str, int]:
        return self._d.get("inventory", {})

    @property
    def storage_used(self) -> int:
        return int(self._d.get("storage_used", 0))

    @property
    def granary_capacity(self) -> int:
        return int(self._d.get("granary_capacity", config.GRANARY_BASE))

    @property
    def buildings(self) -> dict[str, int]:
        return self._d.get("buildings", {})

    def get(self, key: str, default: Any = None) -> Any:
        return self._d.get(key, default)

    def __getitem__(self, key: str) -> Any:
        return self._d[key]

    def to_dict(self) -> dict[str, Any]:
        return dict(self._d)


# --------------------------------------------------------------------- 客户端
class RemoteFarm:
    """远程农场的句柄：接口与 :class:`farm.engine.FarmEngine` 保持一致。"""

    def __init__(
        self,
        endpoint: str = DEFAULT_ENDPOINT,
        farm: str = "default",
        token: str | None = None,
        name: str = "",
        timeout: float = 6.0,
        realtime: bool = True,
        autostart: bool = True,
    ) -> None:
        self.endpoint = self._normalize(endpoint)
        self.farm = farm or "default"
        self.token = token or ""
        self.name = name
        self.timeout = timeout
        self.realtime_enabled = realtime

        self.state = StateView()
        self.connected = False
        self.mode = "http"  # http（只有写请求）/ sse（长连接推送）/ poll（轮询兜底）
        self.last_error = ""
        self.last_message = ""
        self.server_info: dict[str, Any] = {}
        self.on_change: Callable[[], None] | None = None

        self._lock = threading.RLock()
        self._stop = threading.Event()
        self._thread: threading.Thread | None = None
        self._stream_response: Any = None
        self._stream_socket: Any = None
        self._catalog: dict[str, Any] = {
            "crops": [],
            "animals": [],
            "goods": [],
            "buildings": [],
            "weather": {},
            "granary": {},
            "feed_price": config.FEED_UNIT_PRICE,
            "water_capacity": config.WATER_CAPACITY,
        }
        self._log: deque[dict[str, Any]] = deque(maxlen=LOG_KEEP)
        self._log_seq = 0
        self.stats = {"http": 0, "pushes": 0, "errors": 0}

        if autostart:
            self.refresh(raise_on_error=True)
            if realtime:
                self.start()

    # ------------------------------------------------------------ 基础设施
    @staticmethod
    def _normalize(endpoint: str) -> str:
        endpoint = (endpoint or "").strip()
        if endpoint.startswith("tcp://"):
            endpoint = "http://" + endpoint[len("tcp://") :]
        elif not endpoint.startswith(("http://", "https://")):
            endpoint = "http://" + endpoint
        # 补完 scheme 再削尾部斜杠：顺序反了的话 "http://" 会被削成 "http:"，
        # 拼出来的地址没有主机名，请求会悄悄落到系统代理上（表现为莫名其妙的 502）。
        endpoint = endpoint.rstrip("/")
        has_host = urllib.parse.urlsplit(endpoint).netloc
        if endpoint.startswith(("http://", "https://")) and has_host:
            return endpoint
        return DEFAULT_ENDPOINT

    @property
    def host(self) -> str:
        return urllib.parse.urlsplit(self.endpoint).netloc or "127.0.0.1:8765"

    def url(self, path: str, **params: Any) -> str:
        query = {"farm": self.farm}
        if self.token:
            query["token"] = self.token
        query.update({k: v for k, v in params.items() if v is not None})
        return f"{self.endpoint}{path}?{urllib.parse.urlencode(query)}"

    def _http(
        self, method: str, path: str, payload: dict[str, Any] | None = None, **params: Any
    ) -> dict[str, Any]:
        url = self.url(path, **params)
        data = None
        headers = {"Accept": "application/json"}
        if payload is not None:
            data = json.dumps(payload, ensure_ascii=False).encode("utf-8")
            headers["Content-Type"] = "application/json"
        request = urllib.request.Request(url, data=data, headers=headers, method=method)
        self.stats["http"] += 1
        try:
            # 走 net.open_url：本地服务器不会被系统代理截走
            with net.open_url(request, timeout=self.timeout) as response:
                body = response.read().decode("utf-8", errors="replace")
                status = response.status
        except urllib.error.HTTPError as exc:
            body = exc.read().decode("utf-8", errors="replace")
            status = exc.code
        except (urllib.error.URLError, OSError, ValueError) as exc:
            self.stats["errors"] += 1
            self.connected = False
            raise GameError(
                "network",
                f"连不上服务器 {self.endpoint}（{exc}）。请先启动：uv run main.py serve --daemon",
                status=503,
            ) from exc
        try:
            data_out = json.loads(body) if body else {}
        except json.JSONDecodeError as exc:
            raise GameError(
                "bad_response", f"服务器返回了非 JSON 内容（HTTP {status}）", 502
            ) from exc
        if not isinstance(data_out, dict):
            raise GameError("bad_response", "服务器返回的 JSON 不是对象", 502)
        if not data_out.get("ok", True):
            self.stats["errors"] += 1
            raise GameError(
                data_out.get("error", "server_error"),
                data_out.get("message", "服务器拒绝了这次操作"),
                status=status if status >= 400 else 400,
            )
        self.connected = True
        self.last_error = ""
        return data_out

    # ------------------------------------------------------------ 读接口
    def refresh(self, raise_on_error: bool = False) -> bool:
        """主动拉一次全量数据（首次连接、以及 SSE 断开后的兜底）。"""
        try:
            data = self._http("GET", "/api/state")
            self._apply_online(data.get("state") or {}, data.get("online"))
            catalog = self._http("GET", "/api/catalog")
            self._catalog = catalog.get("catalog") or self._catalog
            logs = self._http("GET", "/api/log", limit=30)
            self._merge_log(logs.get("entries") or [])
            return True
        except GameError as exc:
            self.connected = False
            self.last_error = exc.message
            if raise_on_error:
                raise
            return False

    def _apply_online(self, state: dict[str, Any], online: Any) -> None:
        """把"在线人数"合进状态视图：TUI 头部和网页都要显示它。"""
        if state and isinstance(online, int):
            state = dict(state)
            state["online"] = online
        self._apply(state)

    def snapshot(self) -> dict[str, Any]:
        """本地缓存的状态视图，**不会发请求**。

        TUI 每帧都要读它，所以刻意做成纯本地读取；代价是它只反映"上一次同步"
        的结果。开实时通道（``realtime=True``，TUI / 网页走这条）时 SSE 会持续
        刷新它；关掉实时通道的脚本要自己调 :meth:`refresh`，否则轮询到的
        永远是同一份旧数据——一天天推进的服务器看起来就像卡住了。
        """
        return self.state.to_dict()

    def catalog(self) -> dict[str, Any]:
        return self._catalog

    def recent_log(self, limit: int = 12, since: int = 0) -> list[dict[str, Any]]:
        with self._lock:
            entries = [e for e in self._log if int(e.get("seq", 0)) > since]
        return entries[-limit:] if limit > 0 else entries

    def available_actions(self) -> list[dict[str, Any]]:
        data = self._http("GET", "/api/actions")
        self._apply(data.get("state") or {})
        return data.get("actions") or []

    def leaderboard(self) -> list[dict[str, Any]]:
        return self._http("GET", "/api/farms").get("farms") or []

    def world(self) -> dict[str, Any]:
        return self._http("GET", "/api/world")

    def help_text(self) -> str:
        return self._http("GET", "/api/help").get("rules", "")

    def _apply(self, state: dict[str, Any]) -> None:
        if not state:
            return
        self.state.apply(state)
        if self.on_change is not None:
            # UI 回调抛异常不能影响数据同步
            with contextlib.suppress(Exception):
                self.on_change()

    def _merge_log(self, entries: list[dict[str, Any]]) -> None:
        with self._lock:
            known = {int(e.get("seq", 0)) for e in self._log}
            for entry in entries:
                seq = int(entry.get("seq", 0))
                if seq and seq not in known:
                    self._log.append(entry)
                    known.add(seq)
                    self._log_seq = max(self._log_seq, seq)
            self._log = deque(
                sorted(self._log, key=lambda e: int(e.get("seq", 0))), maxlen=LOG_KEEP
            )

    # ------------------------------------------------------------ 写接口
    def _action(self, action: str, params: dict[str, Any] | None = None) -> dict[str, Any]:
        """执行一个动作：写操作统一走 HTTP POST /api/actions。

        实时通道只负责下行推送，所以这里不需要"优先哪条通道"的判断；
        ``full=1`` 让服务器顺手把完整快照带回来，省掉"操作后再查一次"。
        """
        data = self._http(
            "POST", "/api/actions", {"action": action, "params": params or {}}, full=1
        )
        self._apply(data.get("snapshot") or data.get("state") or {})
        self._merge_log(data.get("log") or [])
        return data

    # ---- 与本地引擎同名的方法
    def plant(self, plot: Any, crop: Any) -> dict[str, Any]:
        return self._action("plant", {"plot": plot, "crop": crop})

    def water(self, plot: Any) -> dict[str, Any]:
        return self._action("water", {"plot": plot})

    def harvest(self, plot: Any) -> dict[str, Any]:
        return self._action("harvest", {"plot": plot})

    def harvest_all(self) -> dict[str, Any]:
        return self._action("harvest_all")

    def buy_animal(self, kind: Any) -> dict[str, Any]:
        return self._action("buy_animal", {"kind": kind})

    def feed(self, animal_id: Any, quantity: Any = None) -> dict[str, Any]:
        params: dict[str, Any] = {"animal_id": animal_id}
        if quantity is not None:
            params["quantity"] = quantity
        return self._action("feed", params)

    def collect(self, animal_id: Any) -> dict[str, Any]:
        return self._action("collect", {"animal_id": animal_id})

    def buy_feed(self, quantity: Any = config.FEED_BULK) -> dict[str, Any]:
        return self._action("buy_feed", {"quantity": quantity})

    def buy_plot(self) -> dict[str, Any]:
        return self._action("buy_plot")

    def sell(self, key: Any = None, quantity: Any = None) -> dict[str, Any]:
        """卖出谷仓存货；两个参数都省略就是清仓。"""
        params: dict[str, Any] = {}
        if key is not None:
            params["key"] = key
        if quantity is not None:
            params["quantity"] = quantity
        return self._action("sell", params)

    def build(self, key: Any) -> dict[str, Any]:
        return self._action("build", {"key": key})

    def pause(self, paused: bool | None = None) -> dict[str, Any]:
        # 动作回执里的 state（brief）已带 paused 与 seconds_to_tick，
        # _action 内部会合并进本地视图，这里只需转发参数
        params: dict[str, Any] = {}
        if paused is not None:
            params["paused"] = paused
        return self._action("pause", params)

    def reset(self) -> dict[str, Any]:
        data = self._action("reset")
        with self._lock:
            self._log.clear()
            self._log_seq = 0
        self.refresh()
        return data

    def save(self, path: str | None = None) -> dict[str, Any]:
        """远程存档：位置由服务器决定，``path`` 参数仅为与本地引擎保持同签名。"""
        return self._action("save")

    def batch(self, items: list[dict[str, Any]]) -> dict[str, Any]:
        """批量执行（一个请求干多件事），给机器人/AI 用。"""
        data = self._http("POST", "/api/batch", {"actions": items}, full=1)
        self._apply(data.get("snapshot") or data.get("state") or {})
        self._merge_log(data.get("log") or [])
        return data

    # ------------------------------------------------------------ 实时通道（SSE）
    def start(self) -> None:
        if not self.realtime_enabled or self._thread is not None:
            return
        self._stop.clear()
        self._thread = threading.Thread(target=self._realtime_loop, name="farm-sse", daemon=True)
        self._thread.start()

    def stop(self) -> None:
        self._stop.set()
        # 主动掐断长连接，否则服务端那条流式响应会一直挂着，拖慢它的优雅停机。
        # 注意不能直接 response.close()：读线程正持有那把锁，close 会一直等它，
        # 而它要等下一次心跳（15 秒）才醒。直接 shutdown 底层 socket 才能立刻唤醒。
        sock = self._stream_socket
        if sock is not None:
            with contextlib.suppress(OSError):
                sock.shutdown(socket.SHUT_RDWR)
        response = self._stream_response
        if response is not None:
            with contextlib.suppress(Exception):
                response.close()

    def transport_label(self) -> str:
        if self.mode == "sse":
            return "SSE 实时"
        if self.mode == "poll":
            return "HTTP 轮询"
        return "HTTP"

    @property
    def stream_url(self) -> str:
        """SSE 长连接地址。

        令牌只能走查询参数：``EventSource`` 与 ``urllib`` 都没法给长连接加自定义头。
        """
        query: dict[str, Any] = {"farm": self.farm}
        if self.token:
            query["token"] = self.token
        if self.name:
            query["name"] = self.name
        return f"{self.endpoint}/api/stream?{urllib.parse.urlencode(query)}"

    def _realtime_loop(self) -> None:
        """实时线程：读 SSE 长连接；断了就一边轮询一边重连，永不放弃。"""
        while not self._stop.is_set():
            try:
                self._stream_once()
            except Exception as exc:
                self.connected = False
                self.last_error = f"{type(exc).__name__}: {exc}"
                self.mode = "poll"
            if self._stop.is_set():
                break
            self._poll_until_reconnect()

    def _poll_until_reconnect(self) -> None:
        """SSE 断开期间的降级通道：轻量轮询 + 到点再试一次长连接。"""
        deadline = time.time() + STREAM_RECONNECT_DELAY
        while not self._stop.is_set() and time.time() < deadline:
            try:
                data = self._http("GET", "/api/state")
                self._apply(data.get("state") or {})
            except GameError:
                pass
            self._stop.wait(0.5)

    def _stream_once(self) -> None:
        """读一次 SSE 流，直到断开或主动停止。

        直接按行迭代响应体：一行一行拿，不把响应收完，
        所以这是真正的流式读取，而且只用标准库。
        """
        request = urllib.request.Request(self.stream_url, headers={"Accept": "text/event-stream"})
        # 服务器每 HEARTBEAT_SECONDS 秒发一次心跳，读超时要比它宽裕
        with net.open_url(request, timeout=max(self.timeout, STREAM_READ_TIMEOUT)) as response:
            self._stream_response = response
            self._stream_socket = _underlying_socket(response)
            self.mode = "sse"
            self.connected = True
            self.last_error = ""
            try:
                for raw_line in response:
                    if self._stop.is_set():
                        return
                    line = raw_line.decode("utf-8", errors="replace").strip()
                    if not line or line.startswith(":"):
                        # 空行是帧分隔符；冒号开头是心跳注释，都无需处理
                        continue
                    if line.startswith("data:"):
                        self._handle_stream_message(line[len("data:") :].strip())
            finally:
                self._stream_response = None
                self._stream_socket = None

    def _handle_stream_message(self, payload: str) -> None:
        try:
            message = json.loads(payload)
        except json.JSONDecodeError:
            return
        if not isinstance(message, dict):
            return
        kind = message.get("type")

        if kind in ("welcome", "state", "sync"):
            if message.get("server"):
                self.server_info = message["server"]
            self.farm = message.get("farm", self.farm)
            if message.get("catalog"):
                self._catalog = message["catalog"]
            self._apply_online(message.get("state") or {}, message.get("online"))
            self._merge_log(message.get("log") or [])
            self.stats["pushes"] += 1
            return

        if kind == "day":
            self._merge_log(
                [
                    {
                        "seq": self._log_seq + 1,
                        "day": message.get("day", 0),
                        "kind": "info",
                        "text": f"第 {message.get('day')} 天开始",
                    }
                ]
                + [
                    {
                        "seq": self._log_seq + 2 + i,
                        "day": message.get("day", 0),
                        "kind": "warn",
                        "text": text,
                    }
                    for i, text in enumerate(message.get("events") or [])
                ]
            )
            return

        if kind == "error":
            self.last_error = message.get("message", "")
            self.stats["errors"] += 1
            return

        if kind == "bye":
            # 服务器主动收尾（停机）：记下原因，让上层提示"服务器已停止"
            self.connected = False
            self.last_error = message.get("message", "服务器已停机")


class RemoteBotClient(RemoteFarm):
    """给机器人用的薄封装：多一个"等服务器起来"的便利方法。"""

    def wait_online(self, seconds: float = 30.0) -> bool:
        deadline = time.time() + seconds
        while time.time() < deadline:
            if self.refresh():
                return True
            time.sleep(1.0)
        return False