snowy-yang
bc3ae96a38
`saves/farm.db` 一张 farms 表，farm_id 主键，**一个牧场一行**。 接口签名不变（save_state / load_state / delete_save 只多了 farm_id 参数）， 调用方只改了路径语义。 设计取舍：标量列 + 整份 payload。 * day/gold/level/ap/weather/forecast/net_worth 单独成列——外部工具或 AI 一条 SELECT 就能排榜、画曲线，不必先把每份存档读进内存； * 完整状态仍整份存 payload JSON：游戏数据隔三差五长新字段，规范化成 几十张表意味着每加字段写一次迁移，而存档只做整体读、整体写， 所以结构化列管可查询，完整性交给 JSON——加字段不用改表结构。 崩溃安全升级：写入包在事务里。原来的"临时文件 + os.replace"能挡进程 崩溃，挡不住写到一半掉电；SQLite 的日志能。读取失败照旧降级开新档。 向后兼容：SAVE_VERSION=2 / MIN_SAVE_VERSION=1 不变。老 JSON 存档由 migrate_json_saves() 在服务器启动时自动导入：同名已存在就跳过（宁可不 迁不覆盖），导入成功原文件改名 *.json.migrated 留底，先写库再改名 保证幂等；坏档一律跳过——迁移不许有能力阻止服务器启动。 异常边界：StorageError = sqlite3.Error，刻意不是 OSError 子类—— 调用方原来的 except OSError 全部跟着改，否则写盘失败会穿透出去。 性能实测：PRAGMA journal_mode=WAL 是持久属性，建库时设一次就够， 每次连接重设白花 2.2ms；save_state 约 6.9ms/次，大头是 synchronous=FULL 的 fsync——不省，存档的价值就是不丢档。bench 热路径无回归。 顺手修一个真 bug：connect() 建表失败时连接会泄漏，Windows 上句柄没关 就删不掉文件，而调用方正是靠"读失败→开新档"容错的。补 close。 测试：StorageTestCase 重写（往返/版本/损坏降级/多牧场隔离/迁移）， test_server 断言 *.json 的用例改查库。513 项全过，ruff/pyright 全绿。 端到端：预置 v1/v2 老档起真服务器验 22 项——自动迁移、字段还原、 list_farms() 排榜、重启后进度分毫不差，全过。
758 lines
33 KiB
Python
758 lines
33 KiB
Python
"""FastAPI 应用：REST + SSE + 网页客户端。
|
||
|
||
接口设计原则
|
||
------------
|
||
1. **一切皆 JSON**：每个响应都带 ``ok`` 字段，出错时带机器可读的 ``error``
|
||
与人类可读的 ``message``，AI 可以直接拿 ``error`` 分支决策。
|
||
2. **状态紧随结果**：每个写操作都会在响应里附一份精简状态（``state``），
|
||
这样 AI 一次请求就能同时"做动作 + 看结果"，省一半往返。
|
||
3. **多牧场隔离**：用 ``farm`` 参数（或 ``X-Farm-Id`` 头）区分玩家，
|
||
每个牧场有自己的引擎、时钟与存档；互不影响。
|
||
4. **实时通道**：``GET /api/stream`` 是 SSE 长连接，只负责**服务器 → 客户端**
|
||
的推送（状态、跨天、有人来串门）；写操作统一走 ``POST /api/actions``。
|
||
两者改的是同一个引擎，所以「网页里点一下，AI 那边立刻看到」。
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import asyncio
|
||
import contextlib
|
||
import json
|
||
import time
|
||
from collections.abc import AsyncIterator, Callable
|
||
from dataclasses import dataclass, field
|
||
from typing import Any
|
||
|
||
from fastapi import FastAPI, Request
|
||
from fastapi.responses import HTMLResponse, JSONResponse, PlainTextResponse, StreamingResponse
|
||
|
||
from .. import config
|
||
from ..engine import GameError
|
||
from ..storage import StorageError
|
||
from . import actions as actions_mod
|
||
from .hub import (
|
||
CLOSE_MESSAGE,
|
||
HEARTBEAT_SECONDS,
|
||
IDLE_TICK,
|
||
Client,
|
||
ConnectionHub,
|
||
Publisher,
|
||
next_seq,
|
||
sse_frame,
|
||
sse_heartbeat,
|
||
)
|
||
from .session import DEFAULT_FARM, FarmClock, FarmSession, SessionManager, validate_farm_id
|
||
from .webui import DASHBOARD_HTML, dashboard_text
|
||
|
||
API_VERSION = "2.0"
|
||
SERVER_NAME = "SimpleFarm"
|
||
|
||
|
||
@dataclass
|
||
class ServerConfig:
|
||
"""服务器配置（由命令行解析而来）。"""
|
||
|
||
host: str = "127.0.0.1"
|
||
port: int = 8765
|
||
token: str | None = None
|
||
tick_seconds: float = config.DEFAULT_TICK_SECONDS
|
||
save_dir: str = "saves"
|
||
max_sessions: int = 64
|
||
idle_timeout: float = 1800.0
|
||
autosave_interval: float = 60.0
|
||
allow_remote_control: bool = True
|
||
world_name: str = "SimpleFarm 公共农场"
|
||
extra: dict[str, Any] = field(default_factory=dict)
|
||
|
||
|
||
class ServerHandle:
|
||
"""让 HTTP 接口能够优雅地要求本进程退出，并暴露进程生命周期的钩子。
|
||
|
||
* ``on_start``：ASGI 生命周期启动时触发（用于写 pid 文件等）；
|
||
* ``on_stop``：请求关闭时触发（uvicorn 用它把 ``should_exit`` 置位，开始优雅退出）。
|
||
"""
|
||
|
||
def __init__(self) -> None:
|
||
self._start_callbacks: list[Callable[[], None]] = []
|
||
self._stop_callbacks: list[Callable[[], None]] = []
|
||
self.stopping = False
|
||
|
||
def on_start(self, callback: Callable[[], None]) -> None:
|
||
self._start_callbacks.append(callback)
|
||
|
||
def on_stop(self, callback: Callable[[], None]) -> None:
|
||
self._stop_callbacks.append(callback)
|
||
|
||
def notify_start(self) -> None:
|
||
for callback in list(self._start_callbacks):
|
||
# 单个启动钩子失败不该拖垮整个服务
|
||
with contextlib.suppress(Exception):
|
||
callback()
|
||
|
||
def request_stop(self) -> bool:
|
||
if self.stopping:
|
||
return False
|
||
self.stopping = True
|
||
for callback in list(self._stop_callbacks):
|
||
# 停机要尽力而为：一个钩子抛异常不能拦住其它钩子
|
||
with contextlib.suppress(Exception):
|
||
callback()
|
||
return True
|
||
|
||
|
||
# ---------------------------------------------------------------------- 工具
|
||
# brief() 从精简状态里挑选的字段白名单：够 AI 做下一步决策，又不至于太吵。
|
||
# 数据源是 FarmEngine.brief_source()（**不是**完整快照），加字段时两处一起改。
|
||
BRIEF_KEYS = (
|
||
"day",
|
||
"weather",
|
||
"weather_name",
|
||
"forecast",
|
||
"forecast_name",
|
||
"gold",
|
||
"level",
|
||
"exp",
|
||
"exp_needed",
|
||
"ap",
|
||
"max_ap",
|
||
"feed",
|
||
"plot_count",
|
||
"max_plots",
|
||
"next_plot_price",
|
||
"animal_count",
|
||
"max_animals",
|
||
"storage_used",
|
||
"granary_capacity",
|
||
"granary_free",
|
||
"building_count",
|
||
"net_worth",
|
||
"goal_index",
|
||
"seconds_to_tick",
|
||
"paused",
|
||
)
|
||
|
||
# 特意为 AI 准备的待办清单：不用遍历整个快照就知道该干什么
|
||
BRIEF_TODO_KEYS = ("ripe_plots", "hungry_animals", "collectable_animals")
|
||
|
||
|
||
def brief(session: FarmSession, hub: ConnectionHub) -> dict[str, Any]:
|
||
"""操作后回给调用者的精简状态。
|
||
|
||
只读引擎直接给出的那十几个标量，**不做完整快照**。以前这里是
|
||
``session.snapshot()`` 之后再挑字段——每块地、每只牲畜都白白序列化一遍，
|
||
实测每次动作浪费约 60µs；而 ``full=1`` 时后面还要再序列化一次给快照用。
|
||
"""
|
||
source = session.engine.brief_source()
|
||
data: dict[str, Any] = {"farm": session.farm_id}
|
||
data.update({key: source[key] for key in BRIEF_KEYS})
|
||
data.update({key: source[key] for key in BRIEF_TODO_KEYS})
|
||
data["online"] = hub.online(session.farm_id)
|
||
return data
|
||
|
||
|
||
def rules_text() -> str:
|
||
return (
|
||
f"{SERVER_NAME} 规则速览（给 AI 玩家的上下文）\n"
|
||
"==========================================\n"
|
||
f"1. 时间：每个牧场每 {config.DEFAULT_TICK_SECONDS / 60:.0f} 分钟"
|
||
f"（{config.DEFAULT_TICK_SECONDS:.0f} 秒）自动推进一天，可配置。"
|
||
f"每天开始时行动点恢复为 {config.START_AP} 点，作物生长、牲畜产出、市场价格都在此刻结算。\n"
|
||
f"2. 行动点：播种/浇水/喂食各消耗 1 点，收获与收取产出免费。行动点用完只能等第二天——"
|
||
"时间只能自然流动，没有任何「手动推进一天」的接口，别白费请求。\n"
|
||
f"3. 天气与水分：每天随机一种天气（晴/多云/雨/干旱/暴风雨）。作物每天按天气蒸发"
|
||
f" 0~2 点水分（上限 {config.WATER_CAPACITY}），水分为 0 时停止生长；"
|
||
f"播种自带 {config.PLANT_INITIAL_MOISTURE} 点。雨天与暴风雨会自动补满水分，"
|
||
"暴风雨还可能毁掉未成熟的作物（温室可免疫）。state.forecast 是明天的预报——"
|
||
"提前一天浇水就能避免断水停长。\n"
|
||
"4. 收获与谷仓：harvest / collect 的产出**先进谷仓，不直接换钱**，再用 sell "
|
||
"按当时的市价卖出。仓容 = 基础 "
|
||
f"{config.GRANARY_BASE} + 等级成长 + 谷仓建筑，装不下时收获会被拒绝"
|
||
"（granary_full），先卖货或者建谷仓。\n"
|
||
"5. 牲畜：每天必须喂食才会累计产出周期；连续 3 天未进食会永久离开农场。\n"
|
||
f"6. 市场：作物与畜产品价格每天在基础价 ±{config.PRICE_VOLATILITY:.0%} 内波动。"
|
||
'有了谷仓之后，"什么时候卖"才真正影响收益：囤到高位再 sell。\n'
|
||
f"7. 土地：初始 {config.BASE_PLOTS} 块，每升 1 级多 {config.PLOTS_PER_LEVEL} 块，"
|
||
f"上限 {config.MAX_PLOTS} 块；购地价格每次递增。\n"
|
||
"8. 建筑：build 建造 granary / well / feeder / greenhouse。谷仓加仓容、"
|
||
"水井每天自动浇 1 块干透的地、喂食器每天自动喂 1 只牲畜（只吃库存饲料，"
|
||
"不会替你花钱）、温室免疫暴风雨且水分蒸发减半。同类建筑越造越贵。\n"
|
||
f"9. 目标：把净资产（金币 + 土地 + 牲畜 + 饲料 + 地里作物 + **谷仓存货**）"
|
||
f"做到 {config.GOALS[-1][0]} 金币。\n"
|
||
"10. 多人：每个 farm 是一个独立牧场，用 ?farm=名字 区分，互不影响；"
|
||
"排行榜见 GET /api/world。\n"
|
||
"\n推荐决策循环（AI 最短路径）：\n"
|
||
" GET /api/actions?farm=me → 拿到当前所有**合法**动作（含开销与理由）\n"
|
||
' POST /api/actions → {"action":"<上一步里的 action>",'
|
||
'"params":{...}} 直接执行\n'
|
||
" 响应里的 state 就是执行后的最新状态，无需再查一次。\n"
|
||
"\n实时通道（单向推送，服务器 → 客户端）：GET /api/stream?farm=me 是一条 SSE "
|
||
'长连接，服务器在状态变化时推 {"type":"state",...}；\n'
|
||
"想主动拉一次用 GET /api/state；写操作一律 POST /api/actions。\n"
|
||
)
|
||
|
||
|
||
def as_asgi(app: FastAPI) -> Any:
|
||
"""把应用交给 ASGI 服务器时的类型桥。
|
||
|
||
uvicorn 用它自己那份 ``uvicorn._types.Scope`` 描述 ASGI，starlette（FastAPI）
|
||
用的是 ``starlette.types.Scope``。两者运行期逐字一致，但都是 TypedDict、
|
||
彼此没有子类型关系，所以 pyright 会拦。返回值声明成 Any 就是明说：
|
||
这个转换由我们担保，问题只在两份类型声明的重复上。
|
||
"""
|
||
return app
|
||
|
||
|
||
# ---------------------------------------------------------------------- 运行时
|
||
@dataclass
|
||
class _Services:
|
||
"""create_app 装配出来的运行时依赖。
|
||
|
||
以前这些是 ``create_app`` 的局部变量，于是每个路由都必须是嵌套闭包才能看见
|
||
它们——一个函数膨胀到 475 行、8 层嵌套。收进一个对象后，路由可以拆成模块级
|
||
的注册函数，各自只依赖传进来的 ``svc``，读起来是平的。
|
||
"""
|
||
|
||
cfg: ServerConfig
|
||
manager: SessionManager
|
||
hub: ConnectionHub
|
||
publisher: Publisher
|
||
clock: FarmClock
|
||
handle: ServerHandle
|
||
started_at: float
|
||
counters: dict[str, int]
|
||
|
||
@classmethod
|
||
def create(cls, cfg: ServerConfig) -> _Services:
|
||
manager = SessionManager(
|
||
save_dir=cfg.save_dir,
|
||
tick_seconds=cfg.tick_seconds,
|
||
max_sessions=cfg.max_sessions,
|
||
idle_timeout=cfg.idle_timeout,
|
||
)
|
||
hub = ConnectionHub()
|
||
return cls(
|
||
cfg=cfg,
|
||
manager=manager,
|
||
hub=hub,
|
||
publisher=Publisher(manager, hub, autosave_interval=cfg.autosave_interval),
|
||
clock=FarmClock(manager),
|
||
handle=ServerHandle(),
|
||
started_at=time.time(),
|
||
# 这里没有 ws_messages：那是 WebSocket 时代的遗留计数，
|
||
# 改成 SSE 之后再也没有人加过它，留着只会让人以为它在被用。
|
||
counters={"requests": 0, "actions": 0, "errors": 0},
|
||
)
|
||
|
||
@contextlib.asynccontextmanager
|
||
async def lifespan(self, _app: FastAPI) -> AsyncIterator[None]:
|
||
"""应用生命周期：起时钟与推送线程，退出时按相反顺序收干净。
|
||
|
||
装饰成 ``asynccontextmanager`` 是因为 FastAPI 要的就是"能当异步上下文
|
||
管理器用"的东西，裸 async generator 在类型上并不匹配。
|
||
"""
|
||
self.clock.start()
|
||
self.publisher.start()
|
||
# 生命周期启动完成，可以对外宣告"我在了"。uvicorn 的顺序是"先跑 lifespan、
|
||
# 后 bind 端口"，所以这里代表"服务已上线"，不代表"端口已监听"；
|
||
# bind 失败时 pid 文件由 __main__ 的 finally 负责清掉。
|
||
self.handle.notify_start()
|
||
try:
|
||
yield
|
||
finally:
|
||
self.publisher.stop()
|
||
# 兜底：万一是信号触发退出（没走 /api/admin/shutdown），也把长连接收掉
|
||
self.hub.close_all()
|
||
self.clock.stop()
|
||
self.manager.shutdown()
|
||
|
||
# ------------------------------------------------------------ 鉴权与取牧场
|
||
def authorized(self, supplied: str | None) -> bool:
|
||
# 没配 token 就是谁都能进（本机自娱自乐的默认形态）
|
||
return not self.cfg.token or supplied == self.cfg.token
|
||
|
||
@staticmethod
|
||
def token_of(request: Request) -> str | None:
|
||
return request.headers.get("x-api-token") or request.query_params.get("token")
|
||
|
||
def guard(self, request: Request) -> None:
|
||
"""计数 + 鉴权。所有对外路由的第一句都调它。"""
|
||
self.counters["requests"] += 1
|
||
if not self.authorized(self.token_of(request)):
|
||
raise GameError("unauthorized", "缺少或错误的 API Token", status=401)
|
||
|
||
def session_of(self, request: Request, owner: str = "") -> FarmSession:
|
||
farm = request.headers.get("x-farm-id") or request.query_params.get("farm") or DEFAULT_FARM
|
||
return self.manager.get(farm, create=True, owner=owner)
|
||
|
||
@staticmethod
|
||
def full_requested(request: Request) -> bool:
|
||
raw = (request.query_params.get("full") or "").lower()
|
||
return raw in ("1", "true", "yes", "on")
|
||
|
||
@staticmethod
|
||
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}") from exc
|
||
if not isinstance(data, dict):
|
||
raise GameError("bad_json", "请求体必须是 JSON 对象")
|
||
return data
|
||
|
||
# ------------------------------------------------------------ 动作执行
|
||
def execute_action(self, session: FarmSession, name: Any, params: Any) -> dict[str, Any]:
|
||
"""执行一个动作，返回引擎/会话级的结果字典。
|
||
|
||
``save`` 与 ``reset`` 是**存档类动作**，刻意不在 farm.server.actions
|
||
的通用动作表里（那张表只认识引擎方法，而 ``engine.save()`` 只会写到
|
||
默认存档路径，多牧场环境下会把别人的存档覆盖掉）。它们只能走这里、
|
||
由牧场会话处理；其余全部走统一的动作表。
|
||
"""
|
||
key = name.strip().lower() if isinstance(name, str) else ""
|
||
if key == "save":
|
||
path = session.save()
|
||
session.engine.add_log(f"已存档到 {path}", "info")
|
||
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 actions_mod.run_action(session.engine, name, params)
|
||
|
||
def run_and_wrap(
|
||
self, session: FarmSession, name: Any, params: Any, full: bool = False
|
||
) -> dict[str, Any]:
|
||
"""执行动作并拼装统一响应。HTTP 各入口共用。"""
|
||
result = self.execute_action(session, name, params)
|
||
self.counters["actions"] += 1
|
||
session.touch(count_action=True)
|
||
payload: dict[str, Any] = dict(result)
|
||
payload["ok"] = True
|
||
payload["farm"] = session.farm_id
|
||
payload["seq"] = next_seq()
|
||
payload["state"] = brief(session, self.hub)
|
||
payload["log"] = session.engine.recent_log(limit=3)
|
||
if full:
|
||
payload["snapshot"] = session.snapshot()
|
||
return payload
|
||
|
||
|
||
# ---------------------------------------------------------------------- 路由分组
|
||
# 每个 _register_* 负责一组关系紧密的接口。分组边界照 OpenAPI 的 tags 来切，
|
||
# 所以"接口文档里在哪组"和"代码里在哪段"是对得上的。
|
||
def _register_error_handlers(app: FastAPI, svc: _Services) -> None:
|
||
@app.exception_handler(GameError)
|
||
async def _game_error(_request: Request, exc: GameError) -> JSONResponse:
|
||
svc.counters["errors"] += 1
|
||
return JSONResponse(exc.to_dict(), status_code=exc.status)
|
||
|
||
@app.exception_handler(json.JSONDecodeError)
|
||
async def _bad_json(_request: Request, exc: json.JSONDecodeError) -> JSONResponse:
|
||
svc.counters["errors"] += 1
|
||
return JSONResponse(
|
||
{"ok": False, "error": "bad_json", "message": f"请求体不是合法 JSON：{exc}"},
|
||
status_code=400,
|
||
)
|
||
|
||
|
||
def _register_pages(app: FastAPI, svc: _Services) -> None:
|
||
@app.get("/", response_class=HTMLResponse, include_in_schema=False)
|
||
async def index() -> HTMLResponse:
|
||
return HTMLResponse(DASHBOARD_HTML)
|
||
|
||
@app.get("/text", response_class=PlainTextResponse, include_in_schema=False)
|
||
async def text_index() -> PlainTextResponse:
|
||
return PlainTextResponse(dashboard_text(svc.cfg.host, svc.cfg.port))
|
||
|
||
|
||
def _register_meta_routes(app: FastAPI, svc: _Services) -> None:
|
||
@app.get("/api/health", tags=["元信息"], summary="健康检查（无需鉴权）")
|
||
async def health() -> dict[str, Any]:
|
||
# 刻意不鉴权：探针要能在没有 token 的情况下判断"这个服务还活着吗"
|
||
return {
|
||
"ok": True,
|
||
"service": SERVER_NAME,
|
||
"version": API_VERSION,
|
||
"uptime": round(time.time() - svc.started_at, 1),
|
||
"farms": svc.manager.count(),
|
||
"clients": svc.hub.total(),
|
||
"clock_ticks": svc.clock.ticks,
|
||
"saves_since_start": svc.publisher.pushes,
|
||
"publish_errors": svc.publisher.errors,
|
||
}
|
||
|
||
@app.get("/api/help", tags=["元信息"], summary="玩法规则与接口说明（给 AI 的说明书）")
|
||
async def help_page(request: Request) -> dict[str, Any]:
|
||
svc.guard(request)
|
||
return {
|
||
"ok": True,
|
||
"service": SERVER_NAME,
|
||
"version": API_VERSION,
|
||
"rules": rules_text(),
|
||
"actions": actions_mod.describe_actions(),
|
||
"crops": config.CROPS,
|
||
"animals": config.ANIMALS,
|
||
"buildings": config.BUILDINGS,
|
||
"weather": config.WEATHER,
|
||
"endpoints": {
|
||
"state": "GET /api/state?farm=NAME",
|
||
"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"}',
|
||
"realtime": "GET /api/stream?farm=NAME（SSE 单向推送）",
|
||
},
|
||
}
|
||
|
||
|
||
def _register_world_routes(app: FastAPI, svc: _Services) -> None:
|
||
@app.get("/api/world", tags=["世界"], summary="世界概览：排行榜 + 在线连接 + 时钟")
|
||
async def world(request: Request) -> dict[str, Any]:
|
||
svc.guard(request)
|
||
return {
|
||
"ok": True,
|
||
"world": svc.cfg.world_name,
|
||
"uptime": round(time.time() - svc.started_at, 1),
|
||
"farms": svc.manager.count(),
|
||
"clients": svc.hub.total(),
|
||
"connections": svc.hub.snapshot(),
|
||
"leaderboard": svc.manager.leaderboard(svc.hub.online_map()),
|
||
"tick_seconds": svc.cfg.tick_seconds,
|
||
"started_at": svc.started_at,
|
||
}
|
||
|
||
@app.get("/api/farms", tags=["世界"], summary="牧场排行榜")
|
||
async def list_farms(request: Request) -> dict[str, Any]:
|
||
svc.guard(request)
|
||
rows = svc.manager.leaderboard(svc.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]:
|
||
svc.guard(request)
|
||
body = await svc.body_of(request)
|
||
farm = validate_farm_id(body.get("farm") or body.get("name"))
|
||
owner = str(body.get("owner") or "")
|
||
existed = svc.manager.find(farm) is not None
|
||
session = svc.manager.get(farm, create=True, owner=owner)
|
||
tick = body.get("tick")
|
||
if tick is not None:
|
||
try:
|
||
tick_value = float(tick)
|
||
except (TypeError, ValueError):
|
||
raise GameError("bad_param", "tick 必须是数字") from None
|
||
# 直接赋值绕过了引擎构造时的夹取，所以这里得自己夹一次
|
||
session.engine.tick_seconds = max(
|
||
config.MIN_TICK_SECONDS, min(config.MAX_TICK_SECONDS, tick_value)
|
||
)
|
||
session.engine.next_tick_at = time.time() + session.engine.tick_seconds
|
||
return {
|
||
"ok": True,
|
||
"created": not existed,
|
||
"message": ("牧场已创建" if not existed else "牧场已在运行") + f"：{farm}",
|
||
"farm": farm,
|
||
"summary": session.summary(svc.hub.online(farm)),
|
||
"state": brief(session, svc.hub),
|
||
}
|
||
|
||
|
||
def _register_farm_routes(app: FastAPI, svc: _Services) -> None:
|
||
@app.get("/api/state", tags=["牧场"], summary="完整状态快照")
|
||
async def get_state(request: Request) -> dict[str, Any]:
|
||
svc.guard(request)
|
||
session = svc.session_of(request)
|
||
return {
|
||
"ok": True,
|
||
"farm": session.farm_id,
|
||
"state": session.snapshot(),
|
||
"online": svc.hub.online(session.farm_id),
|
||
}
|
||
|
||
@app.get("/api/catalog", tags=["牧场"], summary="作物/牲畜目录与实时价格")
|
||
async def get_catalog(request: Request) -> dict[str, Any]:
|
||
svc.guard(request)
|
||
session = svc.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]:
|
||
svc.guard(request)
|
||
session = svc.session_of(request)
|
||
items = session.engine.available_actions()
|
||
return {
|
||
"ok": True,
|
||
"farm": session.farm_id,
|
||
"count": len(items),
|
||
"actions": items,
|
||
"state": brief(session, svc.hub),
|
||
}
|
||
|
||
@app.get("/api/log", tags=["牧场"], summary="最近事件")
|
||
async def get_log(request: Request) -> dict[str, Any]:
|
||
svc.guard(request)
|
||
session = svc.session_of(request)
|
||
limit = _int(request.query_params.get("limit"), 20, 1, 200)
|
||
since = _int(request.query_params.get("since"), 0, 0, 10**9)
|
||
entries = session.engine.recent_log(limit=limit, since=since)
|
||
return {
|
||
"ok": True,
|
||
"farm": session.farm_id,
|
||
"count": len(entries),
|
||
"entries": entries,
|
||
"last_seq": entries[-1]["seq"] if entries else since,
|
||
}
|
||
|
||
|
||
def _register_action_routes(app: FastAPI, svc: _Services) -> None:
|
||
@app.post("/api/actions", tags=["动作"], summary="通用动作入口（推荐 AI 使用）")
|
||
async def post_action(request: Request) -> dict[str, Any]:
|
||
svc.guard(request)
|
||
body = await svc.body_of(request)
|
||
session = svc.session_of(request)
|
||
name = body.get("action") or body.get("name")
|
||
params = body.get("params")
|
||
if params is None:
|
||
# 允许把参数平铺在顶层：{"action":"plant","plot":1,"crop":"wheat"}
|
||
params = {k: v for k, v in body.items() if k not in ("action", "name", "params")}
|
||
return svc.run_and_wrap(session, name, params, svc.full_requested(request))
|
||
|
||
@app.post("/api/batch", tags=["动作"], summary="批量执行动作（一个请求干多件事）")
|
||
async def post_batch(request: Request) -> dict[str, Any]:
|
||
"""给 AI 的省流入口：一次提交多个动作，按顺序执行。
|
||
|
||
每个动作独立成败，失败的只记错误不中断后续动作。
|
||
"""
|
||
svc.guard(request)
|
||
body = await svc.body_of(request)
|
||
session = svc.session_of(request)
|
||
items = body.get("actions")
|
||
if not isinstance(items, list) or not items:
|
||
raise GameError("bad_param", "actions 必须是至少含一个元素的数组")
|
||
if len(items) > 50:
|
||
raise GameError("too_many", "单次批量最多 50 个动作")
|
||
results: list[dict[str, Any]] = []
|
||
for item in items:
|
||
if not isinstance(item, dict):
|
||
results.append({"ok": False, "error": "bad_param", "message": "每个动作必须是对象"})
|
||
continue
|
||
try:
|
||
payload = svc.run_and_wrap(
|
||
session, item.get("action"), item.get("params") or {}, full=False
|
||
)
|
||
# 逐个动作不必重复回精简状态：末尾统一给一份就够了
|
||
results.append({k: v for k, v in payload.items() if k != "state"})
|
||
except GameError as exc:
|
||
svc.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] = {
|
||
"ok": True,
|
||
"farm": session.farm_id,
|
||
"done": ok_count,
|
||
"failed": len(results) - ok_count,
|
||
"results": results,
|
||
"state": brief(session, svc.hub),
|
||
}
|
||
if svc.full_requested(request):
|
||
payload["snapshot"] = session.snapshot()
|
||
return payload
|
||
|
||
def make_action_route(action_name: str) -> Callable[[Request], Any]:
|
||
async def route(request: Request) -> dict[str, Any]:
|
||
svc.guard(request)
|
||
body = await svc.body_of(request)
|
||
session = svc.session_of(request)
|
||
return svc.run_and_wrap(session, action_name, body, svc.full_requested(request))
|
||
|
||
route.__name__ = f"post_{action_name}"
|
||
return route
|
||
|
||
# 每个动作都额外挂一条自己的路径（POST /api/water 之类），方便直接 curl；
|
||
# 语义与 POST /api/actions 完全等价。
|
||
for _name, _spec in actions_mod.ACTIONS.items():
|
||
app.add_api_route(
|
||
f"/api/{_name}",
|
||
make_action_route(_name),
|
||
methods=["POST"],
|
||
name=f"action_{_name}",
|
||
tags=["动作"],
|
||
summary=f"{_spec.summary}（等价于 POST /api/actions）",
|
||
)
|
||
|
||
|
||
def _register_admin_routes(app: FastAPI, svc: _Services) -> None:
|
||
@app.post("/api/admin/save", tags=["管理"], summary="把所有牧场写入存档")
|
||
async def admin_save(request: Request) -> dict[str, Any]:
|
||
svc.guard(request)
|
||
saved: list[str] = []
|
||
for session in svc.manager.all():
|
||
try:
|
||
session.save()
|
||
saved.append(session.farm_id)
|
||
except (OSError, StorageError) as exc:
|
||
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]:
|
||
svc.guard(request)
|
||
if not svc.cfg.allow_remote_control:
|
||
raise GameError("forbidden", "服务器未开放远程停机", status=403)
|
||
svc.manager.shutdown()
|
||
# 先让所有 SSE 长连接收尾：否则服务器的优雅停机要一直等它们，变成卡死
|
||
svc.hub.close_all()
|
||
accepted = svc.handle.request_stop()
|
||
return {
|
||
"ok": True,
|
||
"message": "服务器正在优雅停机，存档已写入" if accepted else "停机请求已受理（重复）",
|
||
}
|
||
|
||
|
||
def _register_stream_route(app: FastAPI, svc: _Services) -> None:
|
||
@app.get("/api/stream", tags=["实时"], summary="SSE 实时推送（服务器 → 客户端单向）")
|
||
async def stream(request: Request) -> StreamingResponse:
|
||
"""一条 HTTP 长连接持续推送状态；写操作请走 POST /api/actions。
|
||
|
||
消息类型：``welcome``（首帧全量）/ ``state``（状态变化）/ ``day``（跨天）/
|
||
``notice``（有人来去）/ ``error``，每条都是一行 JSON 放在 ``data:`` 后面。
|
||
客户端断开（关页面、退出 TUI、进程被杀）都会走到生成器的 finally。
|
||
"""
|
||
svc.guard(request)
|
||
params = request.query_params
|
||
farm = validate_farm_id(params.get("farm") or DEFAULT_FARM)
|
||
# 牧场数达上限等情况会抛 GameError，交给全局处理器回 4xx JSON
|
||
session = svc.manager.get(farm, create=True)
|
||
|
||
loop = asyncio.get_running_loop()
|
||
client = Client(farm, loop, name=(params.get("name") or "")[:32])
|
||
svc.hub.register(client)
|
||
session.touch()
|
||
|
||
# "来了/走了"是同一件事的两个方向，差别只有动词，所以共用一个信封
|
||
who = client.name or client.cid
|
||
|
||
def notice(text: str, exclude: Client | None = None) -> None:
|
||
svc.hub.broadcast(farm, {"type": "notice", "farm": farm, "text": text}, exclude=exclude)
|
||
|
||
async def events() -> AsyncIterator[str]:
|
||
# 首帧等价于原来的 welcome：一次把状态、目录、合法动作、日志给全
|
||
yield sse_frame(
|
||
{
|
||
"type": "welcome",
|
||
"client_id": client.cid,
|
||
"farm": farm,
|
||
"server": {
|
||
"name": SERVER_NAME,
|
||
"version": API_VERSION,
|
||
"world": svc.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": svc.hub.online(farm),
|
||
}
|
||
)
|
||
notice(
|
||
f"{who} 来到了牧场（在线 {svc.hub.online(farm)}）",
|
||
exclude=client,
|
||
)
|
||
try:
|
||
last_beat = time.time()
|
||
while True:
|
||
if svc.handle.stopping:
|
||
# 服务器要退了：主动收尾，别让长连接拖住优雅停机
|
||
yield sse_frame(dict(CLOSE_MESSAGE))
|
||
break
|
||
try:
|
||
message = await asyncio.wait_for(client.next_message(), timeout=IDLE_TICK)
|
||
except asyncio.TimeoutError:
|
||
# 空转一秒只是为了让上面的停机检查有机会跑；
|
||
# 心跳按自己的间隔发，别刷屏
|
||
if time.time() - last_beat >= HEARTBEAT_SECONDS:
|
||
last_beat = time.time()
|
||
yield sse_heartbeat()
|
||
continue
|
||
if message.get("type") == "bye":
|
||
yield sse_frame(message)
|
||
break
|
||
yield sse_frame(message)
|
||
finally:
|
||
svc.hub.unregister(client)
|
||
session.touch()
|
||
notice(f"{who} 离开了牧场（在线 {svc.hub.online(farm)}）")
|
||
|
||
return StreamingResponse(
|
||
events(),
|
||
media_type="text/event-stream",
|
||
headers={
|
||
"Cache-Control": "no-cache",
|
||
"Connection": "keep-alive",
|
||
# 让 nginx 之类的反代不要缓冲，否则推送会攒着一次性发出来
|
||
"X-Accel-Buffering": "no",
|
||
},
|
||
)
|
||
|
||
|
||
# ---------------------------------------------------------------------- 应用
|
||
def create_app(cfg: ServerConfig | None = None) -> FastAPI:
|
||
"""装配应用：依赖 → 生命周期 → 各路由分组。
|
||
|
||
路由各自藏在 ``_register_*`` 里，所以这里读起来就是一份目录：
|
||
想找哪个接口，按分组名翻就行，不必在几百行里爬。
|
||
"""
|
||
cfg = cfg or ServerConfig()
|
||
svc = _Services.create(cfg)
|
||
|
||
app = FastAPI(
|
||
title=f"{SERVER_NAME} 游戏服务器",
|
||
version=API_VERSION,
|
||
description="命令行农场游戏的服务端：HTTP 给 AI 用，SSE 给实时界面用。",
|
||
lifespan=svc.lifespan,
|
||
)
|
||
app.state.cfg = cfg
|
||
app.state.manager = svc.manager
|
||
app.state.hub = svc.hub
|
||
app.state.publisher = svc.publisher
|
||
app.state.clock = svc.clock
|
||
app.state.handle = svc.handle
|
||
app.state.started_at = svc.started_at
|
||
app.state.counters = svc.counters
|
||
|
||
_register_error_handlers(app, svc)
|
||
_register_pages(app, svc)
|
||
_register_meta_routes(app, svc)
|
||
_register_world_routes(app, svc)
|
||
_register_farm_routes(app, svc)
|
||
_register_action_routes(app, svc)
|
||
_register_admin_routes(app, svc)
|
||
_register_stream_route(app, svc)
|
||
return app
|
||
|
||
|
||
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}") from None
|
||
return max(low, min(high, value))
|
||
|
||
|
||
def build_app(host: str = "127.0.0.1", port: int = 8765, **kwargs: Any) -> FastAPI:
|
||
"""给外部（测试、嵌入式调用）用的便捷工厂。"""
|
||
return create_app(ServerConfig(host=host, port=port, **kwargs))
|