snowy-yang
c414db4217
项目由 uv 管理，依赖装在 .venv。此前各处的提示语仍写作 `python main.py ...`，用户若在未激活虚拟环境时照抄，会因缺少 fastapi 而失败——这是正确性问题，不只是措辞。 涉及：main.py（12 行，含 prog 与各条 hint）、ai_player.py（5 行）、farm/client.py、farm/server/__main__.py（3 行，含 prog）、farm/server/__init__.py、farm/server/daemon.py、README（4 处示例与目录树注释）。 刻意保留 main.py:26 与 README:49 的「没装 uv？」兜底说明，那里 `python main.py` 是正确写法。 验收：ruff 全通过 / format 26 文件无改动 / pyright 0 错误 / 97 项测试通过。
685 lines
25 KiB
Python
685 lines
25 KiB
Python
#!/usr/bin/env python3
|
||
"""SimpleFarm 统一入口。
|
||
|
||
一个进程管两件事：**服务器**（后台常驻，人多人在线）和**客户端**（你自己玩、或者脚本/AI 调接口）。
|
||
|
||
快速开始
|
||
--------
|
||
uv sync # 建好环境（只有服务端需要第三方依赖）
|
||
uv run main.py serve --daemon # 后台启动小型游戏服务器
|
||
uv run main.py play # 打开终端界面玩自己的牧场
|
||
uv run main.py play --farm alice # 换个牧场名就是另一个玩家
|
||
uv run main.py status # 看服务器状态
|
||
uv run main.py stop # 优雅停机（会把所有牧场存档）
|
||
|
||
给脚本 / AI 用的
|
||
----------------
|
||
uv run main.py actions # 我当前能做哪些事
|
||
uv run main.py do plant '{"plot":1,"crop":"wheat"}'
|
||
uv run main.py state --json # 完整状态（机器可读）
|
||
uv run main.py watch # 实时滚动状态与日志
|
||
uv run main.py bot --farm ai-1 # 让规则机器人替你经营一个牧场
|
||
curl http://127.0.0.1:8765/api/actions # 直接打接口也行
|
||
|
||
浏览器打开 http://127.0.0.1:8765/ 可以看到网页版客户端。
|
||
|
||
没装 uv？用 `python main.py ...` 效果完全一样，命令里的 "uv run " 去掉即可。
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import argparse
|
||
import json
|
||
import os
|
||
import sys
|
||
import time
|
||
|
||
from farm import config
|
||
|
||
DEFAULT_URL = "http://127.0.0.1:8765"
|
||
|
||
BANNER = r"""
|
||
____ _ _ _____
|
||
/ ___|(_)_ __ ___ _ __ | | ___ | ___|_ _ _ __ _ __ ___
|
||
\___ \| | '_ ` _ \| '_ \| |/ _ \| |_ / _` | '__| '_ ` _ \
|
||
___) | | | | | | | |_) | | __/| _| (_| | | | | | | | |
|
||
|____/|_|_| |_| |_| .__/|_|\___||_| \__,_|_| |_| |_| |_|
|
||
|_|
|
||
一个小型农场游戏服务器：人在终端玩，AI 走接口玩
|
||
"""
|
||
|
||
|
||
# ---------------------------------------------------------------------- 输出
|
||
def _supports_color() -> bool:
|
||
if os.environ.get("NO_COLOR"):
|
||
return False
|
||
return sys.stdout.isatty()
|
||
|
||
|
||
def paint(text: str, code: str) -> str:
|
||
if not _supports_color():
|
||
return text
|
||
return f"\x1b[{code}m{text}\x1b[0m"
|
||
|
||
|
||
def ok(text: str) -> str:
|
||
return paint(text, "32")
|
||
|
||
|
||
def warn(text: str) -> str:
|
||
return paint(text, "33")
|
||
|
||
|
||
def bad(text: str) -> str:
|
||
return paint(text, "31")
|
||
|
||
|
||
def dim(text: str) -> str:
|
||
return paint(text, "90")
|
||
|
||
|
||
def die(message: str, hint: str = "", code: int = 1) -> int:
|
||
print(bad("✗ ") + message, file=sys.stderr)
|
||
if hint:
|
||
print(dim(" " + hint.replace("\n", "\n ")), file=sys.stderr)
|
||
return code
|
||
|
||
|
||
# ---------------------------------------------------------------------- 参数
|
||
def build_parser() -> argparse.ArgumentParser:
|
||
parser = argparse.ArgumentParser(
|
||
prog="uv run main.py",
|
||
description="SimpleFarm：终端农场游戏 + 小型多人游戏服务器（HTTP/WebSocket 供 AI 调用）",
|
||
formatter_class=argparse.RawDescriptionHelpFormatter,
|
||
epilog=BANNER,
|
||
)
|
||
sub = parser.add_subparsers(dest="command", metavar="<命令>")
|
||
|
||
# ---- 服务器
|
||
p_serve = sub.add_parser("serve", help="启动游戏服务器（可后台运行）")
|
||
p_serve.add_argument("--daemon", action="store_true", help="转到后台运行（守护进程）")
|
||
p_serve.add_argument(
|
||
"--host", default="127.0.0.1", help="监听地址，默认 127.0.0.1；局域网联机用 0.0.0.0"
|
||
)
|
||
p_serve.add_argument("--port", type=int, default=8765, help="监听端口，默认 8765")
|
||
p_serve.add_argument("--token", default=None, help="访问令牌，设置后接口需要携带")
|
||
p_serve.add_argument(
|
||
"--tick",
|
||
type=float,
|
||
default=config.DEFAULT_TICK_SECONDS,
|
||
help=f"每多少秒推进一天，默认 {config.DEFAULT_TICK_SECONDS:.0f} 秒",
|
||
)
|
||
p_serve.add_argument("--save-dir", default=None, help="存档目录，默认 <项目>/saves")
|
||
p_serve.add_argument("--world", default="SimpleFarm 公共农场", help="世界名称")
|
||
p_serve.add_argument("--max-sessions", type=int, default=64, help="同时在线的牧场上限")
|
||
p_serve.add_argument(
|
||
"--idle-timeout",
|
||
type=float,
|
||
default=1800.0,
|
||
help="无人访问多少秒后把牧场移出内存（存档保留）",
|
||
)
|
||
p_serve.add_argument("--no-remote-control", action="store_true", help="禁止接口停机")
|
||
p_serve.add_argument("--access-log", action="store_true", help="打印每个 HTTP 请求")
|
||
p_serve.add_argument(
|
||
"--log-level", default="info", choices=["critical", "error", "warning", "info", "debug"]
|
||
)
|
||
p_serve.add_argument("--log-file", default=None, help="守护进程日志路径")
|
||
p_serve.add_argument("--pid-file", default=None, help="守护进程 pid 文件路径")
|
||
|
||
p_stop = sub.add_parser("stop", help="停止后台服务器（会先落盘存档）")
|
||
p_stop.add_argument("--port", type=int, default=8765)
|
||
p_stop.add_argument("--host", default="127.0.0.1")
|
||
p_stop.add_argument("--token", default=None)
|
||
p_stop.add_argument("--force", action="store_true", help="超时后强制结束进程")
|
||
|
||
p_status = sub.add_parser("status", help="查看服务器状态")
|
||
p_status.add_argument("--port", type=int, default=8765)
|
||
p_status.add_argument("--host", default="127.0.0.1")
|
||
p_status.add_argument("--token", default=None)
|
||
p_status.add_argument("--logs", action="store_true", help="同时打印日志尾部")
|
||
p_status.add_argument("--json", action="store_true", help="输出机器可读的 JSON")
|
||
|
||
p_logs = sub.add_parser("logs", help="查看服务器日志尾部")
|
||
p_logs.add_argument("-n", "--lines", type=int, default=40)
|
||
|
||
# ---- 客户端
|
||
p_play = sub.add_parser("play", help="打开终端界面（TUI）连接服务器玩")
|
||
p_play.add_argument("--url", default=DEFAULT_URL, help=f"服务器地址，默认 {DEFAULT_URL}")
|
||
p_play.add_argument("--farm", default="default", help="牧场名，默认 default")
|
||
p_play.add_argument("--token", default=None)
|
||
p_play.add_argument("--name", default="", help="在服务器上显示的昵称")
|
||
p_play.add_argument("--no-color", action="store_true")
|
||
p_play.add_argument("--no-realtime", action="store_true", help="不用 WebSocket，改走 HTTP 轮询")
|
||
|
||
p_solo = sub.add_parser("solo", help="离线单机模式（本地引擎 + 终端界面，不开服务器）")
|
||
p_solo.add_argument("--tick", type=float, default=config.DEFAULT_TICK_SECONDS)
|
||
p_solo.add_argument("--save", default=None)
|
||
p_solo.add_argument("--new", action="store_true", help="忽略旧存档")
|
||
p_solo.add_argument("--no-color", action="store_true")
|
||
|
||
p_do = sub.add_parser("do", help="执行一个动作（脚本/AI 用途）")
|
||
p_do.add_argument("action", help="动作名：plant / water / harvest / ...")
|
||
p_do.add_argument(
|
||
"params", nargs="?", default="{}", help='JSON 参数，例如 \'{"plot":1,"crop":"wheat"}\''
|
||
)
|
||
_add_client_args(p_do)
|
||
|
||
p_batch = sub.add_parser("batch", help="批量执行动作（一个请求干多件事）")
|
||
p_batch.add_argument("actions", help='JSON 数组，例如 \'[{"action":"plant","params":{...}}]\'')
|
||
_add_client_args(p_batch)
|
||
|
||
p_actions = sub.add_parser("actions", help="列出当前所有合法动作")
|
||
_add_client_args(p_actions)
|
||
|
||
p_state = sub.add_parser("state", help="查看当前状态")
|
||
_add_client_args(p_state)
|
||
p_state.add_argument("--json", action="store_true", help="输出完整 JSON")
|
||
|
||
p_log = sub.add_parser("log", help="查看最近事件")
|
||
_add_client_args(p_log)
|
||
p_log.add_argument("-n", "--lines", type=int, default=20)
|
||
|
||
p_watch = sub.add_parser("watch", help="实时滚动状态与日志（适合观察 AI 玩）")
|
||
_add_client_args(p_watch)
|
||
p_watch.add_argument("--interval", type=float, default=1.0)
|
||
|
||
p_bot = sub.add_parser("bot", help="启动规则机器人替你经营一个牧场")
|
||
_add_client_args(p_bot)
|
||
p_bot.add_argument("--rounds", type=int, default=0, help="最多执行多少轮，0 表示不限")
|
||
p_bot.add_argument("--minutes", type=float, default=0.0, help="最多跑多少分钟，0 表示不限")
|
||
p_bot.add_argument("--pause", type=float, default=0.25, help="每个动作之间的间隔秒数")
|
||
|
||
p_farms = sub.add_parser("farms", help="查看服务器上的牧场排行榜")
|
||
_add_client_args(p_farms)
|
||
|
||
p_help = sub.add_parser("help", help="玩法与接口说明")
|
||
_add_client_args(p_help)
|
||
|
||
return parser
|
||
|
||
|
||
def _add_client_args(parser: argparse.ArgumentParser) -> None:
|
||
parser.add_argument("--url", default=DEFAULT_URL, help=f"服务器地址，默认 {DEFAULT_URL}")
|
||
parser.add_argument("--farm", default="default", help="牧场名，默认 default")
|
||
parser.add_argument("--token", default=None, help="访问令牌")
|
||
|
||
|
||
# ---------------------------------------------------------------------- 客户端
|
||
def make_client(args, realtime: bool = False):
|
||
"""构造远程客户端；连不上时给出可执行的下一步建议。"""
|
||
from farm.client import RemoteFarm
|
||
from farm.engine import GameError
|
||
|
||
try:
|
||
return RemoteFarm(
|
||
endpoint=args.url,
|
||
farm=getattr(args, "farm", "default"),
|
||
token=getattr(args, "token", None),
|
||
name=getattr(args, "name", "") or "cli",
|
||
realtime=realtime,
|
||
autostart=True,
|
||
)
|
||
except GameError as exc:
|
||
hint = (
|
||
f"服务器地址：{args.url}\n"
|
||
"先启动服务器：uv run main.py serve --daemon\n"
|
||
"再确认地址端口：uv run main.py status"
|
||
)
|
||
raise SystemExit(die(f"无法连接服务器：{exc.message}", hint)) from None
|
||
|
||
|
||
# ---------------------------------------------------------------------- 子命令
|
||
def cmd_serve(args) -> int:
|
||
from farm.server.__main__ import main as server_main
|
||
|
||
argv = [
|
||
"--host",
|
||
args.host,
|
||
"--port",
|
||
str(args.port),
|
||
"--tick",
|
||
str(args.tick),
|
||
"--world",
|
||
args.world,
|
||
"--max-sessions",
|
||
str(args.max_sessions),
|
||
"--idle-timeout",
|
||
str(args.idle_timeout),
|
||
"--log-level",
|
||
args.log_level,
|
||
]
|
||
if args.save_dir:
|
||
argv += ["--save-dir", args.save_dir]
|
||
if args.token:
|
||
argv += ["--token", args.token]
|
||
if args.daemon:
|
||
argv.append("--daemon")
|
||
if args.no_remote_control:
|
||
argv.append("--no-remote-control")
|
||
if args.access_log:
|
||
argv.append("--access-log")
|
||
if args.log_file:
|
||
argv += ["--log-file", args.log_file]
|
||
if args.pid_file:
|
||
argv += ["--pid-file", args.pid_file]
|
||
return server_main(argv)
|
||
|
||
|
||
def cmd_stop(args) -> int:
|
||
from farm.server.daemon import DaemonManager
|
||
|
||
manager = DaemonManager(host=args.host, port=args.port, token=args.token)
|
||
success, message = manager.stop(force=args.force)
|
||
print((ok("✓ ") if success else warn("• ")) + message)
|
||
return 0 if success else 1
|
||
|
||
|
||
def cmd_status(args) -> int:
|
||
from farm.server.daemon import DaemonManager
|
||
|
||
manager = DaemonManager(host=args.host, port=args.port, token=args.token)
|
||
info = manager.status()
|
||
|
||
if args.json:
|
||
print(json.dumps(info, ensure_ascii=False, indent=2))
|
||
return 0
|
||
|
||
if not info["running"]:
|
||
print(warn("• 服务器未运行"))
|
||
print(dim(" 启动：uv run main.py serve --daemon"))
|
||
print(dim(f" 地址：http://{args.host}:{args.port}/"))
|
||
return 1
|
||
|
||
health = info.get("health") or {}
|
||
print(ok("✓ 服务器运行中"))
|
||
print(f" 进程 pid {info['pid'] or '未知（不是本脚本启动的）'}")
|
||
print(f" 地址 http://{info['host']}:{info['port']}/")
|
||
if health:
|
||
print(f" 运行时长 {health.get('uptime', 0):.0f} 秒")
|
||
print(f" 牧场 {health.get('farms', 0)} 个 · 在线连接 {health.get('clients', 0)} 个")
|
||
print(f" 时钟 已推进 {health.get('clock_ticks', 0)} 天")
|
||
print(f" 存档目录 {manager.save_dir}")
|
||
print(f" 日志 {info['log_file']}")
|
||
|
||
world = manager.world()
|
||
if world and world.get("leaderboard"):
|
||
print(" 排行榜")
|
||
for row in world["leaderboard"][:5]:
|
||
print(
|
||
f" #{row['rank']} {row['farm']:<14} 净资产 {row['net_worth']:>8.0f}"
|
||
f" · 第 {row['day']} 天 · Lv.{row['level']} · 在线 {row['online']}"
|
||
)
|
||
|
||
if args.logs:
|
||
print(dim(" ---- 日志尾部 ----"))
|
||
for line in manager.tail(15):
|
||
print(" " + line)
|
||
return 0
|
||
|
||
|
||
def cmd_logs(args) -> int:
|
||
from farm.server.daemon import DaemonManager
|
||
|
||
manager = DaemonManager()
|
||
lines = manager.tail(args.lines)
|
||
if not lines:
|
||
print(warn("• 还没有日志（服务器从未在后台启动过）"))
|
||
return 1
|
||
for line in lines:
|
||
print(line)
|
||
return 0
|
||
|
||
|
||
def cmd_play(args) -> int:
|
||
from farm.tui import TuiApp
|
||
|
||
if not (sys.stdin is not None and sys.stdin.isatty()):
|
||
return die(
|
||
"当前终端不是交互式终端，无法启动界面。",
|
||
"在真实终端里运行，或用命令行模式：uv run main.py actions / do / watch",
|
||
)
|
||
client = make_client(args, realtime=not args.no_realtime)
|
||
label = f"{client.farm} @ {client.host}"
|
||
app = TuiApp(
|
||
client, save_path=None, use_color=not args.no_color, autosave=True, server_label=label
|
||
)
|
||
try:
|
||
app.run()
|
||
finally:
|
||
client.stop()
|
||
return 0
|
||
|
||
|
||
def cmd_solo(args) -> int:
|
||
from farm.engine import FarmEngine, Scheduler
|
||
from farm.storage import default_save_path, load_state
|
||
from farm.tui import TuiApp
|
||
|
||
if not (sys.stdin is not None and sys.stdin.isatty()):
|
||
return die("当前终端不是交互式终端，无法启动界面。")
|
||
|
||
tick = max(config.MIN_TICK_SECONDS, min(config.MAX_TICK_SECONDS, args.tick))
|
||
engine = FarmEngine(tick_seconds=tick)
|
||
save_path = args.save or default_save_path()
|
||
if args.new:
|
||
engine.add_log("新的一天，新的开始。欢迎来到 SimpleFarm！", "info")
|
||
else:
|
||
state, message = load_state(save_path)
|
||
if state is not None:
|
||
engine.state = state
|
||
engine.add_log(f"欢迎回来，第 {state.day} 天继续经营", "info")
|
||
print(message)
|
||
else:
|
||
engine.add_log("新的一天，新的开始。欢迎来到 SimpleFarm！", "info")
|
||
print(f"{message}，已开新档。")
|
||
|
||
scheduler = Scheduler(engine, enabled=True)
|
||
scheduler.start()
|
||
app = TuiApp(engine, save_path=save_path, use_color=not args.no_color)
|
||
try:
|
||
app.run()
|
||
finally:
|
||
scheduler.stop()
|
||
return 0
|
||
|
||
|
||
def cmd_do(args) -> int:
|
||
from farm.engine import GameError
|
||
|
||
client = make_client(args)
|
||
try:
|
||
params = json.loads(args.params) if args.params else {}
|
||
except json.JSONDecodeError as exc:
|
||
return die(
|
||
f"参数不是合法 JSON：{exc}",
|
||
'例子：uv run main.py do plant \'{"plot":1,"crop":"wheat"}\'',
|
||
)
|
||
if not isinstance(params, dict):
|
||
return die("参数必须是 JSON 对象")
|
||
|
||
started = time.time()
|
||
try:
|
||
result = client._action(args.action, params)
|
||
except GameError as exc:
|
||
return die(
|
||
f"{args.action} 失败（{exc.code}）：{exc.message}",
|
||
"看看现在能做什么：uv run main.py actions",
|
||
)
|
||
|
||
print(ok("✓ ") + str(result.get("message", "完成")) + dim(f" ({time.time() - started:.2f}s)"))
|
||
extra = {
|
||
k: v
|
||
for k, v in result.items()
|
||
if k not in ("ok", "message", "state", "log", "action") and not isinstance(v, (dict, list))
|
||
}
|
||
if extra:
|
||
print(dim(" " + " ".join(f"{k}={v}" for k, v in extra.items())))
|
||
state = result.get("state") or {}
|
||
if state:
|
||
print(
|
||
f" 第 {state.get('day')} 天 · 金币 {state.get('gold')} · "
|
||
f"行动点 {state.get('ap')}/{state.get('max_ap')} · 净资产 {state.get('net_worth')}"
|
||
)
|
||
return 0
|
||
|
||
|
||
def cmd_batch(args) -> int:
|
||
from farm.engine import GameError
|
||
|
||
client = make_client(args)
|
||
try:
|
||
items = json.loads(args.actions)
|
||
except json.JSONDecodeError as exc:
|
||
return die(
|
||
f"actions 不是合法 JSON：{exc}",
|
||
"例子：uv run main.py batch "
|
||
'\'[{"action":"harvest_all"},{"action":"buy_feed","params":{"quantity":5}}]\'',
|
||
)
|
||
if not isinstance(items, list):
|
||
return die("actions 必须是 JSON 数组")
|
||
try:
|
||
data = client.batch(items)
|
||
except GameError as exc:
|
||
return die(f"批量执行失败：{exc.message}")
|
||
for item in data.get("results", []):
|
||
mark = ok("✓") if item.get("ok") else bad("✗")
|
||
print(f" {mark} {item.get('action')}: {item.get('message', item.get('error'))}")
|
||
print(f" 共 {data.get('done')} 个成功 / {data.get('failed')} 个失败")
|
||
return 0
|
||
|
||
|
||
def cmd_actions(args) -> int:
|
||
client = make_client(args)
|
||
state = client.snapshot()
|
||
items = client.available_actions()
|
||
print(
|
||
f"牧场 {client.farm} · 第 {state.get('day')} 天 · 金币 {state.get('gold')} · "
|
||
f"行动点 {state.get('ap')}/{state.get('max_ap')}"
|
||
)
|
||
if not items:
|
||
print(warn("• 当前没有可执行的动作"))
|
||
print(dim(" 试试：uv run main.py do sleep （推进一天）"))
|
||
return 0
|
||
print(f"共 {len(items)} 个候选动作：\n")
|
||
for index, item in enumerate(items, start=1):
|
||
cost = item.get("cost") or {}
|
||
cost_text = " ".join(f"{k}:{v}" for k, v in cost.items()) or "免费"
|
||
params_text = json.dumps(item["params"], ensure_ascii=False)
|
||
print(f" [{index:>2}] {item['action']:<12} {params_text:<42} ({cost_text})")
|
||
print(dim(f" {item.get('reason', '')}"))
|
||
print(dim("\n执行方式：uv run main.py do <动作名> '<参数JSON>'"))
|
||
return 0
|
||
|
||
|
||
def cmd_state(args) -> int:
|
||
client = make_client(args)
|
||
if args.json:
|
||
print(json.dumps(client.snapshot(), ensure_ascii=False, indent=2))
|
||
return 0
|
||
snap = client.snapshot()
|
||
print(f"牧场 {snap.get('farm')}")
|
||
print(
|
||
f"第 {snap.get('day')} 天"
|
||
+ (
|
||
f" · {snap.get('seconds_to_tick')}s 后进入下一天"
|
||
if snap.get("seconds_to_tick") is not None
|
||
else " · 已暂停"
|
||
)
|
||
)
|
||
print(
|
||
f"金币 {snap.get('gold')} 等级 Lv.{snap.get('level')} "
|
||
f"({snap.get('exp')}/{snap.get('exp_needed')})"
|
||
)
|
||
print(f"行动点 {snap.get('ap')}/{snap.get('max_ap')} 饲料 {snap.get('feed')}")
|
||
print(f"净资产 {snap.get('net_worth')}")
|
||
print(f"土地 {snap.get('plot_count')}/{snap.get('max_plots')} 块")
|
||
for plot in snap.get("plots", []):
|
||
if plot["is_empty"]:
|
||
continue
|
||
status = "可收获" if plot["is_ripe"] else f"还需{plot['days_left']}天"
|
||
print(
|
||
f" {plot['position']:>2} 号地 {plot['crop_name']:<6} 生长 {plot['growth']}"
|
||
f"/{plot['grow_days']} 水分 {plot['moisture']} · {status}"
|
||
)
|
||
for animal in snap.get("animals", []):
|
||
note = (
|
||
f"{animal['product']} x{animal['pending']} 待收"
|
||
if animal["pending"]
|
||
else ("已喂食" if animal["fed"] else "未喂食")
|
||
)
|
||
print(f" #{animal['id']} {animal['name']:<4} {note}")
|
||
return 0
|
||
|
||
|
||
def cmd_log(args) -> int:
|
||
client = make_client(args)
|
||
entries = client.recent_log(limit=args.lines)
|
||
for entry in entries:
|
||
print(f"[D{entry['day']:>3}] {entry['text']}")
|
||
return 0
|
||
|
||
|
||
def cmd_watch(args) -> int:
|
||
"""实时滚动：状态每轮刷新，日志增量打印。适合盯着 AI 玩。"""
|
||
|
||
client = make_client(args, realtime=True)
|
||
print(BANNER)
|
||
print(f"正在观察牧场「{client.farm}」（{client.url('')}），Ctrl+C 退出\n")
|
||
try:
|
||
last_seq = 0
|
||
while True:
|
||
snap = client.snapshot()
|
||
for entry in client.recent_log(limit=0, since=last_seq):
|
||
last_seq = max(last_seq, int(entry.get("seq", 0)))
|
||
print(f"[D{entry['day']:>3}] {entry['text']}")
|
||
online = snap.get("online")
|
||
head = (
|
||
f"第 {snap.get('day')} 天 | 金币 {snap.get('gold')} | "
|
||
f"Lv.{snap.get('level')} | 行动点 {snap.get('ap')}/{snap.get('max_ap')} | "
|
||
f"净资产 {snap.get('net_worth')}"
|
||
)
|
||
if online is not None:
|
||
head += f" | 在线 {online}"
|
||
print("\r" + head + " " * 8, end="", flush=True)
|
||
time.sleep(max(0.2, args.interval))
|
||
except KeyboardInterrupt:
|
||
print("\n\n已退出观察。")
|
||
finally:
|
||
client.stop()
|
||
return 0
|
||
|
||
|
||
def cmd_bot(args) -> int:
|
||
from farm.bot import RuleBot
|
||
|
||
client = make_client(args, realtime=False)
|
||
bot = RuleBot(client, pause=args.pause, name=args.farm)
|
||
try:
|
||
summary = bot.run(rounds=args.rounds, max_minutes=args.minutes)
|
||
except KeyboardInterrupt:
|
||
bot.stop()
|
||
print("\n已中断。")
|
||
summary = {"actions": bot.actions_executed, "failures": bot.failures}
|
||
print(json.dumps(summary, ensure_ascii=False))
|
||
print(dim("提示：机器人和你调的是同一套接口，随时可以用 uv run main.py actions 接手。"))
|
||
return 0
|
||
|
||
|
||
def cmd_farms(args) -> int:
|
||
client = make_client(args)
|
||
world = client.world()
|
||
print(
|
||
f"世界：{world.get('world')} · 已运行 {world.get('uptime', 0):.0f} 秒 · "
|
||
f"牧场 {world.get('farms')} 个 · 连接 {world.get('clients')} 个"
|
||
)
|
||
print(f"时钟：每 {world.get('tick_seconds')} 秒推进一天\n")
|
||
print(f" {'#':>2} {'牧场':<14} {'净资产':>9} {'金币':>7} {'天':>4} {'等级':>4} {'在线':>4}")
|
||
for row in world.get("leaderboard", []):
|
||
print(
|
||
f" {row['rank']:>2} {row['farm']:<14} {row['net_worth']:>9.0f} "
|
||
f"{row['gold']:>7} {row['day']:>4} {row['level']:>4} {row['online']:>4}"
|
||
)
|
||
return 0
|
||
|
||
|
||
def cmd_help(args) -> int:
|
||
client = make_client(args)
|
||
print(client.help_text())
|
||
return 0
|
||
|
||
|
||
# ---------------------------------------------------------------------- 主入口
|
||
QUICKSTART = """
|
||
快速开始
|
||
--------
|
||
0. 建好环境（只有服务端需要依赖） uv sync
|
||
1. 启动服务器（后台常驻） uv run main.py serve --daemon
|
||
2. 打开网页客户端 http://127.0.0.1:8765/
|
||
3. 终端里玩自己的牧场 uv run main.py play --farm 你的名字
|
||
4. 让 AI / 机器人一起玩 uv run main.py bot --farm ai-1
|
||
5. 收工 uv run main.py stop
|
||
|
||
改代码之后跑这些
|
||
----------------
|
||
uv run ruff check . # 静态检查（未用导入、可疑写法……）
|
||
uv run ruff format . # 统一格式
|
||
uv run pyright # 类型检查
|
||
uv run python -m unittest discover -s tests # 全部测试
|
||
|
||
直接调接口（AI / 脚本）
|
||
-----------------------
|
||
curl 'http://127.0.0.1:8765/api/actions?farm=default'
|
||
curl -X POST 'http://127.0.0.1:8765/api/actions?farm=default' \\
|
||
-H 'Content-Type: application/json' \\
|
||
-d '{"action":"plant","params":{"plot":1,"crop":"wheat"}}'
|
||
|
||
uv run main.py actions # 我当前能做什么
|
||
uv run main.py do harvest '{"plot":1}'
|
||
uv run main.py state --json # 机器可读的完整状态
|
||
uv run main.py watch # 实时看别人/机器人玩
|
||
|
||
WebSocket（实时推送）
|
||
---------------------
|
||
ws://127.0.0.1:8765/ws?farm=default
|
||
服务器在状态变化时推送 {"type":"state", ...}；
|
||
也可以在 WS 上直接发动作：{"type":"action","action":"plant",
|
||
"params":{"plot":1,"crop":"wheat"},"id":1}，会收到 {"type":"result",...} 回执。
|
||
"""
|
||
|
||
|
||
def main(argv=None) -> int:
|
||
parser = build_parser()
|
||
args = parser.parse_args(argv)
|
||
|
||
handlers = {
|
||
"serve": cmd_serve,
|
||
"stop": cmd_stop,
|
||
"status": cmd_status,
|
||
"logs": cmd_logs,
|
||
"play": cmd_play,
|
||
"solo": cmd_solo,
|
||
"do": cmd_do,
|
||
"batch": cmd_batch,
|
||
"actions": cmd_actions,
|
||
"state": cmd_state,
|
||
"log": cmd_log,
|
||
"watch": cmd_watch,
|
||
"bot": cmd_bot,
|
||
"farms": cmd_farms,
|
||
"help": cmd_help,
|
||
}
|
||
if not args.command:
|
||
print(BANNER)
|
||
try:
|
||
from farm.server.daemon import DaemonManager
|
||
|
||
info = DaemonManager().status()
|
||
if info["running"]:
|
||
health = info.get("health") or {}
|
||
print(
|
||
ok("✓ 服务器正在运行")
|
||
+ f" http://{info['host']}:{info['port']}/ "
|
||
+ dim(f"牧场 {health.get('farms', 0)} 个 · 在线 {health.get('clients', 0)}")
|
||
)
|
||
else:
|
||
print(warn("• 服务器未运行"))
|
||
except Exception:
|
||
pass
|
||
print(QUICKSTART)
|
||
return 0
|
||
|
||
handler = handlers.get(args.command)
|
||
if handler is None:
|
||
return die(f"未知命令 {args.command}", "uv run main.py --help 查看全部命令")
|
||
try:
|
||
return handler(args)
|
||
except SystemExit as exc: # make_client 里的友好退出
|
||
return int(exc.code or 0)
|
||
except KeyboardInterrupt:
|
||
print("\n已中断。")
|
||
return 130
|
||
|
||
|
||
if __name__ == "__main__":
|
||
sys.exit(main())
|