snowy-yang
c414db4217
项目由 uv 管理，依赖装在 .venv。此前各处的提示语仍写作 `python main.py ...`，用户若在未激活虚拟环境时照抄，会因缺少 fastapi 而失败——这是正确性问题，不只是措辞。 涉及：main.py（12 行，含 prog 与各条 hint）、ai_player.py（5 行）、farm/client.py、farm/server/__main__.py（3 行，含 prog）、farm/server/__init__.py、farm/server/daemon.py、README（4 处示例与目录树注释）。 刻意保留 main.py:26 与 README:49 的「没装 uv？」兜底说明，那里 `python main.py` 是正确写法。 验收：ruff 全通过 / format 26 文件无改动 / pyright 0 错误 / 97 项测试通过。
749 lines
26 KiB
Python
749 lines
26 KiB
Python
"""远程农场客户端：让本地代码（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）：语义清晰、可重试、失败有明确错误码。
|
||
* **读数走 WebSocket**：服务器在状态变化时主动推过来，界面不用轮询。
|
||
* WS 不可用时自动降级成"HTTP 轮询 + 写操作"，功能不缺，只是延迟高一点。
|
||
|
||
请求库用的是标准库 ``urllib``，所以即使机器上没装任何第三方包，TUI 也能跑。
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import asyncio
|
||
import contextlib
|
||
import json
|
||
import threading
|
||
import time
|
||
import urllib.error
|
||
import urllib.parse
|
||
import urllib.request
|
||
from collections import deque
|
||
from collections.abc import Callable
|
||
from concurrent.futures import Future
|
||
from concurrent.futures import TimeoutError as FutureTimeout
|
||
from typing import Any
|
||
|
||
from . import config, net
|
||
from .engine import GameError
|
||
|
||
DEFAULT_ENDPOINT = "http://127.0.0.1:8765"
|
||
LOG_KEEP = 200
|
||
WS_RECONNECT_DELAY = 3.0
|
||
|
||
|
||
# --------------------------------------------------------------------- 状态视图
|
||
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] = {}
|
||
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 ap(self) -> int:
|
||
return int(self._d.get("ap", 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", {})
|
||
|
||
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 / ws / 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._pending: dict[int, Future] = {}
|
||
self._req_id = 0
|
||
self._stop = threading.Event()
|
||
self._thread: threading.Thread | None = None
|
||
self._ws_socket: Any = None
|
||
self._ws_loop: Any = None
|
||
self._catalog: dict[str, Any] = {
|
||
"crops": [],
|
||
"animals": [],
|
||
"feed_price": config.FEED_UNIT_PRICE,
|
||
"water_capacity": config.WATER_CAPACITY,
|
||
"ap_cost": config.AP_COST,
|
||
}
|
||
self._log: deque[dict[str, Any]] = deque(maxlen=LOG_KEEP)
|
||
self._log_seq = 0
|
||
self.stats = {"http": 0, "ws_actions": 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 DEFAULT_ENDPOINT).strip().rstrip("/")
|
||
if endpoint.startswith("tcp://"):
|
||
endpoint = "http://" + endpoint[len("tcp://") :]
|
||
if not endpoint.startswith(("http://", "https://")):
|
||
endpoint = "http://" + endpoint
|
||
return 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:
|
||
"""主动拉一次全量数据（首次连接、以及 WS 断开后的兜底）。"""
|
||
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]:
|
||
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, transport: str = "auto"
|
||
) -> dict[str, Any]:
|
||
"""执行一个动作。
|
||
|
||
优先走 WebSocket（延迟更低、顺便复用已有连接）；只有**通道类**错误
|
||
（连接断了、超时）才退回 HTTP 重试——业务错误必须原样抛给玩家，
|
||
否则"金币不足"会被悄悄重试成两次操作。
|
||
"""
|
||
params = params or {}
|
||
use_ws = (transport == "ws") or (
|
||
transport == "auto" and self.mode == "ws" and self.connected
|
||
)
|
||
if use_ws:
|
||
try:
|
||
result = self._ws_action(action, params)
|
||
except GameError as exc:
|
||
if exc.code in ("ws_closed", "timeout", "network"):
|
||
self.last_error = f"WebSocket 不可用（{exc.message}），已改走 HTTP"
|
||
else:
|
||
raise
|
||
else:
|
||
self.stats["ws_actions"] += 1
|
||
if isinstance(result.get("snapshot"), dict):
|
||
self._apply(result["snapshot"])
|
||
elif isinstance(result.get("state"), dict):
|
||
self._apply(result["state"])
|
||
self._merge_log(result.get("log") or [])
|
||
return result
|
||
|
||
# full=1：让服务器顺手把完整快照带回来，省掉"操作后再查一次"
|
||
data = self._http("POST", "/api/actions", {"action": action, "params": params}, 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 sleep(self) -> dict[str, Any]:
|
||
return self._action("sleep")
|
||
|
||
def pause(self, paused: bool | None = None) -> dict[str, Any]:
|
||
params: dict[str, Any] = {}
|
||
if paused is not None:
|
||
params["paused"] = paused
|
||
data = self._action("pause", params)
|
||
self._apply(data.get("state") or {})
|
||
# 本地视图也要反映暂停态，TUI 的倒计时依赖它
|
||
if "paused" in data:
|
||
merged = self.state.to_dict()
|
||
merged["paused"] = data["paused"]
|
||
merged["seconds_to_tick"] = None if data["paused"] else merged.get("seconds_to_tick")
|
||
self.state.apply(merged)
|
||
return data
|
||
|
||
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
|
||
|
||
# ------------------------------------------------------------ WebSocket
|
||
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-ws", daemon=True)
|
||
self._thread.start()
|
||
|
||
def stop(self) -> None:
|
||
self._stop.set()
|
||
with self._lock:
|
||
for future in self._pending.values():
|
||
if not future.done():
|
||
future.set_exception(GameError("closed", "连接已关闭"))
|
||
self._pending.clear()
|
||
|
||
def transport_label(self) -> str:
|
||
if self.mode == "ws":
|
||
return "WebSocket 实时"
|
||
if self.mode == "poll":
|
||
return "HTTP 轮询"
|
||
return "HTTP"
|
||
|
||
@property
|
||
def ws_url(self) -> str:
|
||
parts = urllib.parse.urlsplit(self.endpoint)
|
||
scheme = "wss" if parts.scheme == "https" else "ws"
|
||
query = {"farm": self.farm}
|
||
if self.token:
|
||
query["token"] = self.token
|
||
if self.name:
|
||
query["name"] = self.name
|
||
return f"{scheme}://{parts.netloc}/ws?{urllib.parse.urlencode(query)}"
|
||
|
||
def _realtime_loop(self) -> None:
|
||
"""实时线程：先试 WebSocket，不行就退化成 HTTP 轮询，永不放弃重连。"""
|
||
while not self._stop.is_set():
|
||
try:
|
||
self._ws_session()
|
||
except Exception as exc:
|
||
self.connected = False
|
||
self.last_error = f"{type(exc).__name__}: {exc}"
|
||
if self.mode != "ws":
|
||
self.mode = "poll"
|
||
self._fail_pending(self.last_error)
|
||
if self._stop.is_set():
|
||
break
|
||
self._poll_until_reconnect()
|
||
|
||
def _poll_until_reconnect(self) -> None:
|
||
"""WS 断开期间的降级通道：轻量轮询 + 到点再试一次 WS。"""
|
||
if self.mode == "ws":
|
||
self.mode = "poll"
|
||
deadline = time.time() + WS_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 _ws_session(self) -> None:
|
||
import asyncio
|
||
|
||
try:
|
||
import websockets
|
||
except ImportError:
|
||
# 没装 websockets：直接走轮询模式，功能不减
|
||
self.mode = "poll"
|
||
self._stop.wait(WS_RECONNECT_DELAY)
|
||
return
|
||
|
||
asyncio.run(self._ws_main(websockets))
|
||
|
||
async def _ws_main(self, websockets) -> None:
|
||
async with websockets.connect(
|
||
self.ws_url,
|
||
open_timeout=self.timeout,
|
||
ping_interval=20,
|
||
ping_timeout=20,
|
||
max_size=4 * 1024 * 1024,
|
||
) as socket:
|
||
self.mode = "ws"
|
||
self.connected = True
|
||
self.last_error = ""
|
||
self._ws_socket = socket
|
||
self._ws_loop = asyncio.get_running_loop()
|
||
try:
|
||
while not self._stop.is_set():
|
||
try:
|
||
raw = await asyncio.wait_for(socket.recv(), timeout=30.0)
|
||
except asyncio.TimeoutError:
|
||
await socket.send(json.dumps({"type": "ping"}))
|
||
continue
|
||
self._handle_ws_message(raw)
|
||
finally:
|
||
self._ws_socket = None
|
||
self._ws_loop = None
|
||
self._fail_pending("WebSocket 连接已断开")
|
||
|
||
def _fail_pending(self, reason: str) -> None:
|
||
"""连接断了以后，把还在等回执的动作请求全部失败掉。"""
|
||
with self._lock:
|
||
pending, self._pending = self._pending, {}
|
||
for future in pending.values():
|
||
if not future.done():
|
||
future.set_exception(GameError("ws_closed", reason))
|
||
|
||
def _handle_ws_message(self, raw: Any) -> None:
|
||
try:
|
||
message = json.loads(raw)
|
||
except (TypeError, json.JSONDecodeError):
|
||
return
|
||
if not isinstance(message, dict):
|
||
return
|
||
kind = message.get("type")
|
||
|
||
if kind == "welcome":
|
||
self.server_info = message.get("server") or {}
|
||
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 in ("state", "sync"):
|
||
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 == "result":
|
||
rid = message.get("id")
|
||
with self._lock:
|
||
future = self._pending.pop(rid, None) if rid is not None else None
|
||
if future is None or future.done():
|
||
return
|
||
if message.get("ok"):
|
||
self._merge_log(message.get("log") or [])
|
||
future.set_result(
|
||
{
|
||
"ok": True,
|
||
"message": message.get("message", "完成"),
|
||
**(message.get("result") or {}),
|
||
"state": message.get("state"),
|
||
"snapshot": message.get("snapshot"),
|
||
"log": message.get("log"),
|
||
}
|
||
)
|
||
else:
|
||
future.set_exception(
|
||
GameError(message.get("error", "error"), message.get("message", "操作失败"))
|
||
)
|
||
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", "")
|
||
|
||
def _ws_send(self, payload: dict[str, Any]) -> None:
|
||
"""把消息发到 WS 线程的事件循环里（跨线程投递）。"""
|
||
socket = getattr(self, "_ws_socket", None)
|
||
loop = getattr(self, "_ws_loop", None)
|
||
if socket is None or loop is None:
|
||
raise GameError("ws_closed", "WebSocket 尚未连接")
|
||
asyncio.run_coroutine_threadsafe(socket.send(json.dumps(payload)), loop)
|
||
|
||
def _ws_action(
|
||
self, action: str, params: dict[str, Any], timeout: float | None = None
|
||
) -> dict[str, Any]:
|
||
with self._lock:
|
||
self._req_id += 1
|
||
rid = self._req_id
|
||
future: Future = Future()
|
||
self._pending[rid] = future
|
||
try:
|
||
self._ws_send({"type": "action", "id": rid, "action": action, "params": params})
|
||
except GameError:
|
||
with self._lock:
|
||
self._pending.pop(rid, None)
|
||
raise
|
||
try:
|
||
return future.result(timeout=timeout or self.timeout)
|
||
except FutureTimeout:
|
||
with self._lock:
|
||
self._pending.pop(rid, None)
|
||
# 超时是正常控制流：FutureTimeout 是内部实现细节，不必挂到异常链上
|
||
raise GameError("timeout", "服务器没有在超时时间内回应") from None
|
||
|
||
|
||
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
|