hhhge/SimpleFarm
1
0 Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

style: 引入 ruff，474 项告警清零

Browse Source 
配置见 pyproject.toml：
- line-length=100：贴着现有代码定。若用默认的 88 会重排 122 行，
  噪音淹没真正有价值的修复。
- 启用 E/W/F/I/UP/B/C4/SIM/RET/PERF/RUF；**不**启用 PTH。因为本项目的路径
  多是"字符串身份"——要塞进 subprocess 命令行、写进 JSON 存档、打印给用户，
  全换 pathlib 只是包一层再拆一层，换来 70 处纯风格改动。
- 忽略 RUF001/002/003：中文全角标点是正确写法，不是"歧义字符"。
- 忽略 PERF203：剩下两处都是"逐项容错"语义（逐个牧场存档要立刻上报失败、
  收获全部要跳过失败地块）。把 try 提到循环外会毁掉"一项失败不影响其它项"
  这个行为，是错的优化。

自动修复 358 项：Dict/List/Optional 换成 PEP 585/604 写法、导入排序、
失效的 noqa、空占位 f-string 等。

手工修复的（这些是真问题，不是风格偏好）：
- B904 x9：except 中 raise 补异常链。刻意区分——网络/IO 失败用 from exc 保留
  根因；参数解析失败用 from None，因为底层 ValueError 对玩家只是噪音。
- RUF012 x2：StateView 的 plots/animals 原本是**类级共享可变列表**，
  改成在 __init__ 里初始化（apply({}) 会提前返回，所以必须显式摆好）。
- F841：TUI 里一个赋值后再没用过的死变量。
- SIM105 x11：try/except/pass 统一改成 contextlib.suppress。
- SIM115：日志句柄改用 with。Popen 已把 fd 复制给子进程，父进程可安全关闭。
- E501 x7：拆超长行；main.py 那处顺手提取变量，避免超长嵌套表达式。
- E741：把 l/r 改成 left_line/right_line。
- 其余：RUF046 / PERF401 / RET504 / RUF005 / UP031 / B007。

ruff format 重排 24 个文件。

验证：97 项测试全绿。
This commit is contained in:
snowy-yang committed 2026-09-18 21:18:22 +08:00
1 parent c7c8a6308a
commit fcd4cb4649
25 files changed
+1204 -794
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

farm/__init__.py
+9 -2
View File
@@ -15,6 +15,13 @@
__version__ = "2.0.0"
__all__ = [
"config", "models", "engine", "storage", "keys", "tui",
"client", "bot", "server",
"bot",
"client",
"config",
"engine",
"keys",
"models",
"server",
"storage",
"tui",
]
farm/bot.py
+23 -20
View File
@@ -19,7 +19,7 @@ from __future__ import annotations
import random
import threading
import time
from typing import Any, Dict, List, Optional
from typing import Any
from .engine import GameError
@@ -27,13 +27,14 @@ from .engine import GameError
class RuleBot:
"""一个会种地、会养殖、会投资的规则机器人。"""
def __init__(self, client, verbose: bool = True, pause: float = 0.25,
name: str = "bot") -> None:
self.client = client # 任何与 RemoteFarm 接口一致的对象
def __init__(
self, client, verbose: bool = True, pause: float = 0.25, name: str = "bot"
) -> None:
self.client = client # 任何与 RemoteFarm 接口一致的对象
self.verbose = verbose
self.pause = pause
self.name = name
self.done: List[str] = []
self.done: list[str] = []
self.failures = 0
self.actions_executed = 0
self._stop = threading.Event()
@@ -51,8 +52,9 @@ class RuleBot:
return self._stop.is_set()
# ------------------------------------------------------------ 打分
def score(self, action: Dict[str, Any], state: Dict[str, Any],
catalog: Dict[str, Any]) -> float:
def score(
self, action: dict[str, Any], state: dict[str, Any], catalog: dict[str, Any]
) -> float:
"""给一个候选动作打分。分数越高越先执行。
这张表就是整个"策略"：先做免费的，再喂饱牲畜，再补水，再扩张。
@@ -90,15 +92,16 @@ class RuleBot:
return 34.0
return 4.0
if name == "buy_animal":
info = next((a for a in catalog.get("animals", [])
if a["key"] == params.get("kind")), None)
info = next(
(a for a in catalog.get("animals", []) if a["key"] == params.get("kind")), None
)
cost = info["cost"] if info else 0
if cost and gold >= cost * 3 and ap >= 1:
return 30.0 + (info.get("profit_per_day", 0) if info else 0)
return 3.0
return 1.0
def choose(self) -> Optional[Dict[str, Any]]:
def choose(self) -> dict[str, Any] | None:
"""在服务器给出的合法动作里挑一个最优的。"""
actions = self.client.available_actions()
if not actions:
@@ -111,7 +114,7 @@ class RuleBot:
return best
# ------------------------------------------------------------ 单步
def step(self) -> Optional[str]:
def step(self) -> str | None:
"""执行一个动作，返回动作名；没得做或做不了返回 None。"""
action = self.choose()
if action is None:
@@ -119,7 +122,7 @@ class RuleBot:
name = action["action"]
params = action.get("params") or {}
try:
result = self.client._action(name, params) # 走通用动作入口
result = self.client._action(name, params) # 走通用动作入口
except GameError as exc:
self.failures += 1
if exc.code == "cooldown":
@@ -151,8 +154,7 @@ class RuleBot:
return True
# ------------------------------------------------------------ 主循环
def run(self, rounds: int = 0, max_minutes: float = 0.0,
idle_limit: int = 8) -> Dict[str, Any]:
def run(self, rounds: int = 0, max_minutes: float = 0.0, idle_limit: int = 8) -> dict[str, Any]:
"""跑机器人。
``rounds`` 为 0 表示不限轮数；``max_minutes`` 为 0 表示不限时间。
@@ -161,8 +163,7 @@ class RuleBot:
start_day = self.client.snapshot().get("day", 1)
idle = 0
round_no = 0
self.say(f"机器人 {self.name} 上线（第 {start_day} 天，"
f"{self.client.transport_label()}）")
self.say(f"机器人 {self.name} 上线（第 {start_day} 天，{self.client.transport_label()}）")
while not self._stop.is_set():
round_no += 1
@@ -199,8 +200,10 @@ class RuleBot:
"net_worth": self.client.snapshot().get("net_worth"),
"gold": self.client.snapshot().get("gold"),
}
self.say(f"机器人 {self.name} 下线：跑了 {summary['days']} 天、"
f"执行 {summary['actions']} 个动作、净资产 {summary['net_worth']}")
self.say(
f"机器人 {self.name} 下线：跑了 {summary['days']} 天、"
f"执行 {summary['actions']} 个动作、净资产 {summary['net_worth']}"
)
return summary
@@ -209,13 +212,13 @@ def random_bot(client, **kwargs) -> RuleBot:
bot = RuleBot(client, **kwargs)
original = bot.choose
def choose() -> Optional[Dict[str, Any]]:
def choose() -> dict[str, Any] | None:
actions = client.available_actions()
if not actions:
return None
return random.choice(actions)
bot.choose = choose # type: ignore[assignment]
bot.choose = choose # type: ignore[assignment]
bot.name = kwargs.get("name", "random")
del original
return bot
farm/client.py
+132 -105
View File
@@ -27,6 +27,7 @@ TUI 原本直接拿着 ``FarmEngine`` 对象，现在游戏跑在服务器进程
from __future__ import annotations
import asyncio
import contextlib
import json
import threading
import time
@@ -34,12 +35,13 @@ import urllib.error
import urllib.parse
import urllib.request
from collections import deque
from concurrent.futures import Future, TimeoutError as FutureTimeout
from typing import Any, Callable, Deque, Dict, List, Optional
from collections.abc import Callable
from concurrent.futures import Future
from concurrent.futures import TimeoutError as FutureTimeout
from typing import Any
from . import config
from . import config, net
from .engine import GameError
from . import net
DEFAULT_ENDPOINT = "http://127.0.0.1:8765"
LOG_KEEP = 200
@@ -52,7 +54,7 @@ class PlotView:
__slots__ = ("_d",)
def __init__(self, data: Dict[str, Any]) -> None:
def __init__(self, data: dict[str, Any]) -> None:
self._d = data
@property
@@ -64,7 +66,7 @@ class PlotView:
return int(self._d.get("position", self.index + 1))
@property
def crop(self) -> Optional[str]:
def crop(self) -> str | None:
return self._d.get("crop")
@property
@@ -99,7 +101,7 @@ class PlotView:
def stage(self) -> str:
return self._d.get("stage", "")
def to_dict(self) -> Dict[str, Any]:
def to_dict(self) -> dict[str, Any]:
return dict(self._d)
@@ -108,7 +110,7 @@ class AnimalView:
__slots__ = ("_d",)
def __init__(self, data: Dict[str, Any]) -> None:
def __init__(self, data: dict[str, Any]) -> None:
self._d = data
@property
@@ -155,18 +157,18 @@ class AnimalView:
def is_sick(self) -> bool:
return bool(self._d.get("is_sick"))
def to_dict(self) -> Dict[str, Any]:
def to_dict(self) -> dict[str, Any]:
return dict(self._d)
class StateView:
"""整局游戏的只读视图。刻意保留 ``to_dict()``，方便直接落盘或转发。"""
def __init__(self, data: Optional[Dict[str, Any]] = None) -> None:
self._d: Dict[str, Any] = {}
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:
def apply(self, data: dict[str, Any]) -> None:
"""用新数据刷新内部视图。
服务器会发两种东西：**完整快照**（带 plots/animals）和**精简状态**
@@ -210,7 +212,7 @@ class StateView:
return int(self._d.get("feed", 0))
@property
def market(self) -> Dict[str, float]:
def market(self) -> dict[str, float]:
return self._d.get("market", {})
def get(self, key: str, default: Any = None) -> Any:
@@ -219,21 +221,24 @@ class StateView:
def __getitem__(self, key: str) -> Any:
return self._d[key]
def to_dict(self) -> Dict[str, Any]:
def to_dict(self) -> dict[str, Any]:
return dict(self._d)
# 初始化占位，避免属性访问落空
plots: List[PlotView] = []
animals: List[AnimalView] = []
# --------------------------------------------------------------------- 客户端
class RemoteFarm:
"""远程农场的句柄：接口与 :class:`farm.engine.FarmEngine` 保持一致。"""
def __init__(self, endpoint: str = DEFAULT_ENDPOINT, farm: str = "default",
token: Optional[str] = None, name: str = "", timeout: float = 6.0,
realtime: bool = True, autostart: bool = True) -> None:
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 ""
@@ -243,24 +248,27 @@ class RemoteFarm:
self.state = StateView()
self.connected = False
self.mode = "http" # http / ws / poll
self.mode = "http" # http / ws / poll
self.last_error = ""
self.last_message = ""
self.server_info: Dict[str, Any] = {}
self.on_change: Optional[Callable[[], None]] = None
self.server_info: dict[str, Any] = {}
self.on_change: Callable[[], None] | None = None
self._lock = threading.RLock()
self._pending: Dict[int, Future] = {}
self._pending: dict[int, Future] = {}
self._req_id = 0
self._stop = threading.Event()
self._thread: Optional[threading.Thread] = None
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._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}
@@ -274,7 +282,7 @@ class RemoteFarm:
def _normalize(endpoint: str) -> str:
endpoint = (endpoint or DEFAULT_ENDPOINT).strip().rstrip("/")
if endpoint.startswith("tcp://"):
endpoint = "http://" + endpoint[len("tcp://"):]
endpoint = "http://" + endpoint[len("tcp://") :]
if not endpoint.startswith(("http://", "https://")):
endpoint = "http://" + endpoint
return endpoint
@@ -290,8 +298,9 @@ class RemoteFarm:
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: Optional[Dict[str, Any]] = None,
**params: Any) -> Dict[str, Any]:
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"}
@@ -313,14 +322,15 @@ class RemoteFarm:
self.connected = False
raise GameError(
"network",
f"连不上服务器 {self.endpoint}（{exc}）。"
"请先启动：python main.py serve --daemon",
f"连不上服务器 {self.endpoint}（{exc}）。请先启动：python main.py serve --daemon",
status=503,
)
) from exc
try:
data_out = json.loads(body) if body else {}
except json.JSONDecodeError:
raise GameError("bad_response", f"服务器返回了非 JSON 内容（HTTP {status}）", 502)
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):
@@ -352,49 +362,48 @@ class RemoteFarm:
raise
return False
def _apply_online(self, state: Dict[str, Any], online: Any) -> None:
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]:
def snapshot(self) -> dict[str, Any]:
return self.state.to_dict()
def catalog(self) -> Dict[str, Any]:
def catalog(self) -> dict[str, Any]:
return self._catalog
def recent_log(self, limit: int = 12, since: int = 0) -> List[Dict[str, Any]]:
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]]:
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]]:
def leaderboard(self) -> list[dict[str, Any]]:
return self._http("GET", "/api/farms").get("farms") or []
def world(self) -> Dict[str, Any]:
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:
def _apply(self, state: dict[str, Any]) -> None:
if not state:
return
self.state.apply(state)
if self.on_change is not None:
try:
# UI 回调抛异常不能影响数据同步
with contextlib.suppress(Exception):
self.on_change()
except Exception:
pass
def _merge_log(self, entries: List[Dict[str, Any]]) -> None:
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:
@@ -403,12 +412,14 @@ class RemoteFarm:
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)
self._log = deque(
sorted(self._log, key=lambda e: int(e.get("seq", 0))), maxlen=LOG_KEEP
)
# ------------------------------------------------------------ 写接口
def _action(self, action: str, params: Optional[Dict[str, Any]] = None,
transport: str = "auto") -> Dict[str, Any]:
def _action(
self, action: str, params: dict[str, Any] | None = None, transport: str = "auto"
) -> dict[str, Any]:
"""执行一个动作。
优先走 WebSocket（延迟更低、顺便复用已有连接）；只有**通道类**错误
@@ -416,8 +427,9 @@ class RemoteFarm:
否则"金币不足"会被悄悄重试成两次操作。
"""
params = params or {}
use_ws = (transport == "ws") or (transport == "auto" and self.mode == "ws"
and self.connected)
use_ws = (transport == "ws") or (
transport == "auto" and self.mode == "ws" and self.connected
)
if use_ws:
try:
result = self._ws_action(action, params)
@@ -436,48 +448,47 @@ class RemoteFarm:
return result
# full=1：让服务器顺手把完整快照带回来，省掉"操作后再查一次"
data = self._http("POST", "/api/actions", {"action": action, "params": params},
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]:
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]:
def water(self, plot: Any) -> dict[str, Any]:
return self._action("water", {"plot": plot})
def harvest(self, plot: Any) -> Dict[str, Any]:
def harvest(self, plot: Any) -> dict[str, Any]:
return self._action("harvest", {"plot": plot})
def harvest_all(self) -> Dict[str, Any]:
def harvest_all(self) -> dict[str, Any]:
return self._action("harvest_all")
def buy_animal(self, kind: Any) -> Dict[str, Any]:
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}
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]:
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]:
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]:
def buy_plot(self) -> dict[str, Any]:
return self._action("buy_plot")
def sleep(self) -> Dict[str, Any]:
def sleep(self) -> dict[str, Any]:
return self._action("sleep")
def pause(self, paused: Optional[bool] = None) -> Dict[str, Any]:
params: Dict[str, Any] = {}
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)
@@ -490,7 +501,7 @@ class RemoteFarm:
self.state.apply(merged)
return data
def reset(self) -> Dict[str, Any]:
def reset(self) -> dict[str, Any]:
data = self._action("reset")
with self._lock:
self._log.clear()
@@ -498,11 +509,11 @@ class RemoteFarm:
self.refresh()
return data
def save(self, path: Optional[str] = None) -> Dict[str, Any]:
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]:
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 {})
@@ -514,8 +525,7 @@ class RemoteFarm:
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 = threading.Thread(target=self._realtime_loop, name="farm-ws", daemon=True)
self._thread.start()
def stop(self) -> None:
@@ -586,9 +596,13 @@ class RemoteFarm:
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:
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 = ""
@@ -650,37 +664,48 @@ class RemoteFarm:
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"),
})
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", "操作失败")))
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 [])])
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:
def _ws_send(self, payload: dict[str, Any]) -> None:
"""把消息发到 WS 线程的事件循环里（跨线程投递）。"""
socket = getattr(self, "_ws_socket", None)
loop = getattr(self, "_ws_loop", None)
@@ -688,8 +713,9 @@ class RemoteFarm:
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: Optional[float] = None) -> Dict[str, Any]:
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
@@ -706,7 +732,8 @@ class RemoteFarm:
except FutureTimeout:
with self._lock:
self._pending.pop(rid, None)
raise GameError("timeout", "服务器没有在超时时间内回应")
# 超时是正常控制流：FutureTimeout 是内部实现细节，不必挂到异常链上
raise GameError("timeout", "服务器没有在超时时间内回应") from None
class RemoteBotClient(RemoteFarm):
farm/config.py
+80 -39
View File
@@ -6,32 +6,32 @@
from __future__ import annotations
from typing import Any, Dict
from typing import Any
# --------------------------------------------------------------------------
# 基础经济参数
# --------------------------------------------------------------------------
START_GOLD = 120 # 初始金币
START_LEVEL = 1 # 初始等级
START_AP = 10 # 每天行动点上限
START_FEED = 5 # 初始饲料库存
START_GOLD = 120 # 初始金币
START_LEVEL = 1 # 初始等级
START_AP = 10 # 每天行动点上限
START_FEED = 5 # 初始饲料库存
BASE_PLOTS = 6 # 初始土地数量
MAX_PLOTS = 16 # 土地数量硬上限
PLOT_BASE_PRICE = 200 # 第一块新地的价格
BASE_PLOTS = 6 # 初始土地数量
MAX_PLOTS = 16 # 土地数量硬上限
PLOT_BASE_PRICE = 200 # 第一块新地的价格
PLOT_PRICE_GROWTH = 1.75 # 每多买一块，价格乘以该系数
FEED_UNIT_PRICE = 6 # 主动购买饲料的单价
FEED_UNIT_PRICE = 6 # 主动购买饲料的单价
FEED_EMERGENCY_PRICE = 9 # 自动补货（缺饲料时喂食）的惩罚单价
FEED_BULK = 10 # 一次购买饲料的默认批量
FEED_BULK = 10 # 一次购买饲料的默认批量
WATER_CAPACITY = 3 # 一次浇水可支撑的天数
WATER_CAPACITY = 3 # 一次浇水可支撑的天数
PLANT_INITIAL_MOISTURE = 2 # 播种后地块自带的水分
PRICE_VOLATILITY = 0.18 # 每天价格波动幅度（±18%）
LUCKY_CHANCE = 0.12 # 丰收暴击概率（产出翻倍）
MAX_LOG_ENTRIES = 300 # 内存中保留的事件条数上限
MAX_FEED_PURCHASE = 200 # 单次购买饲料的数量上限
PRICE_VOLATILITY = 0.18 # 每天价格波动幅度（±18%）
LUCKY_CHANCE = 0.12 # 丰收暴击概率（产出翻倍）
MAX_LOG_ENTRIES = 300 # 内存中保留的事件条数上限
MAX_FEED_PURCHASE = 200 # 单次购买饲料的数量上限
# 每级提供的土地上限（基础 6 + 2/级，封顶 MAX_PLOTS）
PLOTS_PER_LEVEL = 2
@@ -48,11 +48,13 @@ AP_COST = {
"buy_feed": 0,
}
# 经验曲线：升到 level+1 所需经验
def exp_to_next(level: int) -> int:
"""返回从 level 升到 level+1 所需的经验值。"""
return int(80 * (max(1, level) ** 1.6))
# 升级奖励金币
def level_up_bonus(level: int) -> int:
"""升到 level 时发放的金币奖励。"""
@@ -70,30 +72,54 @@ def level_up_bonus(level: int) -> int:
# exp : 每次收获获得的经验
# level_req : 解锁所需等级
CROPS: Dict[str, Dict[str, Any]] = {
CROPS: dict[str, dict[str, Any]] = {
"wheat": {
"name": "小麦", "seed_cost": 5, "grow_days": 2,
"base_price": 14, "exp": 3, "level_req": 1,
"name": "小麦",
"seed_cost": 5,
"grow_days": 2,
"base_price": 14,
"exp": 3,
"level_req": 1,
},
"carrot": {
"name": "胡萝卜", "seed_cost": 8, "grow_days": 3,
"base_price": 26, "exp": 5, "level_req": 1,
"name": "胡萝卜",
"seed_cost": 8,
"grow_days": 3,
"base_price": 26,
"exp": 5,
"level_req": 1,
},
"tomato": {
"name": "番茄", "seed_cost": 16, "grow_days": 4,
"base_price": 52, "exp": 9, "level_req": 2,
"name": "番茄",
"seed_cost": 16,
"grow_days": 4,
"base_price": 52,
"exp": 9,
"level_req": 2,
},
"corn": {
"name": "玉米", "seed_cost": 28, "grow_days": 6,
"base_price": 95, "exp": 14, "level_req": 3,
"name": "玉米",
"seed_cost": 28,
"grow_days": 6,
"base_price": 95,
"exp": 14,
"level_req": 3,
},
"pumpkin": {
"name": "南瓜", "seed_cost": 45, "grow_days": 8,
"base_price": 180, "exp": 22, "level_req": 4,
"name": "南瓜",
"seed_cost": 45,
"grow_days": 8,
"base_price": 180,
"exp": 22,
"level_req": 4,
},
"strawberry": {
"name": "草莓", "seed_cost": 70, "grow_days": 10,
"base_price": 320, "exp": 35, "level_req": 5,
"name": "草莓",
"seed_cost": 70,
"grow_days": 10,
"base_price": 320,
"exp": 35,
"level_req": 5,
},
}
@@ -119,21 +145,36 @@ GROWTH_STAGES = [
# exp : 每次收集产出获得的经验
# level_req : 解锁所需等级
ANIMALS: Dict[str, Dict[str, Any]] = {
ANIMALS: dict[str, dict[str, Any]] = {
"chicken": {
"name": "鸡", "cost": 90, "product": "鸡蛋",
"base_price": 15, "cycle": 1, "feed_per_day": 1,
"exp": 4, "level_req": 1,
"name": "鸡",
"cost": 90,
"product": "鸡蛋",
"base_price": 15,
"cycle": 1,
"feed_per_day": 1,
"exp": 4,
"level_req": 1,
},
"sheep": {
"name": "羊", "cost": 220, "product": "羊毛",
"base_price": 55, "cycle": 2, "feed_per_day": 1,
"exp": 9, "level_req": 2,
"name": "羊",
"cost": 220,
"product": "羊毛",
"base_price": 55,
"cycle": 2,
"feed_per_day": 1,
"exp": 9,
"level_req": 2,
},
"cow": {
"name": "牛", "cost": 480, "product": "牛奶",
"base_price": 72, "cycle": 1, "feed_per_day": 2,
"exp": 15, "level_req": 3,
"name": "牛",
"cost": 480,
"product": "牛奶",
"base_price": 72,
"cycle": 1,
"feed_per_day": 2,
"exp": 15,
"level_req": 3,
},
}
@@ -147,7 +188,7 @@ MAX_ANIMALS = 8 # 牧场总容量
DEFAULT_TICK_SECONDS = 15.0 # 自动推进一天所需的真实秒数
MIN_TICK_SECONDS = 3.0
MAX_TICK_SECONDS = 600.0
SLEEP_COOLDOWN = 2.0 # 手动"休息"推进一天的冷却（秒）
SLEEP_COOLDOWN = 2.0 # 手动"休息"推进一天的冷却（秒）
# --------------------------------------------------------------------------
farm/engine.py
+56 -52
View File
@@ -16,10 +16,12 @@ TUI 与 HTTP/gRPC 接口都调用同一组方法，因此三方共享完全一
from __future__ import annotations
import contextlib
import random
import threading
import time
from typing import Any, Callable, Dict, List, Optional
from collections.abc import Callable
from typing import Any
from . import config
from .models import Animal, GameState, Plot, clamp
@@ -34,7 +36,7 @@ class GameError(Exception):
self.message = message
self.status = status
def to_dict(self) -> Dict[str, Any]:
def to_dict(self) -> dict[str, Any]:
return {"ok": False, "error": self.code, "message": self.message}
@@ -52,9 +54,9 @@ class FarmEngine:
self._last_sleep = 0.0
self.paused = False
# 由 Scheduler 维护：下一次自动推进的绝对时间戳（无调度器时为 None）
self.next_tick_at: Optional[float] = None
self.next_tick_at: float | None = None
# 状态变化回调（TUI 用它触发重绘）
self._listeners: List[Callable[[], None]] = []
self._listeners: list[Callable[[], None]] = []
# ================================================================ 订阅
def subscribe(self, callback: Callable[[], None]) -> None:
@@ -63,11 +65,9 @@ class FarmEngine:
def _notify(self) -> None:
for callback in list(self._listeners):
try:
# 监听器异常不能影响游戏主流程
with contextlib.suppress(Exception):
callback()
except Exception:
# 监听器异常不能影响游戏主流程
pass
# ================================================================ 日志
def add_log(self, text: str, kind: str = "info") -> None:
@@ -77,20 +77,20 @@ class FarmEngine:
{
"seq": self._log_seq,
"day": self.state.day,
"kind": kind, # info / good / bad / warn / action
"kind": kind, # info / good / bad / warn / action
"text": text,
}
)
if len(self.state.log) > config.MAX_LOG_ENTRIES:
del self.state.log[: len(self.state.log) - config.MAX_LOG_ENTRIES]
def recent_log(self, limit: int = 12, since: int = 0) -> List[Dict[str, Any]]:
def recent_log(self, limit: int = 12, since: int = 0) -> list[dict[str, Any]]:
with self._lock:
entries = [e for e in self.state.log if e["seq"] > since]
return entries[-limit:] if limit > 0 else entries
# ================================================================ 查询
def snapshot(self) -> Dict[str, Any]:
def snapshot(self) -> dict[str, Any]:
"""返回完整状态快照（供 TUI 渲染与 API 响应）。"""
with self._lock:
data = self.state.to_dict()
@@ -101,7 +101,7 @@ class FarmEngine:
return data
@property
def seconds_to_tick(self) -> Optional[float]:
def seconds_to_tick(self) -> float | None:
if self.paused or self.next_tick_at is None:
return None
return round(max(0.0, self.next_tick_at - time.time()), 1)
@@ -110,7 +110,7 @@ class FarmEngine:
with self._lock:
return round(self.state.market.get(key, 0.0), 2)
def catalog(self) -> Dict[str, Any]:
def catalog(self) -> dict[str, Any]:
"""返回所有可种植/可养殖对象及其在市场上的实时价格。"""
with self._lock:
crops = []
@@ -148,9 +148,7 @@ class FarmEngine:
"exp": info["exp"],
"level_req": info["level_req"],
"locked": self.state.level < info["level_req"],
"profit_per_day": round(
(price - feed_cost) / max(1, info["cycle"]), 2
),
"profit_per_day": round((price - feed_cost) / max(1, info["cycle"]), 2),
}
)
return {
@@ -161,14 +159,14 @@ class FarmEngine:
"ap_cost": config.AP_COST,
}
def available_actions(self) -> List[Dict[str, Any]]:
def available_actions(self) -> list[dict[str, Any]]:
"""列出当前**合法**的动作，供 AI 直接决策使用。
每个动作包含：``action``（接口名）、``params``（参数模板）、
``cost``（金币/行动点开销）、``reason``（为什么值得做）。
"""
with self._lock:
actions: List[Dict[str, Any]] = []
actions: list[dict[str, Any]] = []
state = self.state
# ---- 收获 / 浇水 / 播种
@@ -193,7 +191,9 @@ class FarmEngine:
"action": "harvest",
"params": {"plot": plot.index + 1},
"cost": {"gold": 0, "ap": config.AP_COST["harvest"]},
"reason": f"{plot.index + 1} 号地的{config.CROPS[plot.crop]['name']}已成熟",
"reason": (
f"{plot.index + 1} 号地的{config.CROPS[plot.crop]['name']}已成熟"
),
}
)
elif plot.moisture == 0 and state.ap >= config.AP_COST["water"]:
@@ -274,7 +274,8 @@ class FarmEngine:
try:
number = int(plot_no)
except (TypeError, ValueError):
raise GameError("bad_param", f"地块编号必须是整数，收到 {plot_no!r}")
# 底层 ValueError 的细节对调用者没意义，只保留友好提示
raise GameError("bad_param", f"地块编号必须是整数，收到 {plot_no!r}") from None
if number < 1 or number > len(self.state.plots):
raise GameError(
"no_such_plot",
@@ -286,7 +287,7 @@ class FarmEngine:
try:
aid = int(animal_id)
except (TypeError, ValueError):
raise GameError("bad_param", f"动物编号必须是整数，收到 {animal_id!r}")
raise GameError("bad_param", f"动物编号必须是整数，收到 {animal_id!r}") from None
for animal in self.state.animals:
if animal.aid == aid:
return animal
@@ -316,14 +317,14 @@ class FarmEngine:
self.state.total_spent += amount
def _earn(self, amount: float, reason: str) -> int:
earned = max(0, int(round(amount)))
earned = max(0, round(amount))
self.state.gold += earned
self.state.total_earned += earned
return earned
def _gain_exp(self, amount: int) -> List[str]:
def _gain_exp(self, amount: int) -> list[str]:
"""增加经验并处理连续升级，返回升级提示文本。"""
messages: List[str] = []
messages: list[str] = []
self.state.exp += int(amount)
guard = 0
while self.state.exp >= self.state.exp_needed and guard < 50:
@@ -353,7 +354,7 @@ class FarmEngine:
return 2 if random.random() < config.LUCKY_CHANCE else 1
# ================================================================ 动作
def plant(self, plot_no: Any, crop: Any) -> Dict[str, Any]:
def plant(self, plot_no: Any, crop: Any) -> dict[str, Any]:
with self._lock:
if not isinstance(crop, str) or crop not in config.CROPS:
raise GameError(
@@ -368,7 +369,9 @@ class FarmEngine:
)
plot = self._get_plot(plot_no)
if plot.crop is not None:
raise GameError("plot_busy", f"{plot.index + 1} 号地已经种了{config.CROPS[plot.crop]['name']}")
raise GameError(
"plot_busy", f"{plot.index + 1} 号地已经种了{config.CROPS[plot.crop]['name']}"
)
self._spend_gold(info["seed_cost"], f"购买{info['name']}种子")
self._spend_ap("plant")
@@ -382,11 +385,13 @@ class FarmEngine:
self._notify()
return {
"ok": True,
"message": f"{plot.index + 1} 号地已种下{info['name']}，{info['grow_days']} 天后成熟",
"message": (
f"{plot.index + 1} 号地已种下{info['name']}，{info['grow_days']} 天后成熟"
),
"plot": plot.to_dict(),
}
def water(self, plot_no: Any) -> Dict[str, Any]:
def water(self, plot_no: Any) -> dict[str, Any]:
with self._lock:
plot = self._get_plot(plot_no)
if plot.crop is None:
@@ -402,7 +407,7 @@ class FarmEngine:
self._notify()
return {"ok": True, "message": f"{plot.index + 1} 号地已浇水", "plot": plot.to_dict()}
def harvest(self, plot_no: Any) -> Dict[str, Any]:
def harvest(self, plot_no: Any) -> dict[str, Any]:
with self._lock:
plot = self._get_plot(plot_no)
if plot.crop is None:
@@ -426,9 +431,7 @@ class FarmEngine:
self.state.total_harvested += quantity
lucky = "（丰收！）" if quantity > 1 else ""
self.add_log(
f"收获 {quantity} 份{info['name']}{lucky}，+{earned} 金币", "good"
)
self.add_log(f"收获 {quantity} 份{info['name']}{lucky}，+{earned} 金币", "good")
levelups = self._gain_exp(info["exp"])
self._notify()
return {
@@ -440,7 +443,7 @@ class FarmEngine:
"level_ups": levelups,
}
def buy_animal(self, kind: Any) -> Dict[str, Any]:
def buy_animal(self, kind: Any) -> dict[str, Any]:
with self._lock:
if not isinstance(kind, str) or kind not in config.ANIMALS:
raise GameError(
@@ -469,7 +472,7 @@ class FarmEngine:
"animal": animal.to_dict(),
}
def feed(self, animal_id: Any, quantity: Any = None) -> Dict[str, Any]:
def feed(self, animal_id: Any, quantity: Any = None) -> dict[str, Any]:
with self._lock:
animal = self._get_animal(animal_id)
if animal.fed:
@@ -481,7 +484,7 @@ class FarmEngine:
try:
quantity = int(quantity)
except (TypeError, ValueError):
raise GameError("bad_param", "喂食数量必须是整数")
raise GameError("bad_param", "喂食数量必须是整数") from None
if quantity <= 0:
raise GameError("bad_param", "喂食数量必须大于 0")
if quantity > need:
@@ -507,7 +510,7 @@ class FarmEngine:
"animal": animal.to_dict(),
}
def collect(self, animal_id: Any) -> Dict[str, Any]:
def collect(self, animal_id: Any) -> dict[str, Any]:
with self._lock:
animal = self._get_animal(animal_id)
if animal.pending <= 0:
@@ -524,7 +527,8 @@ class FarmEngine:
self.state.total_collected += quantity
self.add_log(
f"收取 {quantity} 份{animal.product}（{animal.name} #{animal.aid}），+{earned} 金币",
f"收取 {quantity} 份{animal.product}（{animal.name} #{animal.aid}），"
f"+{earned} 金币",
"good",
)
levelups = self._gain_exp(animal.info["exp"] * quantity)
@@ -537,12 +541,12 @@ class FarmEngine:
"level_ups": levelups,
}
def buy_feed(self, quantity: Any = config.FEED_BULK) -> Dict[str, Any]:
def buy_feed(self, quantity: Any = config.FEED_BULK) -> dict[str, Any]:
with self._lock:
try:
quantity = int(quantity)
except (TypeError, ValueError):
raise GameError("bad_param", "饲料数量必须是整数")
raise GameError("bad_param", "饲料数量必须是整数") from None
if quantity <= 0:
raise GameError("bad_param", "饲料数量必须大于 0")
if quantity > config.MAX_FEED_PURCHASE:
@@ -561,7 +565,7 @@ class FarmEngine:
"feed": self.state.feed,
}
def buy_plot(self) -> Dict[str, Any]:
def buy_plot(self) -> dict[str, Any]:
with self._lock:
price = self.state.next_plot_price
if price < 0:
@@ -583,7 +587,7 @@ class FarmEngine:
"next_plot_price": self.state.next_plot_price,
}
def sleep(self) -> Dict[str, Any]:
def sleep(self) -> dict[str, Any]:
"""手动推进一天，带冷却避免刷时间。"""
with self._lock:
elapsed = time.time() - self._last_sleep
@@ -594,14 +598,14 @@ class FarmEngine:
events = self.advance_day(reason="休息")
return {"ok": True, "message": f"进入第 {self.state.day} 天", "events": events}
def pause(self, paused: Optional[bool] = None) -> Dict[str, Any]:
def pause(self, paused: bool | None = None) -> dict[str, Any]:
with self._lock:
self.paused = bool(paused) if paused is not None else not self.paused
self.add_log("时间已暂停" if self.paused else "时间继续流动", "warn")
self._notify()
return {"ok": True, "paused": self.paused}
def reset(self, keep_tick: bool = True) -> Dict[str, Any]:
def reset(self, keep_tick: bool = True) -> dict[str, Any]:
with self._lock:
tick = self.tick_seconds
self.state = GameState()
@@ -614,7 +618,7 @@ class FarmEngine:
self._notify()
return {"ok": True, "message": "农场已重置"}
def save(self, path: Optional[str] = None) -> Dict[str, Any]:
def save(self, path: str | None = None) -> dict[str, Any]:
"""把当前状态写入磁盘。
单机模式下由 TUI 直接调用；服务器模式里存档位置由牧场会话决定，
@@ -627,13 +631,13 @@ class FarmEngine:
return {"ok": True, "message": f"已存档到 {target}", "path": target}
# ================================================================ 时间推进
def advance_day(self, reason: str = "自动") -> List[str]:
def advance_day(self, reason: str = "自动") -> list[str]:
"""推进游戏内一天，返回本日发生的事件文本。"""
with self._lock:
state = self.state
state.day += 1
state.ap = state.max_ap
events: List[str] = []
events: list[str] = []
# ---- 作物生长
for plot in state.plots:
@@ -643,12 +647,14 @@ class FarmEngine:
plot.moisture -= 1
plot.growth = min(plot.grow_days, plot.growth + 1)
if plot.is_ripe:
events.append(f"{plot.index + 1} 号地的{config.CROPS[plot.crop]['name']}已成熟")
events.append(
f"{plot.index + 1} 号地的{config.CROPS[plot.crop]['name']}已成熟"
)
else:
events.append(f"{plot.index + 1} 号地缺水，作物停止生长")
# ---- 牲畜
dead: List[Animal] = []
dead: list[Animal] = []
for animal in state.animals:
if animal.fed:
animal.fed = False
@@ -710,7 +716,7 @@ class Scheduler:
self.engine = engine
self.enabled = enabled
self._stop = threading.Event()
self._thread: Optional[threading.Thread] = None
self._thread: threading.Thread | None = None
def start(self) -> None:
if not self.enabled or self._thread is not None:
@@ -734,10 +740,8 @@ class Scheduler:
if deadline is None:
deadline = time.time() + engine.tick_seconds
if time.time() >= deadline:
try:
with contextlib.suppress(Exception):
engine.advance_day(reason="时间流逝")
except Exception:
pass
engine.next_tick_at = time.time() + engine.tick_seconds
else:
engine.next_tick_at = deadline
farm/keys.py
+4 -5
View File
@@ -11,7 +11,6 @@ from __future__ import annotations
import os
import sys
from typing import Optional
IS_WINDOWS = os.name == "nt"
@@ -42,7 +41,7 @@ class KeyReader:
def __init__(self) -> None:
self.available = sys.stdin is not None and sys.stdin.isatty()
self._fd: Optional[int] = None
self._fd: int | None = None
self._saved_term = None
if self.available and not IS_WINDOWS:
@@ -65,13 +64,13 @@ class KeyReader:
self.available = False
# ------------------------------------------------------------------ 读取
def poll(self) -> Optional[str]:
def poll(self) -> str | None:
"""非阻塞读取一个按键；没有按键时返回 None。"""
if not self.available:
return None
return self._poll_windows() if IS_WINDOWS else self._poll_posix()
def _poll_windows(self) -> Optional[str]:
def _poll_windows(self) -> str | None:
import msvcrt
if not msvcrt.kbhit():
@@ -86,7 +85,7 @@ class KeyReader:
return _WIN_CTRL[char]
return char.lower()
def _poll_posix(self) -> Optional[str]:
def _poll_posix(self) -> str | None:
import select
if self._fd is None:
farm/models.py
+28 -29
View File
@@ -7,7 +7,7 @@
from __future__ import annotations
import itertools
from typing import Any, Dict, List, Optional
from typing import Any
from . import config
@@ -20,13 +20,13 @@ def clamp(value: float, low: float, high: float) -> float:
class Plot:
"""一块土地。"""
__slots__ = ("index", "crop", "growth", "moisture", "planted_day")
__slots__ = ("crop", "growth", "index", "moisture", "planted_day")
def __init__(self, index: int) -> None:
self.index = index # 0-based 下标，对外展示用 index+1
self.crop: Optional[str] = None
self.growth: float = 0.0 # 已生长天数
self.moisture: int = 0 # 剩余可支撑天数
self.index = index # 0-based 下标，对外展示用 index+1
self.crop: str | None = None
self.growth: float = 0.0 # 已生长天数
self.moisture: int = 0 # 剩余可支撑天数
self.planted_day: int = 0
# ---------------------------------------------------------------- 查询
@@ -63,7 +63,7 @@ class Plot:
return name
# ---------------------------------------------------------------- 序列化
def to_dict(self) -> Dict[str, Any]:
def to_dict(self) -> dict[str, Any]:
info = config.CROPS.get(self.crop) if self.crop else None
return {
"index": self.index,
@@ -81,7 +81,7 @@ class Plot:
}
@classmethod
def from_dict(cls, data: Dict[str, Any]) -> "Plot":
def from_dict(cls, data: dict[str, Any]) -> Plot:
plot = cls(int(data.get("index", 0)))
crop = data.get("crop")
# 校验存档里的作物名仍然合法，避免旧存档导致 KeyError
@@ -97,19 +97,19 @@ class Animal:
_id_counter = itertools.count(1)
__slots__ = ("aid", "kind", "fed", "progress", "starve", "pending")
__slots__ = ("aid", "fed", "kind", "pending", "progress", "starve")
def __init__(self, kind: str, aid: Optional[int] = None) -> None:
def __init__(self, kind: str, aid: int | None = None) -> None:
self.aid = aid if aid is not None else next(Animal._id_counter)
self.kind = kind
self.fed = False # 今天是否已喂食
self.progress = 0 # 距下次产出的累计天数
self.starve = 0 # 连续未进食天数
self.pending = 0 # 待收集的产出数量
self.fed = False # 今天是否已喂食
self.progress = 0 # 距下次产出的累计天数
self.starve = 0 # 连续未进食天数
self.pending = 0 # 待收集的产出数量
# ---------------------------------------------------------------- 查询
@property
def info(self) -> Dict[str, Any]:
def info(self) -> dict[str, Any]:
return config.ANIMALS[self.kind]
@property
@@ -134,7 +134,7 @@ class Animal:
return self.starve >= 2
# ---------------------------------------------------------------- 序列化
def to_dict(self) -> Dict[str, Any]:
def to_dict(self) -> dict[str, Any]:
return {
"id": self.aid,
"kind": self.kind,
@@ -150,7 +150,7 @@ class Animal:
}
@classmethod
def from_dict(cls, data: Dict[str, Any]) -> Optional["Animal"]:
def from_dict(cls, data: dict[str, Any]) -> Animal | None:
kind = data.get("kind")
if kind not in config.ANIMALS:
return None
@@ -174,12 +174,12 @@ class GameState:
self.exp: int = 0
self.ap: int = config.START_AP
self.feed: int = config.START_FEED
self.plots: List[Plot] = [Plot(i) for i in range(config.BASE_PLOTS)]
self.animals: List[Animal] = []
self.log: List[Dict[str, Any]] = []
self.plots: list[Plot] = [Plot(i) for i in range(config.BASE_PLOTS)]
self.animals: list[Animal] = []
self.log: list[dict[str, Any]] = []
# 市场价格（每天波动）：key -> 当前单价
self.market: Dict[str, float] = {
self.market: dict[str, float] = {
key: float(info["base_price"]) for key, info in config.CROPS.items()
}
for key, info in config.ANIMALS.items():
@@ -208,7 +208,7 @@ class GameState:
"""下一块土地的价格；已达上限时返回 -1。"""
if len(self.plots) >= self.max_plots:
return -1
return int(config.PLOT_BASE_PRICE * (config.PLOT_PRICE_GROWTH ** self.plots_bought))
return int(config.PLOT_BASE_PRICE * (config.PLOT_PRICE_GROWTH**self.plots_bought))
@property
def exp_needed(self) -> int:
@@ -229,8 +229,7 @@ class GameState:
def net_worth(self) -> float:
"""净资产：金币 + 已购地价 + 牲畜折价 + 库存 + 地里未收的作物。"""
land = sum(
config.PLOT_BASE_PRICE * (config.PLOT_PRICE_GROWTH ** i)
for i in range(self.plots_bought)
config.PLOT_BASE_PRICE * (config.PLOT_PRICE_GROWTH**i) for i in range(self.plots_bought)
)
crops = 0.0
for plot in self.plots:
@@ -243,7 +242,7 @@ class GameState:
return config.goal_index(self.net_worth)
@property
def next_goal(self) -> Optional[Dict[str, Any]]:
def next_goal(self) -> dict[str, Any] | None:
nxt = self.goal_index + 1
if nxt >= len(config.GOALS):
return None
@@ -251,7 +250,7 @@ class GameState:
return {"amount": amount, "text": text}
# ---------------------------------------------------------------- 序列化
def to_dict(self) -> Dict[str, Any]:
def to_dict(self) -> dict[str, Any]:
return {
"day": self.day,
"gold": self.gold,
@@ -281,7 +280,7 @@ class GameState:
}
@classmethod
def from_dict(cls, data: Dict[str, Any]) -> "GameState":
def from_dict(cls, data: dict[str, Any]) -> GameState:
state = cls()
state.day = max(1, int(data.get("day", 1)))
state.gold = max(0, int(data.get("gold", config.START_GOLD)))
@@ -295,11 +294,11 @@ class GameState:
count = int(clamp(len(plots) or config.BASE_PLOTS, config.BASE_PLOTS, config.MAX_PLOTS))
if len(plots) < count:
plots.extend(Plot(i) for i in range(len(plots), count))
state.plots = plots[:config.MAX_PLOTS]
state.plots = plots[: config.MAX_PLOTS]
for i, plot in enumerate(state.plots):
plot.index = i
animals: List[Animal] = []
animals: list[Animal] = []
for item in data.get("animals", [])[: config.MAX_ANIMALS]:
if isinstance(item, dict):
animal = Animal.from_dict(item)
farm/net.py
+6 -5
View File
@@ -18,14 +18,14 @@ from __future__ import annotations
import urllib.parse
import urllib.request
from typing import Any, Optional, Union
from typing import Any
LOOPBACK_HOSTS = {"127.0.0.1", "localhost", "::1", "[::1]", "0.0.0.0", ""}
_no_proxy_opener = urllib.request.build_opener(urllib.request.ProxyHandler({}))
def is_loopback(host: Optional[str]) -> bool:
def is_loopback(host: str | None) -> bool:
"""判断主机名是否是本机回环（含 127.x.x.x 网段与 IPv6 回环）。"""
if host is None:
return False
@@ -35,7 +35,7 @@ def is_loopback(host: Optional[str]) -> bool:
return host.startswith("127.")
def host_of(target: Union[str, urllib.request.Request]) -> Optional[str]:
def host_of(target: str | urllib.request.Request) -> str | None:
url = target.full_url if isinstance(target, urllib.request.Request) else target
try:
return urllib.parse.urlsplit(url).hostname
@@ -43,8 +43,9 @@ def host_of(target: Union[str, urllib.request.Request]) -> Optional[str]:
return None
def open_url(target: Union[str, urllib.request.Request], timeout: float = 6.0,
host: Optional[str] = None) -> Any:
def open_url(
target: str | urllib.request.Request, timeout: float = 6.0, host: str | None = None
) -> Any:
"""打开一个 URL 或 Request，回环地址自动绕过代理。
返回值是 ``http.client.HTTPResponse``，用法与 ``urllib.request.urlopen`` 一致。
farm/server/__init__.py
+29 -12
View File
@@ -22,19 +22,38 @@ from __future__ import annotations
from typing import Any
_LAZY = {
"API_VERSION", "SERVER_NAME", "ServerConfig", "ServerHandle",
"build_app", "create_app", "rules_text", "brief",
"Action", "Param", "ACTIONS", "run_action", "describe_actions", "action_names",
"FarmSession", "SessionManager", "FarmClock", "validate_farm_id", "DEFAULT_FARM",
"Client", "ConnectionHub", "Publisher",
"DaemonManager", "pid_alive",
"DASHBOARD_HTML", "dashboard_text",
"API_VERSION",
"SERVER_NAME",
"ServerConfig",
"ServerHandle",
"build_app",
"create_app",
"rules_text",
"brief",
"Action",
"Param",
"ACTIONS",
"run_action",
"describe_actions",
"action_names",
"FarmSession",
"SessionManager",
"FarmClock",
"validate_farm_id",
"DEFAULT_FARM",
"Client",
"ConnectionHub",
"Publisher",
"DaemonManager",
"pid_alive",
"DASHBOARD_HTML",
"dashboard_text",
}
__all__ = sorted(_LAZY)
def __getattr__(name: str) -> Any: # PEP 562：模块级惰性属性
def __getattr__(name: str) -> Any: # PEP 562：模块级惰性属性
if name not in _LAZY:
raise AttributeError(f"module {__name__!r} has no attribute {name!r}")
from importlib import import_module
@@ -43,11 +62,9 @@ def __getattr__(name: str) -> Any: # PEP 562：模块级惰性属性
module = import_module("farm.server.daemon")
elif name in ("DASHBOARD_HTML", "dashboard_text"):
module = import_module("farm.server.webui")
elif name in ("Action", "Param", "ACTIONS", "run_action", "describe_actions",
"action_names"):
elif name in ("Action", "Param", "ACTIONS", "run_action", "describe_actions", "action_names"):
module = import_module("farm.server.actions")
elif name in ("FarmSession", "SessionManager", "FarmClock", "validate_farm_id",
"DEFAULT_FARM"):
elif name in ("FarmSession", "SessionManager", "FarmClock", "validate_farm_id", "DEFAULT_FARM"):
module = import_module("farm.server.session")
elif name in ("Client", "ConnectionHub", "Publisher"):
module = import_module("farm.server.hub")
farm/server/__main__.py
+43 -23
View File
@@ -20,28 +20,40 @@ def build_parser() -> argparse.ArgumentParser:
description="SimpleFarm 游戏服务器：HTTP 接口给 AI，WebSocket 给实时界面。",
formatter_class=argparse.RawDescriptionHelpFormatter,
)
parser.add_argument("--host", default="127.0.0.1",
help="监听地址，默认 127.0.0.1；局域网联机用 0.0.0.0")
parser.add_argument(
"--host", default="127.0.0.1", help="监听地址，默认 127.0.0.1；局域网联机用 0.0.0.0"
)
parser.add_argument("--port", type=int, default=8765, help="监听端口，默认 8765")
parser.add_argument("--token", default=None, help="访问令牌，设置后所有接口都要带")
parser.add_argument("--tick", type=float, default=config.DEFAULT_TICK_SECONDS,
help=f"每多少秒推进一天，默认 {config.DEFAULT_TICK_SECONDS:.0f} 秒")
parser.add_argument(
"--tick",
type=float,
default=config.DEFAULT_TICK_SECONDS,
help=f"每多少秒推进一天，默认 {config.DEFAULT_TICK_SECONDS:.0f} 秒",
)
parser.add_argument("--save-dir", default=None, help="存档目录，默认 <项目根>/saves")
parser.add_argument("--save", dest="save_file", default=None,
help="兼容旧参数：指定 default 牧场的存档文件")
parser.add_argument(
"--save", dest="save_file", default=None, help="兼容旧参数：指定 default 牧场的存档文件"
)
parser.add_argument("--max-sessions", type=int, default=64, help="同时在线的牧场上限")
parser.add_argument("--idle-timeout", type=float, default=1800.0,
help="无人访问多少秒后把牧场移出内存（存档保留），默认 1800")
parser.add_argument(
"--idle-timeout",
type=float,
default=1800.0,
help="无人访问多少秒后把牧场移出内存（存档保留），默认 1800",
)
parser.add_argument("--world", default="SimpleFarm 公共农场", help="世界名称")
parser.add_argument("--log-level", default="info",
choices=["critical", "error", "warning", "info", "debug"])
parser.add_argument(
"--log-level", default="info", choices=["critical", "error", "warning", "info", "debug"]
)
parser.add_argument("--access-log", action="store_true", help="打印每个 HTTP 请求")
parser.add_argument("--no-remote-control", action="store_true",
help="禁止通过接口停机")
parser.add_argument("--daemon", action="store_true",
help="转到后台运行（等价于 python main.py serve --daemon）")
parser.add_argument("--pid-file", default=None,
help="把本进程 pid 写到该文件（守护进程模式会自动指定）")
parser.add_argument("--no-remote-control", action="store_true", help="禁止通过接口停机")
parser.add_argument(
"--daemon", action="store_true", help="转到后台运行（等价于 python main.py serve --daemon）"
)
parser.add_argument(
"--pid-file", default=None, help="把本进程 pid 写到该文件（守护进程模式会自动指定）"
)
parser.add_argument("--log-file", default=DEFAULT_LOG_FILE, help=argparse.SUPPRESS)
return parser
@@ -68,8 +80,11 @@ def main(argv=None) -> int:
if args.daemon:
manager = DaemonManager(
host=args.host, port=args.port, token=args.token,
tick=clamp_tick(args.tick), save_dir=save_dir,
host=args.host,
port=args.port,
token=args.token,
tick=clamp_tick(args.tick),
save_dir=save_dir,
pid_file=args.pid_file or DEFAULT_PID_FILE,
log_file=args.log_file or DEFAULT_LOG_FILE,
)
@@ -84,6 +99,7 @@ def main(argv=None) -> int:
return 2
try:
import websockets # noqa: F401 只是探测，uvicorn 需要它来做 WS 协议升级
ws_ok = True
except ImportError:
ws_ok = False
@@ -135,8 +151,10 @@ def main(argv=None) -> int:
print(f" 存档目录 {save_dir}")
print(f" 实时通道 ws://{cfg.host}:{cfg.port}/ws?farm=<牧场名>")
if not ws_ok:
print(" ⚠ 未检测到 websockets 库，/ws 实时通道可能不可用；"
"网页端会自动降级为轮询。安装：pip install 'uvicorn[standard]'")
print(
" ⚠ 未检测到 websockets 库，/ws 实时通道可能不可用；"
"网页端会自动降级为轮询。安装：pip install 'uvicorn[standard]'"
)
if cfg.token:
print(" 已启用访问令牌：请求需带 X-API-Token 头或 ?token= 参数")
@@ -146,15 +164,17 @@ def main(argv=None) -> int:
if args.pid_file:
try:
if os.path.exists(args.pid_file):
with open(args.pid_file, "r", encoding="utf-8") as handle:
with open(args.pid_file, encoding="utf-8") as handle:
if handle.read().strip() == str(os.getpid()):
os.remove(args.pid_file)
except OSError:
pass
if not getattr(server, "started", False):
print(f"启动失败：端口 {cfg.port} 可能已被占用。"
f"换一个端口试试：--port {cfg.port + 1}", file=sys.stderr)
print(
f"启动失败：端口 {cfg.port} 可能已被占用。换一个端口试试：--port {cfg.port + 1}",
file=sys.stderr,
)
return 3
print("服务器已停止。")
return 0
farm/server/actions.py
+113 -48
View File
@@ -13,7 +13,8 @@
from __future__ import annotations
from typing import Any, Callable, Dict, List, Optional
from collections.abc import Callable
from typing import Any
from .. import config
from ..engine import FarmEngine, GameError
@@ -22,12 +23,13 @@ from ..engine import FarmEngine, GameError
class Param:
"""一个动作参数的规格说明（同时用于校验与自动生成给 AI 的文档）。"""
__slots__ = ("name", "kind", "required", "default", "desc")
__slots__ = ("default", "desc", "kind", "name", "required")
def __init__(self, name: str, kind: str, required: bool = True,
default: Any = None, desc: str = "") -> None:
def __init__(
self, name: str, kind: str, required: bool = True, default: Any = None, desc: str = ""
) -> None:
self.name = name
self.kind = kind # "int" / "str" / "bool"
self.kind = kind # "int" / "str" / "bool"
self.required = required
self.default = default
self.desc = desc
@@ -58,7 +60,7 @@ class Param:
raise GameError("bad_param", f"参数 {self.name} 需要字符串，收到 {raw!r}")
return raw
def to_dict(self) -> Dict[str, Any]:
def to_dict(self) -> dict[str, Any]:
return {
"name": self.name,
"type": self.kind,
@@ -71,18 +73,23 @@ class Param:
class Action:
"""一个可执行动作。"""
__slots__ = ("name", "summary", "params", "handler", "realtime")
__slots__ = ("handler", "name", "params", "realtime", "summary")
def __init__(self, name: str, summary: str, params: List[Param],
handler: Callable[[FarmEngine, Dict[str, Any]], Dict[str, Any]],
realtime: bool = True) -> None:
def __init__(
self,
name: str,
summary: str,
params: list[Param],
handler: Callable[[FarmEngine, dict[str, Any]], dict[str, Any]],
realtime: bool = True,
) -> None:
self.name = name
self.summary = summary
self.params = params
self.handler = handler
self.realtime = realtime
def to_dict(self) -> Dict[str, Any]:
def to_dict(self) -> dict[str, Any]:
return {
"action": self.name,
"summary": self.summary,
@@ -91,12 +98,12 @@ class Action:
# ---------------------------------------------------------------------- 处理器
def _harvest_all(engine: FarmEngine, _params: Dict[str, Any]) -> Dict[str, Any]:
def _harvest_all(engine: FarmEngine, _params: dict[str, Any]) -> dict[str, Any]:
"""收获所有成熟作物。逐块调用引擎，个别地块失败不影响其它地块。"""
total_gold = 0
total_units = 0
plots: List[int] = []
errors: List[str] = []
plots: list[int] = []
errors: list[str] = []
# 复制列表：harvest 会修改 state.plots 中元素的内容，但不会增删元素
for plot in list(engine.state.plots):
if not plot.is_ripe:
@@ -121,81 +128,139 @@ def _harvest_all(engine: FarmEngine, _params: Dict[str, Any]) -> Dict[str, Any]:
}
def _save(engine: FarmEngine, _params: Dict[str, Any]) -> Dict[str, Any]:
def _save(engine: FarmEngine, _params: dict[str, Any]) -> dict[str, Any]:
return engine.save()
def _reset(engine: FarmEngine, _params: Dict[str, Any]) -> Dict[str, Any]:
def _reset(engine: FarmEngine, _params: dict[str, Any]) -> dict[str, Any]:
return engine.reset()
def _pause(engine: FarmEngine, params: Dict[str, Any]) -> Dict[str, Any]:
def _pause(engine: FarmEngine, params: dict[str, Any]) -> dict[str, Any]:
value = params.get("paused")
return engine.pause(value if isinstance(value, bool) else None)
# ---------------------------------------------------------------------- 动作表
ACTIONS: Dict[str, Action] = {}
ACTIONS: dict[str, Action] = {}
def _register(action: Action) -> None:
ACTIONS[action.name] = action
_register(Action("plant", "在指定地块播种", [
Param("plot", "int", True, None, "地块编号，从 1 开始"),
Param("crop", "str", True, None, "作物标识，如 wheat"),
], lambda e, p: e.plant(p["plot"], p["crop"])))
_register(
Action(
"plant",
"在指定地块播种",
[
Param("plot", "int", True, None, "地块编号，从 1 开始"),
Param("crop", "str", True, None, "作物标识，如 wheat"),
],
lambda e, p: e.plant(p["plot"], p["crop"]),
)
)
_register(Action("water", "给指定地块浇水（补满水分）", [
Param("plot", "int", True, None, "地块编号，从 1 开始"),
], lambda e, p: e.water(p["plot"])))
_register(
Action(
"water",
"给指定地块浇水（补满水分）",
[
Param("plot", "int", True, None, "地块编号，从 1 开始"),
],
lambda e, p: e.water(p["plot"]),
)
)
_register(Action("harvest", "收获指定地块的成熟作物", [
Param("plot", "int", True, None, "地块编号，从 1 开始"),
], lambda e, p: e.harvest(p["plot"])))
_register(
Action(
"harvest",
"收获指定地块的成熟作物",
[
Param("plot", "int", True, None, "地块编号，从 1 开始"),
],
lambda e, p: e.harvest(p["plot"]),
)
)
_register(Action("harvest_all", "一键收获所有成熟作物", [], _harvest_all))
_register(Action("buy_animal", "购买牲畜", [
Param("kind", "str", True, None, "牲畜标识：chicken / sheep / cow"),
], lambda e, p: e.buy_animal(p["kind"])))
_register(
Action(
"buy_animal",
"购买牲畜",
[
Param("kind", "str", True, None, "牲畜标识：chicken / sheep / cow"),
],
lambda e, p: e.buy_animal(p["kind"]),
)
)
_register(Action("feed", "喂食指定牲畜", [
Param("animal_id", "int", True, None, "牲畜编号（state.animals[].id）"),
Param("quantity", "int", False, None, "喂食份数，默认按需求；多喂无益"),
], lambda e, p: e.feed(p["animal_id"], p.get("quantity"))))
_register(
Action(
"feed",
"喂食指定牲畜",
[
Param("animal_id", "int", True, None, "牲畜编号（state.animals[].id）"),
Param("quantity", "int", False, None, "喂食份数，默认按需求；多喂无益"),
],
lambda e, p: e.feed(p["animal_id"], p.get("quantity")),
)
)
_register(Action("collect", "收取牲畜的产出并自动出售", [
Param("animal_id", "int", True, None, "牲畜编号"),
], lambda e, p: e.collect(p["animal_id"])))
_register(
Action(
"collect",
"收取牲畜的产出并自动出售",
[
Param("animal_id", "int", True, None, "牲畜编号"),
],
lambda e, p: e.collect(p["animal_id"]),
)
)
_register(Action("buy_feed", "购买饲料", [
Param("quantity", "int", False, config.FEED_BULK, f"份数，默认 {config.FEED_BULK}"),
], lambda e, p: e.buy_feed(p.get("quantity", config.FEED_BULK))))
_register(
Action(
"buy_feed",
"购买饲料",
[
Param("quantity", "int", False, config.FEED_BULK, f"份数，默认 {config.FEED_BULK}"),
],
lambda e, p: e.buy_feed(p.get("quantity", config.FEED_BULK)),
)
)
_register(Action("buy_plot", "花钱开垦一块新土地", [], lambda e, p: e.buy_plot()))
_register(Action("sleep", "主动推进一天（有冷却）", [], lambda e, p: e.sleep()))
_register(Action("pause", "暂停或恢复自动推进时间", [
Param("paused", "bool", False, None, "true 暂停 / false 恢复；省略则反转当前状态"),
], _pause))
_register(
Action(
"pause",
"暂停或恢复自动推进时间",
[
Param("paused", "bool", False, None, "true 暂停 / false 恢复；省略则反转当前状态"),
],
_pause,
)
)
_register(Action("save", "把当前农场写入存档文件", [], _save))
_register(Action("reset", "重置农场，从头开始经营", [], _reset))
def action_names() -> List[str]:
def action_names() -> list[str]:
return sorted(ACTIONS)
def describe_actions() -> List[Dict[str, Any]]:
def describe_actions() -> list[dict[str, Any]]:
return [ACTIONS[name].to_dict() for name in action_names()]
def run_action(engine: FarmEngine, name: Any, params: Optional[Dict[str, Any]] = None) -> Dict[str, Any]:
def run_action(
engine: FarmEngine, name: Any, params: dict[str, Any] | None = None
) -> dict[str, Any]:
"""校验参数并执行动作。
这是所有入口的**唯一**执行通道，因此输入校验只需要在这里做一次。
@@ -223,7 +288,7 @@ def run_action(engine: FarmEngine, name: Any, params: Optional[Dict[str, Any]] =
f"动作 {key} 不认识参数 {'、'.join(unknown)}（可用：{allowed}）",
)
clean: Dict[str, Any] = {}
clean: dict[str, Any] = {}
for spec in action.params:
if spec.name in params and params[spec.name] is not None:
clean[spec.name] = spec.coerce(params[spec.name])
farm/server/app.py
+154 -111
View File
@@ -15,10 +15,12 @@
from __future__ import annotations
import asyncio
import contextlib
import json
import time
from collections.abc import Callable
from dataclasses import dataclass, field
from typing import Any, Callable, Dict, List, Optional
from typing import Any
from fastapi import FastAPI, Request, WebSocket, WebSocketDisconnect
from fastapi.responses import HTMLResponse, JSONResponse, PlainTextResponse
@@ -40,7 +42,7 @@ class ServerConfig:
host: str = "127.0.0.1"
port: int = 8765
token: Optional[str] = None
token: str | None = None
tick_seconds: float = config.DEFAULT_TICK_SECONDS
save_dir: str = "saves"
max_sessions: int = 64
@@ -48,7 +50,7 @@ class ServerConfig:
autosave_interval: float = 60.0
allow_remote_control: bool = True
world_name: str = "SimpleFarm 公共农场"
extra: Dict[str, Any] = field(default_factory=dict)
extra: dict[str, Any] = field(default_factory=dict)
class ServerHandle:
@@ -59,8 +61,8 @@ class ServerHandle:
"""
def __init__(self) -> None:
self._start_callbacks: List[Callable[[], None]] = []
self._stop_callbacks: List[Callable[[], None]] = []
self._start_callbacks: list[Callable[[], None]] = []
self._stop_callbacks: list[Callable[[], None]] = []
self.stopping = False
def on_start(self, callback: Callable[[], None]) -> None:
@@ -71,25 +73,23 @@ class ServerHandle:
def notify_start(self) -> None:
for callback in list(self._start_callbacks):
try:
# 单个启动钩子失败不该拖垮整个服务
with contextlib.suppress(Exception):
callback()
except Exception:
pass
def request_stop(self) -> bool:
if self.stopping:
return False
self.stopping = True
for callback in list(self._stop_callbacks):
try:
# 停机要尽力而为：一个钩子抛异常不能拦住其它钩子
with contextlib.suppress(Exception):
callback()
except Exception:
pass
return True
# ---------------------------------------------------------------------- 工具
def brief(session: FarmSession, hub: ConnectionHub) -> Dict[str, Any]:
def brief(session: FarmSession, hub: ConnectionHub) -> dict[str, Any]:
"""操作后回给调用者的精简状态：够做下一步决策，又不至于太吵。"""
snap = session.snapshot()
return {
@@ -125,7 +125,7 @@ def rules_text() -> str:
f"1. 时间：每个牧场每 {config.DEFAULT_TICK_SECONDS:.0f} 秒自动推进一天（可配置）。"
f"每天开始时行动点恢复为 {config.START_AP} 点，作物生长、牲畜产出、市场价格都在此刻结算。\n"
f"2. 行动点：播种/浇水/喂食各消耗 1 点，收获与收取产出免费。行动点用完只能等第二天"
"（POST /api/actions {\"action\":\"sleep\"} 可主动推进一天，有 2 秒冷却）。\n"
'（POST /api/actions {"action":"sleep"} 可主动推进一天，有 2 秒冷却）。\n'
f"3. 水分：作物每天消耗 1 点水分（上限 {config.WATER_CAPACITY}），水分为 0 时停止生长。"
f"播种时自带 {config.PLANT_INITIAL_MOISTURE} 点水分。\n"
f"4. 收获：成熟后 harvest 获得金币与经验，有 {config.LUCKY_CHANCE:.0%} 概率丰收翻倍。\n"
@@ -139,17 +139,17 @@ def rules_text() -> str:
"排行榜见 GET /api/world。\n"
"\n推荐决策循环（AI 最短路径）：\n"
" GET /api/actions?farm=me → 拿到当前所有**合法**动作（含开销与理由）\n"
" POST /api/actions → {\"action\":\"<上一步里的 action>\","
"\"params\":{...}} 直接执行\n"
' POST /api/actions → {"action":"<上一步里的 action>",'
'"params":{...}} 直接执行\n'
" 响应里的 state 就是执行后的最新状态，无需再查一次。\n"
"\n实时通道：ws://host/ws?farm=me ，服务器在状态变化时推送 state 消息；"
"也可以直接在 WS 上发送 {\"type\":\"action\",\"action\":\"plant\","
"\"params\":{\"plot\":1,\"crop\":\"wheat\"},\"id\":1} 并等待 result 回执。\n"
'也可以直接在 WS 上发送 {"type":"action","action":"plant",'
'"params":{"plot":1,"crop":"wheat"},"id":1} 并等待 result 回执。\n'
)
# ---------------------------------------------------------------------- 应用
def create_app(cfg: Optional[ServerConfig] = None) -> FastAPI:
def create_app(cfg: ServerConfig | None = None) -> FastAPI:
cfg = cfg or ServerConfig()
manager = SessionManager(
save_dir=cfg.save_dir,
@@ -206,12 +206,12 @@ def create_app(cfg: Optional[ServerConfig] = None) -> FastAPI:
)
# ------------------------------------------------------------ 鉴权与取牧场
def authorized(supplied: Optional[str]) -> bool:
def authorized(supplied: str | None) -> bool:
if not cfg.token:
return True
return supplied == cfg.token
def token_of(request: Request) -> Optional[str]:
def token_of(request: Request) -> str | None:
return request.headers.get("x-api-token") or request.query_params.get("token")
def guard(request: Request) -> None:
@@ -227,19 +227,19 @@ def create_app(cfg: Optional[ServerConfig] = None) -> FastAPI:
raw = (request.query_params.get("full") or "").lower()
return raw in ("1", "true", "yes", "on")
async def body_of(request: Request) -> Dict[str, Any]:
async def body_of(request: Request) -> dict[str, Any]:
raw = await request.body()
if not raw:
return {}
try:
data = json.loads(raw.decode("utf-8"))
except (UnicodeDecodeError, json.JSONDecodeError) as exc:
raise GameError("bad_json", f"请求体不是合法 JSON：{exc}")
raise GameError("bad_json", f"请求体不是合法 JSON：{exc}") from exc
if not isinstance(data, dict):
raise GameError("bad_json", "请求体必须是 JSON 对象")
return data
def execute_action(session: FarmSession, name: Any, params: Any) -> Dict[str, Any]:
def execute_action(session: FarmSession, name: Any, params: Any) -> dict[str, Any]:
"""执行一个动作，返回引擎/会话级的结果字典。
``save`` 与 ``reset`` 必须由**牧场会话**处理而不是引擎：
@@ -250,21 +250,24 @@ def create_app(cfg: Optional[ServerConfig] = None) -> FastAPI:
if key == "save":
path = session.save()
session.engine.add_log(f"已存档到 {path}", "info")
return {"ok": True, "action": "save", "path": path,
"message": f"已存档到 {path}"}
return {"ok": True, "action": "save", "path": path, "message": f"已存档到 {path}"}
if key == "reset":
session.reset()
return {"ok": True, "action": "reset",
"message": f"牧场「{session.farm_id}」已重置，重新开始经营"}
return {
"ok": True,
"action": "reset",
"message": f"牧场「{session.farm_id}」已重置，重新开始经营",
}
return actions_mod.run_action(session.engine, name, params)
def run_and_wrap(session: FarmSession, name: Any, params: Any,
full: bool = False) -> Dict[str, Any]:
def run_and_wrap(
session: FarmSession, name: Any, params: Any, full: bool = False
) -> dict[str, Any]:
"""执行动作并拼装统一响应。HTTP 与 WS 共用。"""
result = execute_action(session, name, params)
counters["actions"] += 1
session.touch(count_action=True)
payload: Dict[str, Any] = dict(result)
payload: dict[str, Any] = dict(result)
payload["ok"] = True
payload["farm"] = session.farm_id
payload["seq"] = next_seq()
@@ -285,7 +288,7 @@ def create_app(cfg: Optional[ServerConfig] = None) -> FastAPI:
# ================================================================ 元信息
@app.get("/api/health", tags=["元信息"], summary="健康检查（无需鉴权）")
async def health() -> Dict[str, Any]:
async def health() -> dict[str, Any]:
return {
"ok": True,
"service": SERVER_NAME,
@@ -298,7 +301,7 @@ def create_app(cfg: Optional[ServerConfig] = None) -> FastAPI:
}
@app.get("/api/help", tags=["元信息"], summary="玩法规则与接口说明（给 AI 的说明书）")
async def help_page(request: Request) -> Dict[str, Any]:
async def help_page(request: Request) -> dict[str, Any]:
guard(request)
return {
"ok": True,
@@ -313,14 +316,16 @@ def create_app(cfg: Optional[ServerConfig] = None) -> FastAPI:
"catalog": "GET /api/catalog?farm=NAME",
"actions": "GET /api/actions?farm=NAME",
"log": "GET /api/log?farm=NAME&limit=20&since=0",
"execute": "POST /api/actions {\"action\":\"plant\",\"params\":{\"plot\":1,\"crop\":\"wheat\"}}",
"farms": "GET /api/farms | POST /api/farms {\"farm\":\"NAME\"}",
"execute": (
'POST /api/actions {"action":"plant","params":{"plot":1,"crop":"wheat"}}'
),
"farms": 'GET /api/farms | POST /api/farms {"farm":"NAME"}',
"realtime": "WS /ws?farm=NAME",
},
}
@app.get("/api/world", tags=["世界"], summary="世界概览：排行榜 + 在线连接 + 时钟")
async def world(request: Request) -> Dict[str, Any]:
async def world(request: Request) -> dict[str, Any]:
guard(request)
return {
"ok": True,
@@ -336,13 +341,13 @@ def create_app(cfg: Optional[ServerConfig] = None) -> FastAPI:
# ================================================================ 牧场
@app.get("/api/farms", tags=["世界"], summary="牧场排行榜")
async def list_farms(request: Request) -> Dict[str, Any]:
async def list_farms(request: Request) -> dict[str, Any]:
guard(request)
rows = manager.leaderboard(hub.online_map())
return {"ok": True, "count": len(rows), "farms": rows}
@app.post("/api/farms", tags=["世界"], summary="创建或进入一个牧场")
async def create_farm(request: Request) -> Dict[str, Any]:
async def create_farm(request: Request) -> dict[str, Any]:
guard(request)
body = await body_of(request)
farm = validate_farm_id(body.get("farm") or body.get("name"))
@@ -354,7 +359,7 @@ def create_app(cfg: Optional[ServerConfig] = None) -> FastAPI:
try:
tick_value = float(tick)
except (TypeError, ValueError):
raise GameError("bad_param", "tick 必须是数字")
raise GameError("bad_param", "tick 必须是数字") from None
session.engine.tick_seconds = max(
config.MIN_TICK_SECONDS, min(config.MAX_TICK_SECONDS, tick_value)
)
@@ -370,7 +375,7 @@ def create_app(cfg: Optional[ServerConfig] = None) -> FastAPI:
# ================================================================ 单牧场只读
@app.get("/api/state", tags=["牧场"], summary="完整状态快照")
async def get_state(request: Request) -> Dict[str, Any]:
async def get_state(request: Request) -> dict[str, Any]:
guard(request)
session = session_of(request)
return {
@@ -381,13 +386,13 @@ def create_app(cfg: Optional[ServerConfig] = None) -> FastAPI:
}
@app.get("/api/catalog", tags=["牧场"], summary="作物/牲畜目录与实时价格")
async def get_catalog(request: Request) -> Dict[str, Any]:
async def get_catalog(request: Request) -> dict[str, Any]:
guard(request)
session = session_of(request)
return {"ok": True, "farm": session.farm_id, "catalog": session.engine.catalog()}
@app.get("/api/actions", tags=["牧场"], summary="当前所有合法动作（AI 推荐入口）")
async def get_actions(request: Request) -> Dict[str, Any]:
async def get_actions(request: Request) -> dict[str, Any]:
guard(request)
session = session_of(request)
items = session.engine.available_actions()
@@ -400,11 +405,11 @@ def create_app(cfg: Optional[ServerConfig] = None) -> FastAPI:
}
@app.get("/api/log", tags=["牧场"], summary="最近事件")
async def get_log(request: Request) -> Dict[str, Any]:
async def get_log(request: Request) -> dict[str, Any]:
guard(request)
session = session_of(request)
limit = _int(request.query_params.get("limit"), 20, 1, 200)
since = _int(request.query_params.get("since"), 0, 0, 10 ** 9)
since = _int(request.query_params.get("since"), 0, 0, 10**9)
entries = session.engine.recent_log(limit=limit, since=since)
return {
"ok": True,
@@ -416,7 +421,7 @@ def create_app(cfg: Optional[ServerConfig] = None) -> FastAPI:
# ================================================================ 写操作
@app.post("/api/actions", tags=["动作"], summary="通用动作入口（推荐 AI 使用）")
async def post_action(request: Request) -> Dict[str, Any]:
async def post_action(request: Request) -> dict[str, Any]:
guard(request)
body = await body_of(request)
session = session_of(request)
@@ -428,7 +433,7 @@ def create_app(cfg: Optional[ServerConfig] = None) -> FastAPI:
return run_and_wrap(session, name, params, full_requested(request))
@app.post("/api/batch", tags=["动作"], summary="批量执行动作（一个请求干多件事）")
async def post_batch(request: Request) -> Dict[str, Any]:
async def post_batch(request: Request) -> dict[str, Any]:
"""给 AI 的省流入口：一次提交多个动作，按顺序执行。
每个动作独立成败，失败的只记错误不中断后续动作。
@@ -441,7 +446,7 @@ def create_app(cfg: Optional[ServerConfig] = None) -> FastAPI:
raise GameError("bad_param", "actions 必须是至少含一个元素的数组")
if len(items) > 50:
raise GameError("too_many", "单次批量最多 50 个动作")
results: List[Dict[str, Any]] = []
results: list[dict[str, Any]] = []
for item in items:
if not isinstance(item, dict):
results.append({"ok": False, "error": "bad_param", "message": "每个动作必须是对象"})
@@ -455,7 +460,7 @@ def create_app(cfg: Optional[ServerConfig] = None) -> FastAPI:
counters["errors"] += 1
results.append({**exc.to_dict(), "action": item.get("action")})
ok_count = sum(1 for r in results if r.get("ok"))
payload: Dict[str, Any] = {
payload: dict[str, Any] = {
"ok": True,
"farm": session.farm_id,
"done": ok_count,
@@ -468,7 +473,7 @@ def create_app(cfg: Optional[ServerConfig] = None) -> FastAPI:
return payload
def make_action_route(action_name: str) -> Callable[[Request], Any]:
async def route(request: Request) -> Dict[str, Any]:
async def route(request: Request) -> dict[str, Any]:
guard(request)
body = await body_of(request)
session = session_of(request)
@@ -489,19 +494,21 @@ def create_app(cfg: Optional[ServerConfig] = None) -> FastAPI:
# ================================================================ 管理
@app.post("/api/admin/save", tags=["管理"], summary="把所有牧场写入存档")
async def admin_save(request: Request) -> Dict[str, Any]:
async def admin_save(request: Request) -> dict[str, Any]:
guard(request)
saved: List[str] = []
saved: list[str] = []
for session in manager.all():
try:
session.save()
saved.append(session.farm_id)
except OSError as exc:
raise GameError("save_failed", f"{session.farm_id} 存档失败：{exc}", status=500)
raise GameError(
"save_failed", f"{session.farm_id} 存档失败：{exc}", status=500
) from exc
return {"ok": True, "message": f"已保存 {len(saved)} 个牧场", "farms": saved}
@app.post("/api/admin/shutdown", tags=["管理"], summary="优雅停机（守护进程 stop 使用）")
async def admin_shutdown(request: Request) -> Dict[str, Any]:
async def admin_shutdown(request: Request) -> dict[str, Any]:
guard(request)
if not cfg.allow_remote_control:
raise GameError("forbidden", "服务器未开放远程停机", status=403)
@@ -535,33 +542,37 @@ def create_app(cfg: Optional[ServerConfig] = None) -> FastAPI:
session = manager.get(farm, create=True)
except GameError as exc:
# 例如牧场数量达到上限：说清楚原因再关，别让客户端以为是网络问题
await websocket.send_json({"type": "error", "error": exc.code,
"message": exc.message})
await websocket.send_json({"type": "error", "error": exc.code, "message": exc.message})
await websocket.close(code=1013, reason=exc.message[:100])
return
hub.register(client)
session.touch()
await websocket.send_json({
"type": "welcome",
"client_id": client.cid,
"farm": farm,
"server": {
"name": SERVER_NAME,
"version": API_VERSION,
"world": cfg.world_name,
"tick_seconds": session.engine.tick_seconds,
},
"state": session.snapshot(),
"catalog": session.engine.catalog(),
"actions": session.engine.available_actions(),
"log": session.engine.recent_log(limit=12),
"online": hub.online(farm),
})
await websocket.send_json(
{
"type": "welcome",
"client_id": client.cid,
"farm": farm,
"server": {
"name": SERVER_NAME,
"version": API_VERSION,
"world": cfg.world_name,
"tick_seconds": session.engine.tick_seconds,
},
"state": session.snapshot(),
"catalog": session.engine.catalog(),
"actions": session.engine.available_actions(),
"log": session.engine.recent_log(limit=12),
"online": hub.online(farm),
}
)
hub.broadcast(
farm,
{"type": "notice", "farm": farm,
"text": f"{client.name or client.cid} 来到了牧场（在线 {hub.online(farm)}）"},
{
"type": "notice",
"farm": farm,
"text": f"{client.name or client.cid} 来到了牧场（在线 {hub.online(farm)}）",
},
exclude=client,
)
@@ -585,12 +596,14 @@ def create_app(cfg: Optional[ServerConfig] = None) -> FastAPI:
try:
data = json.loads(raw)
except json.JSONDecodeError as exc:
await websocket.send_json({"type": "error", "error": "bad_json",
"message": f"不是合法 JSON：{exc}"})
await websocket.send_json(
{"type": "error", "error": "bad_json", "message": f"不是合法 JSON：{exc}"}
)
return
if not isinstance(data, dict):
await websocket.send_json({"type": "error", "error": "bad_json",
"message": "消息必须是 JSON 对象"})
await websocket.send_json(
{"type": "error", "error": "bad_json", "message": "消息必须是 JSON 对象"}
)
return
kind = str(data.get("type") or "").lower()
@@ -601,55 +614,79 @@ def create_app(cfg: Optional[ServerConfig] = None) -> FastAPI:
return
if kind in ("sync", "refresh", "state"):
await websocket.send_json({
"type": "sync",
"id": rid,
"seq": next_seq(),
"farm": session.farm_id,
"state": session.snapshot(),
"actions": session.engine.available_actions(),
"log": session.engine.recent_log(limit=20),
})
await websocket.send_json(
{
"type": "sync",
"id": rid,
"seq": next_seq(),
"farm": session.farm_id,
"state": session.snapshot(),
"actions": session.engine.available_actions(),
"log": session.engine.recent_log(limit=20),
}
)
return
if kind == "help":
await websocket.send_json({"type": "help", "id": rid, "rules": rules_text(),
"actions": actions_mod.describe_actions()})
await websocket.send_json(
{
"type": "help",
"id": rid,
"rules": rules_text(),
"actions": actions_mod.describe_actions(),
}
)
return
if kind == "action":
try:
# 引擎是同步的，丢到线程池里跑，别堵住事件循环
result = await asyncio.to_thread(
execute_action, session, data.get("action"),
execute_action,
session,
data.get("action"),
data.get("params") or {},
)
except GameError as exc:
counters["errors"] += 1
await websocket.send_json({"type": "result", "id": rid, "ok": False,
"error": exc.code, "message": exc.message,
"farm": session.farm_id})
await websocket.send_json(
{
"type": "result",
"id": rid,
"ok": False,
"error": exc.code,
"message": exc.message,
"farm": session.farm_id,
}
)
return
counters["actions"] += 1
session.touch(count_action=True)
await websocket.send_json({
"type": "result",
"id": rid,
"ok": True,
"farm": session.farm_id,
"message": result.get("message", "完成"),
"result": result,
"state": brief(session, hub),
# 回执里带上完整快照：客户端可以立刻刷新画面，不必等下一次推送
"snapshot": session.snapshot(),
"log": session.engine.recent_log(limit=3),
"seq": next_seq(),
})
await websocket.send_json(
{
"type": "result",
"id": rid,
"ok": True,
"farm": session.farm_id,
"message": result.get("message", "完成"),
"result": result,
"state": brief(session, hub),
# 回执里带上完整快照：客户端可以立刻刷新画面，不必等下一次推送
"snapshot": session.snapshot(),
"log": session.engine.recent_log(limit=3),
"seq": next_seq(),
}
)
return
await websocket.send_json({"type": "error", "error": "unknown_message",
"message": f"不认识的消息类型 {kind!r}",
"hint": "可用：action / ping / sync / help"})
await websocket.send_json(
{
"type": "error",
"error": "unknown_message",
"message": f"不认识的消息类型 {kind!r}",
"hint": "可用：action / ping / sync / help",
}
)
writer_task = asyncio.create_task(writer())
reader_task = asyncio.create_task(reader())
@@ -666,20 +703,26 @@ def create_app(cfg: Optional[ServerConfig] = None) -> FastAPI:
session.touch()
hub.broadcast(
session.farm_id,
{"type": "notice", "farm": session.farm_id,
"text": f"{client.name or client.cid} 离开了牧场（在线 {hub.online(session.farm_id)}）"},
{
"type": "notice",
"farm": session.farm_id,
"text": (
f"{client.name or client.cid} 离开了牧场"
f"（在线 {hub.online(session.farm_id)}）"
),
},
)
return app
def _int(raw: Optional[str], default: int, low: int, high: int) -> int:
def _int(raw: str | None, default: int, low: int, high: int) -> int:
if raw is None or raw == "":
return default
try:
value = int(raw)
except (TypeError, ValueError):
raise GameError("bad_param", f"参数必须是整数，收到 {raw!r}")
raise GameError("bad_param", f"参数必须是整数，收到 {raw!r}") from None
return max(low, min(high, value))
farm/server/daemon.py
+68 -57
View File
@@ -15,6 +15,7 @@
from __future__ import annotations
import contextlib
import json
import os
import signal
@@ -24,7 +25,7 @@ import time
import urllib.error
import urllib.parse
import urllib.request
from typing import Any, Dict, List, Optional, Tuple
from typing import Any
from .. import net
@@ -48,7 +49,7 @@ def pid_alive(pid: int) -> bool:
import ctypes
kernel32 = ctypes.windll.kernel32
handle = kernel32.OpenProcess(0x1000, False, pid) # QUERY_LIMITED_INFORMATION
handle = kernel32.OpenProcess(0x1000, False, pid) # QUERY_LIMITED_INFORMATION
if not handle:
return False
try:
@@ -68,10 +69,17 @@ def pid_alive(pid: int) -> bool:
class DaemonManager:
"""负责在后台拉起/停止 ``farm.server`` 进程。"""
def __init__(self, host: str = "127.0.0.1", port: int = 8765,
token: Optional[str] = None, tick: float = 15.0,
save_dir: str = "", pid_file: str = DEFAULT_PID_FILE,
log_file: str = DEFAULT_LOG_FILE, python: Optional[str] = None) -> None:
def __init__(
self,
host: str = "127.0.0.1",
port: int = 8765,
token: str | None = None,
tick: float = 15.0,
save_dir: str = "",
pid_file: str = DEFAULT_PID_FILE,
log_file: str = DEFAULT_LOG_FILE,
python: str | None = None,
) -> None:
self.host = host
self.port = port
self.token = token
@@ -83,9 +91,9 @@ class DaemonManager:
self.python = python or sys.executable
# ------------------------------------------------------------ pid 文件
def read_pid(self) -> Optional[int]:
def read_pid(self) -> int | None:
try:
with open(self.pid_file, "r", encoding="utf-8") as handle:
with open(self.pid_file, encoding="utf-8") as handle:
return int(handle.read().strip())
except (OSError, ValueError):
return None
@@ -96,12 +104,10 @@ class DaemonManager:
handle.write(str(pid))
def clear_pid(self) -> None:
try:
with contextlib.suppress(OSError):
os.remove(self.pid_file)
except OSError:
pass
def current(self) -> Optional[int]:
def current(self) -> int | None:
"""返回正在运行的服务器 pid；pid 文件过期则顺手清理。"""
pid = self.read_pid()
if pid is None:
@@ -112,7 +118,7 @@ class DaemonManager:
return None
# ------------------------------------------------------------ 探测
def probe(self, timeout: float = 1.5) -> Optional[Dict[str, Any]]:
def probe(self, timeout: float = 1.5) -> dict[str, Any] | None:
"""访问 /api/health，成功说明服务真的在监听。
用 :func:`farm.net.open_url`：本地回环请求不能被系统代理截走，
@@ -125,7 +131,7 @@ class DaemonManager:
except (urllib.error.URLError, OSError, ValueError):
return None
def world(self, timeout: float = 2.0) -> Optional[Dict[str, Any]]:
def world(self, timeout: float = 2.0) -> dict[str, Any] | None:
url = f"http://{self.host}:{self.port}/api/world"
if self.token:
url += "?token=" + urllib.parse.quote(self.token)
@@ -141,27 +147,36 @@ class DaemonManager:
with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as sock:
sock.settimeout(0.6)
return sock.connect_ex((self.host if self.host != "0.0.0.0" else "127.0.0.1",
self.port)) == 0
return (
sock.connect_ex((self.host if self.host != "0.0.0.0" else "127.0.0.1", self.port))
== 0
)
# ------------------------------------------------------------ 启动
def command(self) -> List[str]:
def command(self) -> list[str]:
args = [
self.python, "-m", "farm.server",
"--host", self.host,
"--port", str(self.port),
"--tick", str(self.tick),
"--save-dir", self.save_dir,
self.python,
"-m",
"farm.server",
"--host",
self.host,
"--port",
str(self.port),
"--tick",
str(self.tick),
"--save-dir",
self.save_dir,
# 关键：pid 文件由**服务进程自己**写。
# Windows 上 venv 的 python.exe 可能是个转发器，父进程拿到的 pid
# 未必是真正在监听端口的那个进程，写错 pid 会导致 stop 杀错人。
"--pid-file", self.pid_file,
"--pid-file",
self.pid_file,
]
if self.token:
args += ["--token", self.token]
return args
def start(self, wait: float = 12.0) -> Tuple[bool, str]:
def start(self, wait: float = 12.0) -> tuple[bool, str]:
existing = self.current()
if existing is not None:
return False, f"服务器已经在运行（pid {existing}），无需重复启动"
@@ -172,13 +187,13 @@ class DaemonManager:
os.makedirs(os.path.dirname(self.log_file), exist_ok=True)
os.makedirs(os.path.dirname(self.pid_file), exist_ok=True)
self.clear_pid() # 清掉上一次留下的陈旧 pid
self.clear_pid() # 清掉上一次留下的陈旧 pid
env = dict(os.environ)
env["PYTHONUNBUFFERED"] = "1"
env["PYTHONIOENCODING"] = "utf-8"
env["PYTHONPATH"] = PROJECT_ROOT + os.pathsep + env.get("PYTHONPATH", "")
kwargs: Dict[str, Any] = {
kwargs: dict[str, Any] = {
"cwd": PROJECT_ROOT,
"env": env,
"stdin": subprocess.DEVNULL,
@@ -194,20 +209,18 @@ class DaemonManager:
else:
kwargs["start_new_session"] = True
log_handle = open(self.log_file, "a", encoding="utf-8")
try:
# 日志交给子进程自己写文件：父进程退出后文件描述符依然有效
process = subprocess.Popen(
self.command(),
stdout=log_handle,
stderr=log_handle,
**kwargs,
)
except OSError as exc:
log_handle.close()
return False, f"启动失败：{exc}"
else:
log_handle.close()
# 日志交给子进程自己写文件：父进程退出后文件描述符依然有效。
# 用 with 管理句柄：Popen 已经把 fd 复制给子进程，父进程这边可以安全关闭。
with open(self.log_file, "a", encoding="utf-8") as log_handle:
try:
process = subprocess.Popen(
self.command(),
stdout=log_handle,
stderr=log_handle,
**kwargs,
)
except OSError as exc:
return False, f"启动失败：{exc}"
# 等它真的能响应，而不是"进程起来了但端口还没监听"
deadline = time.time() + wait
@@ -227,12 +240,10 @@ class DaemonManager:
)
time.sleep(0.25)
return False, (
f"服务器进程已启动但 {wait:.0f} 秒内没响应。请查看日志：{self.log_file}"
)
return False, (f"服务器进程已启动但 {wait:.0f} 秒内没响应。请查看日志：{self.log_file}")
# ------------------------------------------------------------ 停止
def stop(self, force: bool = False) -> Tuple[bool, str]:
def stop(self, force: bool = False) -> tuple[bool, str]:
pid = self.current()
online = self.probe() is not None
if pid is None and not online:
@@ -261,16 +272,15 @@ class DaemonManager:
return True, f"服务器已停止（pid {pid}）"
if force and os.name == "nt":
subprocess.run(["taskkill", "/F", "/T", "/PID", str(pid)],
capture_output=True, check=False)
subprocess.run(
["taskkill", "/F", "/T", "/PID", str(pid)], capture_output=True, check=False
)
if self._wait_offline(pid, 3.0):
self.clear_pid()
return True, f"服务器已强制结束（pid {pid}）"
elif force:
try:
with contextlib.suppress(OSError):
os.kill(pid, signal.SIGKILL)
except OSError:
pass
if self._wait_offline(pid, 3.0):
self.clear_pid()
return True, f"服务器已强制结束（pid {pid}）"
@@ -281,7 +291,7 @@ class DaemonManager:
+ f" 日志：{self.log_file}"
)
def restart(self) -> Tuple[bool, str]:
def restart(self) -> tuple[bool, str]:
stopped, message = self.stop()
if not stopped:
return False, message
@@ -291,8 +301,9 @@ class DaemonManager:
url = f"http://{self.host}:{self.port}/api/admin/shutdown"
if self.token:
url += "?token=" + urllib.parse.quote(self.token)
request = urllib.request.Request(url, data=b"{}", method="POST",
headers={"Content-Type": "application/json"})
request = urllib.request.Request(
url, data=b"{}", method="POST", headers={"Content-Type": "application/json"}
)
try:
with net.open_url(request, timeout=3.0) as response:
return response.status == 200
@@ -308,7 +319,7 @@ class DaemonManager:
time.sleep(0.2)
return not pid_alive(pid)
def _wait_offline(self, pid: Optional[int], seconds: float) -> bool:
def _wait_offline(self, pid: int | None, seconds: float) -> bool:
"""等到进程消失 **且** 端口不再响应。
两件事都要等：进程可能已经死了但端口还被 TIME_WAIT 占着，
@@ -324,10 +335,10 @@ class DaemonManager:
return process_gone and self.probe(timeout=0.6) is None
# ------------------------------------------------------------ 状态
def status(self) -> Dict[str, Any]:
def status(self) -> dict[str, Any]:
pid = self.current()
health = self.probe()
info: Dict[str, Any] = {
info: dict[str, Any] = {
"pid": pid,
"running": pid is not None or health is not None,
"host": self.host,
@@ -343,9 +354,9 @@ class DaemonManager:
info["uptime"] = health.get("uptime")
return info
def tail(self, lines: int = 40) -> List[str]:
def tail(self, lines: int = 40) -> list[str]:
try:
with open(self.log_file, "r", encoding="utf-8", errors="replace") as handle:
with open(self.log_file, encoding="utf-8", errors="replace") as handle:
return handle.read().splitlines()[-lines:]
except OSError:
return []
farm/server/hub.py
+44 -29
View File
@@ -22,14 +22,15 @@ uvicorn 跑在 asyncio 事件循环里，而游戏引擎是同步的、跑在工
from __future__ import annotations
import asyncio
import contextlib
import itertools
import threading
import time
from collections import defaultdict
from typing import Any, Dict, List, Optional, Set
from typing import Any
STATE_INTERVAL = 0.25 # 状态推送的最小间隔（秒），天然把高频操作合并成一次
QUEUE_SIZE = 32 # 单连接出口缓冲区；满了就丢最旧的（状态类消息可丢）
STATE_INTERVAL = 0.25 # 状态推送的最小间隔（秒），天然把高频操作合并成一次
QUEUE_SIZE = 32 # 单连接出口缓冲区；满了就丢最旧的（状态类消息可丢）
_seq_counter = itertools.count(1)
@@ -48,8 +49,17 @@ class Client:
_ids = itertools.count(1)
__slots__ = ("cid", "farm_id", "loop", "queue", "name", "joined_at", "alive",
"_sent", "_dropped")
__slots__ = (
"_dropped",
"_sent",
"alive",
"cid",
"farm_id",
"joined_at",
"loop",
"name",
"queue",
)
def __init__(self, farm_id: str, loop: asyncio.AbstractEventLoop, name: str = "") -> None:
self.cid = f"c{next(Client._ids)}"
@@ -63,7 +73,7 @@ class Client:
self._dropped = 0
# ------------------------------------------------------------ 线程安全投递
def push(self, message: Dict[str, Any]) -> None:
def push(self, message: dict[str, Any]) -> None:
"""从任意线程把消息排队到该连接的写协程。"""
if not self.alive:
return
@@ -73,7 +83,7 @@ class Client:
# 事件循环已关闭（客户端刚好断开）
self.alive = False
def _enqueue(self, message: Dict[str, Any]) -> None:
def _enqueue(self, message: dict[str, Any]) -> None:
if self.queue.full():
# 丢最旧的一条：实时状态宁可少一帧，也不能让连接被拖死
try:
@@ -83,12 +93,12 @@ class Client:
pass
self.queue.put_nowait(message)
async def next_message(self) -> Dict[str, Any]:
async def next_message(self) -> dict[str, Any]:
message = await self.queue.get()
self._sent += 1
return message
def stats(self) -> Dict[str, Any]:
def stats(self) -> dict[str, Any]:
return {
"client_id": self.cid,
"farm": self.farm_id,
@@ -104,8 +114,8 @@ class ConnectionHub:
def __init__(self) -> None:
self._lock = threading.RLock()
self._by_farm: Dict[str, Set[Client]] = defaultdict(set)
self._all: Set[Client] = set()
self._by_farm: dict[str, set[Client]] = defaultdict(set)
self._all: set[Client] = set()
# ------------------------------------------------------------ 增删
def register(self, client: Client) -> None:
@@ -124,7 +134,7 @@ class ConnectionHub:
self._all.discard(client)
# ------------------------------------------------------------ 查询
def clients(self, farm_id: str) -> List[Client]:
def clients(self, farm_id: str) -> list[Client]:
with self._lock:
return list(self._by_farm.get(farm_id, ()))
@@ -132,7 +142,7 @@ class ConnectionHub:
with self._lock:
return len(self._by_farm.get(farm_id, ()))
def online_map(self) -> Dict[str, int]:
def online_map(self) -> dict[str, int]:
with self._lock:
return {farm: len(clients) for farm, clients in self._by_farm.items() if clients}
@@ -140,13 +150,14 @@ class ConnectionHub:
with self._lock:
return len(self._all)
def snapshot(self) -> List[Dict[str, Any]]:
def snapshot(self) -> list[dict[str, Any]]:
with self._lock:
return [c.stats() for c in sorted(self._all, key=lambda c: c.cid)]
# ------------------------------------------------------------ 广播
def broadcast(self, farm_id: str, message: Dict[str, Any],
exclude: Optional[Client] = None) -> int:
def broadcast(
self, farm_id: str, message: dict[str, Any], exclude: Client | None = None
) -> int:
sent = 0
for client in self.clients(farm_id):
if client is exclude:
@@ -155,7 +166,7 @@ class ConnectionHub:
sent += 1
return sent
def broadcast_all(self, message: Dict[str, Any], exclude: Optional[Client] = None) -> int:
def broadcast_all(self, message: dict[str, Any], exclude: Client | None = None) -> int:
with self._lock:
targets = list(self._all)
sent = 0
@@ -207,10 +218,9 @@ class Publisher(threading.Thread):
if now - self._last_autosave >= self.autosave_interval:
self._last_autosave = now
try:
# 自动存档失败不能把推送线程带走
with contextlib.suppress(Exception):
self.sessions.autosave_all()
except Exception:
pass
def _push_state(self, session, now: float) -> None:
payload = session.state_message(seq=next_seq())
@@ -219,13 +229,16 @@ class Publisher(threading.Thread):
day = session.engine.state.day
if day != session.published_day:
session.published_day = day
self.hub.broadcast(session.farm_id, {
"type": "day",
"farm": session.farm_id,
"day": day,
"events": session.last_events,
"at": now,
})
self.hub.broadcast(
session.farm_id,
{
"type": "day",
"farm": session.farm_id,
"day": day,
"events": session.last_events,
"at": now,
},
)
self.hub.broadcast(session.farm_id, payload)
self.pushes += 1
@@ -233,9 +246,11 @@ class Publisher(threading.Thread):
return round(time.time() - self._started_at, 1)
def build_state_message(session, seq: int, note: str = "", extra: Optional[Dict[str, Any]] = None) -> Dict[str, Any]:
def build_state_message(
session, seq: int, note: str = "", extra: dict[str, Any] | None = None
) -> dict[str, Any]:
"""构造给客户端的状态消息（HTTP 与 WS 共用同一份结构）。"""
message: Dict[str, Any] = {
message: dict[str, Any] = {
"type": "state",
"seq": seq,
"farm": session.farm_id,
farm/server/session.py
+29 -26
View File
@@ -15,11 +15,12 @@
from __future__ import annotations
import contextlib
import os
import re
import threading
import time
from typing import Any, Dict, List, Optional, Tuple
from typing import Any
from .. import config
from ..engine import FarmEngine, GameError
@@ -29,7 +30,7 @@ DEFAULT_FARM = "default"
MAX_FARM_ID = 24
# 牧场 ID 会变成存档文件名，所以只允许"安全且可读"的字符
_FARM_ID_RE = re.compile(r"^[A-Za-z0-9_\-\u4e00-\u9fff]{1,%d}$" % MAX_FARM_ID)
_FARM_ID_RE = re.compile(rf"^[A-Za-z0-9_\-\u4e00-\u9fff]{{1,{MAX_FARM_ID}}}$")
def validate_farm_id(raw: Any, default: str = DEFAULT_FARM) -> str:
@@ -63,8 +64,7 @@ def validate_owner(raw: Any) -> str:
class FarmSession:
"""一个独立的农场玩家。"""
def __init__(self, farm_id: str, save_dir: str, tick_seconds: float,
owner: str = "") -> None:
def __init__(self, farm_id: str, save_dir: str, tick_seconds: float, owner: str = "") -> None:
self.farm_id = farm_id
self.owner = validate_owner(owner) or farm_id
self.save_dir = save_dir
@@ -75,7 +75,7 @@ class FarmSession:
# 由 FarmClock 维护，Publisher 读取；用普通属性即可，都是原子读写
self.dirty = True
self.published_day = 0
self.last_events: List[str] = []
self.last_events: list[str] = []
self.created_at = time.time()
self.last_active = time.time()
self.saved_at = 0.0
@@ -122,10 +122,9 @@ class FarmSession:
return False
def shutdown(self) -> None:
try:
# 停机时尽力落盘；写失败也不能挡住进程退出
with contextlib.suppress(OSError):
self.save()
except OSError:
pass
def touch(self, count_action: bool = False) -> None:
self.last_active = time.time()
@@ -133,7 +132,7 @@ class FarmSession:
self.actions_taken += 1
self.dirty = True
def advance_day(self, reason: str = "时间流逝") -> List[str]:
def advance_day(self, reason: str = "时间流逝") -> list[str]:
with self._lock:
events = self.engine.advance_day(reason=reason)
self.last_events = events
@@ -146,14 +145,14 @@ class FarmSession:
self.touch(count_action=True)
# ------------------------------------------------------------ 查询
def snapshot(self) -> Dict[str, Any]:
def snapshot(self) -> dict[str, Any]:
data = self.engine.snapshot()
data["farm"] = self.farm_id
data["owner"] = self.owner
data["tick_seconds"] = self.engine.tick_seconds
return data
def summary(self, online: int = 0) -> Dict[str, Any]:
def summary(self, online: int = 0) -> dict[str, Any]:
"""给小而全的一行摘要：排行榜、`/api/world`、命令行 `status` 都用它。"""
snap = self.engine.snapshot()
return {
@@ -172,7 +171,7 @@ class FarmSession:
"idle_seconds": round(time.time() - self.last_active, 1),
}
def state_message(self, seq: int = 0, note: str = "") -> Dict[str, Any]:
def state_message(self, seq: int = 0, note: str = "") -> dict[str, Any]:
from .hub import build_state_message
return build_state_message(self, seq, note)
@@ -181,14 +180,19 @@ class FarmSession:
class SessionManager:
"""所有牧场的登记处。"""
def __init__(self, save_dir: str, tick_seconds: float = config.DEFAULT_TICK_SECONDS,
max_sessions: int = 64, idle_timeout: float = 1800.0) -> None:
def __init__(
self,
save_dir: str,
tick_seconds: float = config.DEFAULT_TICK_SECONDS,
max_sessions: int = 64,
idle_timeout: float = 1800.0,
) -> None:
self.save_dir = save_dir
self.tick_seconds = tick_seconds
self.max_sessions = max_sessions
self.idle_timeout = idle_timeout
self._lock = threading.RLock()
self._sessions: Dict[str, FarmSession] = {}
self._sessions: dict[str, FarmSession] = {}
os.makedirs(save_dir, exist_ok=True)
# ------------------------------------------------------------ 取用
@@ -203,23 +207,22 @@ class SessionManager:
if len(self._sessions) >= self.max_sessions:
raise GameError(
"too_many_farms",
f"服务器牧场数量已达上限（{self.max_sessions}），"
"请稍后再试或联系管理员扩容",
f"服务器牧场数量已达上限（{self.max_sessions}），请稍后再试或联系管理员扩容",
status=503,
)
session = FarmSession(farm_id, self.save_dir, self.tick_seconds, owner)
self._sessions[farm_id] = session
return session
def find(self, farm_id: str) -> Optional[FarmSession]:
def find(self, farm_id: str) -> FarmSession | None:
with self._lock:
return self._sessions.get(validate_farm_id(farm_id))
def all(self) -> List[FarmSession]:
def all(self) -> list[FarmSession]:
with self._lock:
return list(self._sessions.values())
def ids(self) -> List[str]:
def ids(self) -> list[str]:
with self._lock:
return sorted(self._sessions)
@@ -228,7 +231,7 @@ class SessionManager:
return len(self._sessions)
# ------------------------------------------------------------ 世界视图
def leaderboard(self, online_map: Optional[Dict[str, int]] = None) -> List[Dict[str, Any]]:
def leaderboard(self, online_map: dict[str, int] | None = None) -> list[dict[str, Any]]:
online_map = online_map or {}
rows = [s.summary(online_map.get(s.farm_id, 0)) for s in self.all()]
rows.sort(key=lambda row: (-row["net_worth"], -row["day"], row["farm"]))
@@ -244,22 +247,22 @@ class SessionManager:
saved += 1
return saved
def reap(self, online_map: Optional[Dict[str, int]] = None) -> List[str]:
def reap(self, online_map: dict[str, int] | None = None) -> list[str]:
"""回收长时间无人访问的牧场：先落盘，再从内存里移除。"""
online_map = online_map or {}
now = time.time()
dropped: List[str] = []
dropped: list[str] = []
for session in self.all():
if online_map.get(session.farm_id, 0) > 0:
continue
if session.engine.paused:
continue # 暂停是人为操作，说明玩家还会回来
continue # 暂停是人为操作，说明玩家还会回来
if now - session.last_active < self.idle_timeout:
continue
try:
session.shutdown()
except OSError:
continue # 存不下就别丢，宁可占着内存
continue # 存不下就别丢，宁可占着内存
with self._lock:
self._sessions.pop(session.farm_id, None)
dropped.append(session.farm_id)
@@ -305,5 +308,5 @@ class FarmClock(threading.Thread):
# 单个牧场出错不影响其它牧场的时间
engine.next_tick_at = now + engine.tick_seconds
def stats(self) -> Tuple[int, float]:
def stats(self) -> tuple[int, float]:
return self.ticks, self.accuracy
farm/server/webui.py
+4 -4
View File
@@ -431,7 +431,7 @@ def dashboard_text(host: str, port: int) -> str:
f"curl 'http://{host}:{port}/api/actions?farm=default'\n"
f"curl -X POST 'http://{host}:{port}/api/actions?farm=default' \\\n"
" -H 'Content-Type: application/json' \\\n"
" -d '{\"action\":\"plant\",\"params\":{\"plot\":1,\"crop\":\"wheat\"}}'\n"
' -d \'{"action":"plant","params":{"plot":1,"crop":"wheat"}}\'\n'
"\n"
"读取类\n"
"------\n"
@@ -445,11 +445,11 @@ def dashboard_text(host: str, port: int) -> str:
"\n"
"写入类\n"
"------\n"
"POST /api/actions {\"action\":\"...\",\"params\":{...}} 通用入口\n"
"POST /api/batch {\"actions\":[{...},{...}]} 批量\n"
'POST /api/actions {"action":"...","params":{...}} 通用入口\n'
'POST /api/batch {"actions":[{...},{...}]} 批量\n'
"POST /api/plant|water|harvest|harvest_all|buy_animal|feed|\n"
" collect|buy_feed|buy_plot|sleep|pause|save|reset 便捷入口\n"
"POST /api/farms {\"farm\":\"名字\"} 创建/进入牧场\n"
'POST /api/farms {"farm":"名字"} 创建/进入牧场\n'
"POST /api/admin/save 全部存档\n"
"POST /api/admin/shutdown 优雅停机\n"
)
farm/storage.py
+6 -6
View File
@@ -9,7 +9,7 @@ from __future__ import annotations
import json
import os
import time
from typing import Any, Dict, Optional, Tuple
from typing import Any
from .models import GameState
@@ -22,7 +22,7 @@ def default_save_path() -> str:
return os.path.join(root, "saves", "save.json")
def save_state(state: GameState, path: Optional[str] = None) -> str:
def save_state(state: GameState, path: str | None = None) -> str:
"""把状态写入磁盘，返回实际写入路径。"""
path = path or default_save_path()
directory = os.path.dirname(os.path.abspath(path))
@@ -45,7 +45,7 @@ def save_state(state: GameState, path: Optional[str] = None) -> str:
return path
def _raw_state(state: GameState) -> Dict[str, Any]:
def _raw_state(state: GameState) -> dict[str, Any]:
"""导出可完整还原的原始字段（与 to_dict 的展示字段不同）。"""
return {
"day": state.day,
@@ -85,7 +85,7 @@ def _raw_state(state: GameState) -> Dict[str, Any]:
}
def load_state(path: Optional[str] = None) -> Tuple[Optional[GameState], str]:
def load_state(path: str | None = None) -> tuple[GameState | None, str]:
"""读取存档。
返回 ``(state, message)``；失败时 ``state`` 为 None 并给出原因，
@@ -96,7 +96,7 @@ def load_state(path: Optional[str] = None) -> Tuple[Optional[GameState], str]:
return None, f"未找到存档 {path}"
try:
with open(path, "r", encoding="utf-8") as handle:
with open(path, encoding="utf-8") as handle:
payload = json.load(handle)
except (OSError, json.JSONDecodeError) as exc:
return None, f"存档无法解析：{exc}"
@@ -120,7 +120,7 @@ def load_state(path: Optional[str] = None) -> Tuple[Optional[GameState], str]:
return state, f"已载入存档（第 {state.day} 天）"
def delete_save(path: Optional[str] = None) -> bool:
def delete_save(path: str | None = None) -> bool:
path = path or default_save_path()
try:
os.remove(path)
farm/tui.py
+120 -73
View File
@@ -13,13 +13,14 @@
from __future__ import annotations
import contextlib
import os
import re
import shutil
import sys
import time
import unicodedata
from typing import Any, Dict, List, Optional, Tuple
from typing import Any
from . import config
from .engine import GameError
@@ -41,11 +42,11 @@ C_PURPLE = "\x1b[38;5;176m"
C_BOLD = "\x1b[1m"
ANSI_RE = re.compile(r"\x1b\[[0-9;]*m")
CELL_WIDTH = 17 # 单块农田格子的显示宽度
CELL_WIDTH = 17 # 单块农田格子的显示宽度
CELL_INNER = CELL_WIDTH - 2
SIDE_WIDTH = 36 # 右侧栏宽度
SIDE_WIDTH = 36 # 右侧栏宽度
MIN_TERM_WIDTH = 76
MAX_VISIBLE_ANIMALS = 5 # 侧栏同时最多展示的牲畜数量
MAX_VISIBLE_ANIMALS = 5 # 侧栏同时最多展示的牲畜数量
def strip_ansi(text: str) -> str:
@@ -103,7 +104,7 @@ def bar(filled: int, total: int, width: int, on: str = "█", off: str = "░")
"""生成进度条字符串（不含颜色）。"""
if total <= 0:
total = 1
filled = max(0, min(width, int(round(width * filled / total))))
filled = max(0, min(width, round(width * filled / total)))
return on * filled + off * (width - filled)
@@ -127,14 +128,15 @@ def enable_ansi() -> bool:
return False
def terminal_size() -> Tuple[int, int]:
def terminal_size() -> tuple[int, int]:
size = shutil.get_terminal_size(fallback=(110, 30))
return max(MIN_TERM_WIDTH, size.columns), max(18, size.lines)
# ---------------------------------------------------------------- 绘制工具
def draw_box(title: str, body: List[str], width: int, accent: Optional[str] = None,
use_color: bool = True) -> List[str]:
def draw_box(
title: str, body: list[str], width: int, accent: str | None = None, use_color: bool = True
) -> list[str]:
"""画一个带标题的方框，返回行列表（宽度精确等于 ``width``）。"""
inner = max(0, width - 2)
head = f" {title} "
@@ -144,30 +146,34 @@ def draw_box(title: str, body: List[str], width: int, accent: Optional[str] = No
top = "┌─" + head + "─" * fill_len + "┐"
bottom = "└" + "─" * inner + "┘"
lines = [top]
for row in body:
lines.append("│" + pad(truncate(row, inner), inner) + "│")
lines.extend("│" + pad(truncate(row, inner), inner) + "│" for row in body)
lines.append(bottom)
return lines
def hstack(left: List[str], right: List[str], gap: int = 1) -> List[str]:
def hstack(left: list[str], right: list[str], gap: int = 1) -> list[str]:
"""把两列行按行并排，短的一侧用空格补齐。"""
height = max(len(left), len(right))
left_width = max((display_width(line) for line in left), default=0)
out = []
for i in range(height):
l = left[i] if i < len(left) else ""
r = right[i] if i < len(right) else ""
out.append(pad(l, left_width) + " " * gap + r)
left_line = left[i] if i < len(left) else ""
right_line = right[i] if i < len(right) else ""
out.append(pad(left_line, left_width) + " " * gap + right_line)
return out
class TuiApp:
"""交互式界面主程序。"""
def __init__(self, engine: Any, save_path: Optional[str] = None,
use_color: bool = True, autosave: bool = True,
server_label: str = "") -> None:
def __init__(
self,
engine: Any,
save_path: str | None = None,
use_color: bool = True,
autosave: bool = True,
server_label: str = "",
) -> None:
# engine 既可以是本地 FarmEngine，也可以是远程 RemoteFarm——只要接口一致
self.engine = engine
self.save_path = save_path or default_save_path()
@@ -177,11 +183,11 @@ class TuiApp:
self.reader = KeyReader()
self.running = True
self.mode = "main" # main / help / select_crop / select_animal / confirm_reset
self.mode = "main" # main / help / select_crop / select_animal / confirm_reset
self.plot_cursor = 0
self.animal_cursor = 0
self.grid_cols = 3
self.menu: List[Dict[str, Any]] = []
self.menu: list[dict[str, Any]] = []
self.menu_title = ""
self.pending_action = ""
self.message = ""
@@ -192,20 +198,18 @@ class TuiApp:
# ================================================================ 生命周期
def _enter(self) -> None:
if enable_ansi() and os.name == "nt":
try:
# 个别终端不支持重新配置，失败就按原编码继续
with contextlib.suppress(Exception):
sys.stdout.reconfigure(encoding="utf-8", errors="replace")
except Exception:
pass
sys.stdout.write("\x1b[?25l\x1b[2J\x1b[H")
sys.stdout.write(C_RESET)
sys.stdout.flush()
def _exit(self) -> None:
if self.autosave:
try:
self.engine.save(self.save_path) # 本地引擎写文件；远程引擎让服务器落盘
except Exception:
pass
# 本地引擎写文件；远程引擎让服务器落盘
with contextlib.suppress(Exception):
self.engine.save(self.save_path)
self.reader.close()
sys.stdout.write(C_RESET + "\x1b[?25h\n")
sys.stdout.flush()
@@ -348,7 +352,7 @@ class TuiApp:
self._flash(f"{plot.index + 1} 号地已种了作物，先收获再说", C_RED)
return
items: List[Dict[str, Any]] = []
items: list[dict[str, Any]] = []
catalog = {c["key"]: c for c in self.engine.catalog()["crops"]}
for key, info in config.CROPS.items():
entry = catalog.get(key, {})
@@ -470,7 +474,7 @@ class TuiApp:
middle = hstack(farm_col, side_col, gap=1)
bottom = self._render_bottom(snapshot, width)
lines: List[str] = []
lines: list[str] = []
lines.append(self._render_header(snapshot, width))
lines.extend(self._render_status(snapshot, width))
lines.extend(middle)
@@ -480,15 +484,14 @@ class TuiApp:
lines = lines[: height - 1]
out = ["\x1b[H"]
for line in lines:
out.append(truncate(line, width) + "\x1b[K\r\n")
out.extend(truncate(line, width) + "\x1b[K\r\n" for line in lines)
# 清掉本轮剩余行
out.append("\x1b[J")
sys.stdout.write("".join(out))
sys.stdout.flush()
# ---------------------------------------------------------- header
def _render_header(self, snap: Dict[str, Any], width: int) -> str:
def _render_header(self, snap: dict[str, Any], width: int) -> str:
title = color(" SimpleFarm", C_BOLD + C_GREEN, self.use_color)
if self.server_label:
title += color(f" {self.server_label} ", C_BOLD + C_CYAN, self.use_color)
@@ -514,14 +517,12 @@ class TuiApp:
)
gap = max(1, width - display_width(left) - display_width(right) - 1)
line = left + "─" * gap + right + "╮"
return line
return left + "─" * gap + right + "╮"
# ---------------------------------------------------------- 状态区
def _render_status(self, snap: Dict[str, Any], width: int) -> List[str]:
def _render_status(self, snap: dict[str, Any], width: int) -> list[str]:
net_worth = snap["net_worth"]
goal = snap["next_goal"]
reached = snap["goal_index"]
ap_ratio = snap["ap"] / max(1, snap["max_ap"])
ap_color = C_GREEN if ap_ratio > 0.4 else (C_ORANGE if ap_ratio > 0 else C_RED)
@@ -568,14 +569,14 @@ class TuiApp:
return [pad(line1, width), pad(line2, width)]
# ---------------------------------------------------------- 农田
def _render_farm(self, snap: Dict[str, Any], width: int) -> List[str]:
def _render_farm(self, snap: dict[str, Any], width: int) -> list[str]:
inner = width - 2
# 格子之间留 1 列间隙，避免相邻边框粘成 "││"
cols = max(1, (inner + 1) // (CELL_WIDTH + 1))
self.grid_cols = cols
plots = snap["plots"]
rows: List[str] = []
rows: list[str] = []
for start in range(0, len(plots), cols):
chunk = plots[start : start + cols]
parts = [self._render_cell(cell, width).split("\n") for cell in chunk]
@@ -590,7 +591,7 @@ class TuiApp:
title = f"农田 {snap['plot_count']}/{snap['max_plots']} 块"
return draw_box(title, rows, width, accent=C_GREEN, use_color=self.use_color)
def _render_cell(self, plot: Dict[str, Any], farm_width: int) -> str:
def _render_cell(self, plot: dict[str, Any], farm_width: int) -> str:
selected = plot["index"] == self.plot_cursor
border = C_BOLD + C_GREEN if selected else C_DIM
@@ -611,9 +612,12 @@ class TuiApp:
name_text = color(f" {plot['position']} {info['name']}", C_WHITE, self.use_color)
ratio = plot["growth"] / max(1, plot["grow_days"])
filled_count = max(0, min(10, int(round(ratio * 10))))
progress = color("█" * filled_count + "░" * (10 - filled_count),
C_GREEN if plot["is_ripe"] else C_CYAN, self.use_color)
filled_count = max(0, min(10, round(ratio * 10)))
progress = color(
"█" * filled_count + "░" * (10 - filled_count),
C_GREEN if plot["is_ripe"] else C_CYAN,
self.use_color,
)
water_level = plot["moisture"]
water = color(
@@ -629,30 +633,40 @@ class TuiApp:
top = color("╭" + "─" * CELL_INNER + "╮", border, self.use_color)
bottom = color("╰" + "─" * CELL_INNER + "╯", border, self.use_color)
body = [
color("│", border, self.use_color) + pad(line1, CELL_INNER) + color("│", border, self.use_color),
color("│", border, self.use_color) + pad(line2, CELL_INNER) + color("│", border, self.use_color),
color("│", border, self.use_color)
+ pad(line1, CELL_INNER)
+ color("│", border, self.use_color),
color("│", border, self.use_color)
+ pad(line2, CELL_INNER)
+ color("│", border, self.use_color),
]
return "\n".join([top] + body + [bottom])
return "\n".join([top, *body, bottom])
# ---------------------------------------------------------- 侧栏
def _render_side(self, snap: Dict[str, Any], side_width: int = SIDE_WIDTH) -> List[str]:
def _render_side(self, snap: dict[str, Any], side_width: int = SIDE_WIDTH) -> list[str]:
animals = snap["animals"]
rows: List[str] = []
rows: list[str] = []
if not animals:
rows.append(color(" 牧场空空的，按 B 购买", C_DIM, self.use_color))
rows.append(color(" 鸡/羊/牛需每日喂食才产出", C_DIM, self.use_color))
else:
# 只渲染一个跟随光标的窗口，避免牲畜变多后把界面撑爆
start = max(0, min(max(0, len(animals) - MAX_VISIBLE_ANIMALS),
self.animal_cursor - MAX_VISIBLE_ANIMALS // 2))
start = max(
0,
min(
max(0, len(animals) - MAX_VISIBLE_ANIMALS),
self.animal_cursor - MAX_VISIBLE_ANIMALS // 2,
),
)
for index in range(start, min(len(animals), start + MAX_VISIBLE_ANIMALS)):
animal = animals[index]
focused = index == self.animal_cursor
marker = color("▸", C_BOLD + C_GREEN, self.use_color) if focused else " "
if animal["pending"] > 0:
status = color(f"{animal['product']}x{animal['pending']}",
C_BOLD + C_GOLD, self.use_color)
status = color(
f"{animal['product']}x{animal['pending']}", C_BOLD + C_GOLD, self.use_color
)
elif animal["is_sick"]:
status = color("虚弱!", C_RED, self.use_color)
elif not animal["fed"]:
@@ -662,18 +676,22 @@ class TuiApp:
else:
status = color(f"再{animal['days_to_product']}天", C_GRAY, self.use_color)
head = marker + color(f" #{animal['id']} {animal['name']}",
C_WHITE, self.use_color)
head = marker + color(f" #{animal['id']} {animal['name']}", C_WHITE, self.use_color)
rows.append(pad(head, 13) + status)
hidden = len(animals) - start - MAX_VISIBLE_ANIMALS
if hidden > 0:
rows.append(color(f" …还有 {hidden} 只（Tab 切换）", C_DIM, self.use_color))
pasture = draw_box(f"牧场 {snap['animal_count']}/{snap['max_animals']}", rows,
side_width, accent=C_ORANGE, use_color=self.use_color)
pasture = draw_box(
f"牧场 {snap['animal_count']}/{snap['max_animals']}",
rows,
side_width,
accent=C_ORANGE,
use_color=self.use_color,
)
# ---- 市场行情（中国习惯：涨红跌绿）
market_rows: List[str] = []
market_rows: list[str] = []
name_by_key = {k: v["name"] for k, v in config.CROPS.items()}
ranked = sorted(
((k, v) for k, v in snap["market"].items() if not k.startswith("animal:")),
@@ -694,19 +712,20 @@ class TuiApp:
market_rows.append(
color(f" 饲料 {config.FEED_UNIT_PRICE} 金币/份（按 K 购买）", C_DIM, self.use_color)
)
market = draw_box("市场行情", market_rows, side_width,
accent=C_CYAN, use_color=self.use_color)
market = draw_box(
"市场行情", market_rows, side_width, accent=C_CYAN, use_color=self.use_color
)
return pasture + market
# ---------------------------------------------------------- 底部
def _render_bottom(self, snap: Dict[str, Any], width: int) -> List[str]:
def _render_bottom(self, snap: dict[str, Any], width: int) -> list[str]:
if self.mode == "help":
return self._render_help(width)
if self.mode in ("select_crop", "select_animal"):
return self._render_menu(width)
rows: List[str] = []
rows: list[str] = []
now = time.time()
if self.message and now < self.message_until:
rows.append(color(" " + self.message, self.message_color, self.use_color))
@@ -737,8 +756,8 @@ class TuiApp:
]
return log_box + [pad(h, width) for h in hint]
def _render_menu(self, width: int) -> List[str]:
rows: List[str] = []
def _render_menu(self, width: int) -> list[str]:
rows: list[str] = []
rows.append(color(" " + self.menu_title + "（按 Esc 取消）", C_CYAN, self.use_color))
for i, item in enumerate(self.menu, start=1):
if item["locked"]:
@@ -752,32 +771,60 @@ class TuiApp:
rows.append(label + pad("", max(1, 20 - display_width(label))) + detail)
return draw_box("菜单", rows, width, accent=C_GOLD, use_color=self.use_color)
def _render_help(self, width: int) -> List[str]:
def _render_help(self, width: int) -> list[str]:
rows = [
color(" 玩法目标：靠种地和养殖把净资产做到 10000 金币，解锁全部 5 个里程碑。", C_WHITE, self.use_color),
color(
" 玩法目标：靠种地和养殖把净资产做到 10000 金币，解锁全部 5 个里程碑。",
C_WHITE,
self.use_color,
),
"",
color(" 时间", C_BOLD + C_GOLD, self.use_color)
+ color(" 每 15 秒自动推进一天，作物成长、牲畜产奶、市场价格都在这一天结算。", C_GRAY, self.use_color),
+ color(
" 每 15 秒自动推进一天，作物成长、牲畜产奶、市场价格都在这一天结算。",
C_GRAY,
self.use_color,
),
color(" 水分", C_BOLD + C_GOLD, self.use_color)
+ color(" 作物每天消耗 1 点水分，水为 0 就停止生长。按 W 浇水恢复到 3 点。", C_GRAY, self.use_color),
+ color(
" 作物每天消耗 1 点水分，水为 0 就停止生长。按 W 浇水恢复到 3 点。",
C_GRAY,
self.use_color,
),
color(" 行动点", C_BOLD + C_GOLD, self.use_color)
+ color(" 每天 10 点，播种/浇水/喂食各消耗 1 点，收获和收取免费。", C_GRAY, self.use_color),
+ color(
" 每天 10 点，播种/浇水/喂食各消耗 1 点，收获和收取免费。", C_GRAY, self.use_color
),
color(" 牲畜", C_BOLD + C_GOLD, self.use_color)
+ color(" 必须每天喂食才会产出；连续 3 天没吃东西会离开农场。", C_GRAY, self.use_color),
+ color(
" 必须每天喂食才会产出；连续 3 天没吃东西会离开农场。", C_GRAY, self.use_color
),
color(" 市场", C_BOLD + C_GOLD, self.use_color)
+ color(" 作物和畜产品价格每天波动 ±18%，高点卖出更赚。", C_GRAY, self.use_color),
"",
color(" 快捷键", C_BOLD + C_GOLD, self.use_color)
+ color(" P 播种 / W 浇水 / H 收获 / E 一键收获 / F 喂食 / C 收取产出", C_GRAY, self.use_color),
+ color(
" P 播种 / W 浇水 / H 收获 / E 一键收获 / F 喂食 / C 收取产出",
C_GRAY,
self.use_color,
),
color(" ", C_GRAY, self.use_color)
+ color(" B 买动物 / L 开垦土地 / K 买饲料 / N 休息 / 空格 暂停 / S 存档 / R 重置", C_GRAY, self.use_color),
+ color(
" B 买动物 / L 开垦土地 / K 买饲料 / N 休息 / 空格 暂停 / S 存档 / R 重置",
C_GRAY,
self.use_color,
),
"",
color(" AI 接入", C_BOLD + C_GOLD, self.use_color)
+ color(" 游戏跑在服务器进程里，HTTP 接口与 WebSocket 通道同时开放，",
C_GRAY, self.use_color),
+ color(
" 游戏跑在服务器进程里，HTTP 接口与 WebSocket 通道同时开放，",
C_GRAY,
self.use_color,
),
color(" ", C_GRAY, self.use_color)
+ color(" AI 与你在同一个农场里操作（见 README「给 AI 的接口」）。",
C_GRAY, self.use_color),
+ color(
" AI 与你在同一个农场里操作（见 README「给 AI 的接口」）。", C_GRAY, self.use_color
),
"",
color(" 按任意键返回", C_DIM, self.use_color),
]