From e713ff369a6f07939f7dbbe9292eaac029e09e2d Mon Sep 17 00:00:00 2001 From: JWeBot Contributors Date: Thu, 15 Oct 2026 15:17:32 +0800 Subject: [PATCH] feat: add CS Major autochess tournament --- CHANGELOG.md | 33 + README.md | 13 +- VERSION | 2 +- docs/cs-major-autochess-design.md | 658 +++++ .../jwebot/cs-major-autochess/README.md | 30 + .../jwebot/cs-major-autochess/plugin.toml | 14 + .../src/cs_major_autochess.py | 37 + .../cs-major-autochess/tests/test_plugin.py | 29 + packages/jwebot-config/pyproject.toml | 2 +- .../src/jwebot_config/__init__.py | 2 +- packages/jwebot-extension-host/pyproject.toml | 6 +- .../src/jwebot_extension_host/__init__.py | 2 +- packages/jwebot-kernel/pyproject.toml | 8 +- .../src/jwebot_kernel/__init__.py | 2 +- .../src/jwebot_kernel/edge_runtime.py | 309 +++ .../jwebot-official-extensions/pyproject.toml | 4 +- .../jwebot_official_extensions/__init__.py | 24 +- .../adapters/cs_major_autochess.py | 37 + .../cs_autochess.py | 2169 +++++++++++++++++ .../jwebot_official_extensions/descriptors.py | 21 + .../local_commands.py | 4 + packages/jwebot-sdk/pyproject.toml | 2 +- .../jwebot-sdk/src/jwebot_sdk/__init__.py | 2 +- pyproject.toml | 2 +- tests/test_official_extension_inventory.py | 1 + tests/test_release_064.py | 508 ++++ 26 files changed, 3901 insertions(+), 20 deletions(-) create mode 100644 docs/cs-major-autochess-design.md create mode 100644 extensions/jwebot/cs-major-autochess/README.md create mode 100644 extensions/jwebot/cs-major-autochess/plugin.toml create mode 100644 extensions/jwebot/cs-major-autochess/src/cs_major_autochess.py create mode 100644 extensions/jwebot/cs-major-autochess/tests/test_plugin.py create mode 100644 packages/jwebot-official-extensions/src/jwebot_official_extensions/adapters/cs_major_autochess.py create mode 100644 packages/jwebot-official-extensions/src/jwebot_official_extensions/cs_autochess.py create mode 100644 tests/test_release_064.py diff --git a/CHANGELOG.md b/CHANGELOG.md index 252d266..74f091e 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -1,5 +1,38 @@ # Changelog +## alpha 0.6.4.1015 + +- Add a reviewed 2-8 player CS Major autochess extension with a fixed + five-player starting lineup, one always-active coach, an eight-slot bench, + and a six-candidate shop that shows complete game data. +- Ship a fixed 60-player local pool with rifle, sniper, and in-game-leader + roles; team and nation synergies; shared copy counts; and deterministic + three-copy/two-star and nine-copy/three-star upgrades. +- Keep all five player slots occupied. Lineup changes are one-for-one swaps, + starters cannot be sold directly, and coach upgrades only add a 2%-10% + team bonus. +- Add player-selected firepower, stability, and clutch tactics. Resolve + stability over firepower, firepower over clutch, clutch over stability, + plus sniper/rifle/leader role counters before bounded +/-2% map variance. +- Generate the competition format once at tournament start: two players go + directly to the final, three play a round robin for two final places, four + use seeded 1v4 and 2v3 semifinals, and five to eight play a complete round + robin for four semifinal places. +- Add an explicit three-human-plus-one-AI test mode. The host can add one + deterministic local AI as seed four; it buys players, upgrades its coach, + swaps its strongest lineup, and selects a counter tactic without an online + model. +- Give Taffy a persistent host state machine: welcome and format briefing, + rotating human preparation order, turn enforcement, explicit handoffs, + round introductions, matchup calls, overtime narration, advancement, and + championship presentation. +- Run group matches as BO3 and knockout matches as BO5. Simulate every map + under MR15, including 15-15 overtime, 19-17/22-20 scores, and separately + archived overtime stories. +- Persist run state, idempotent action responses, structured battles, and + every generated story in serialized SQLite transactions. Invalid archives + roll back the full action. + ## alpha 0.6.3.1015 - Replace instant personal lottery tickets with one persistent multiplayer diff --git a/README.md b/README.md index 50328fb..a4a7081 100644 --- a/README.md +++ b/README.md @@ -8,7 +8,7 @@ JWeBot 是运行于 Linux 的单群微信自动回复系统。它通过本地 OC 观察选定的微信窗口，将消息写入持久化 Edge 账本，调用经过审核的命令插件， 并通过带视觉确认的 GUI 发送链路回复。 -当前版本：`alpha 0.6.3.1015`。 +当前版本：`alpha 0.6.4.1015`。 0.6.1 新增本地上下文总结、人物与别名索引、单人虚拟金币游戏，并增加 两秒画面复核以补抓遗漏的消息变化。命令和游戏管理规则见 @@ -20,6 +20,12 @@ JWeBot 是运行于 Linux 的单群微信自动回复系统。它通过本地 OC 0.6.3 将彩票改为群内多人轮次，加入报名、状态、主持人开奖和取消； 抽奖、彩票、21 点统一使用 `Jacky #N：...` 局号并支持 `/游戏记录`。 完整玩法见[多人彩票与游戏记录](docs/group-lottery-0.6.3.md)。 +0.6.4 新增 2～8 人 CS Major 自走棋：固定五人首发与教练、每轮 6 人 +公开商店、属性/定位/战术克制、小组单循环、1v4 与 2v3 淘汰赛，以及 +BO3/BO5 的 MR15 加时赛。支持三名真人加入一支确定性 AI 测试战队； +Taffy 负责赛制播报、顺序准备、交棒和全程剧情主持。所有选秀、战斗和 +加时剧情均在本地事务内落档。 +完整规则见[CS Major 自走棋设计稿](docs/cs-major-autochess-design.md)。 ### 运行流程 @@ -90,6 +96,9 @@ PATH="$PWD/.venv/bin:$PATH" ./tools/check-local.sh /21点 开始 [1～500]；开局后 /21点 要牌|停牌|状态|结束 /游戏记录 [页码]（/gamehistory） /游戏帮助（/games） +/自走棋开始|报名|开赛|状态|结算|档案 +/自走棋加AI战队（三名真人测试模式） +/选1～6；/战术 火力|稳定|残局；/阵容；/结束回合 /丢骰子 [NdM] /猜数字 开始 [上限] /石头剪刀布 [开始|加入|出|状态|结算|结束] @@ -258,7 +267,7 @@ visually confirmed GUI bridge. Current version: ```text -alpha 0.6.3.1015 +alpha 0.6.4.1015 ``` ## Runtime Flow diff --git a/VERSION b/VERSION index 3a6b344..2c012ef 100644 --- a/VERSION +++ b/VERSION @@ -1 +1 @@ -alpha 0.6.3.1015 +alpha 0.6.4.1015 diff --git a/docs/cs-major-autochess-design.md b/docs/cs-major-autochess-design.md new file mode 100644 index 0000000..122e498 --- /dev/null +++ b/docs/cs-major-autochess-design.md @@ -0,0 +1,658 @@ +# CS Major 自走棋设计稿 + +本文档定义一个完全本地、确定性结算、以 CS 职业选手为棋子的群聊 +小游戏方案。赛事包装统一为“第 n 届 IEM 盐田 Major”。AI 默认负责主持、 +裁判、播报和剧情生成；主持人也可显式加入一支本地确定性 AI 战队。 + +## 1. 设计目标 + +1. 做成一个多人报名、多人运营、多人对战的群聊自走棋。 +2. 玩法重心放在每轮选秀和阵容运营，而不是复杂微操。 +3. 用 Major 赛事包装 PvE、小组单循环、淘汰赛和终局剧情。 +4. 结算逻辑必须完全本地、可重放、可持久化、可回档验证。 +5. 每一段剧情都要落档，且剧情只能在数值结算之后生成。 + +## 2. 核心体验 + +一句话定义： + +`公开 6 人选秀 -> 凑羁绊与升星 -> 打 PvE 拿资源 -> 打 PvP 跑 Major 赛程 -> 保存每轮赛报与剧情` + +玩家感受到的乐趣应当主要来自： + +- 抢 6 人选秀里的关键卡； +- 决定要不要升教练等级、保经济还是强行 D 牌； +- 赌阵容成型时点； +- 看小组单循环和淘汰赛赛报； +- 看最终冠军剧情与本届 Major 档案。 + +## 3. 对局对象与规模 + +### 3.1 报名 + +- 开局命令由任意玩家发起。 +- 其他玩家在报名窗口内加入。 +- 建议一局支持 `2~8` 名真人玩家。 +- 主持人可在报名阶段发送 `/自走棋加AI战队`，最多加入一支 AI 战队。 +- 三名真人加一支 AI 时，按四队赛制直接生成 `1v4`、`2v3` 半决赛。 +- AI 占一个参赛席位，但不对应群成员，也不调用在线模型。 +- AI 每回合以确定性规则买牌、升级教练、调整首发和选择克制战术。 + +### 3.2 玩家初始状态 + +每位玩家开局拥有： + +- 生命值：`100` +- 金币：`10` +- 教练等级：`1` +- 完整首发：`5 名一星选手 + 1 名教练` +- 阵容位：固定 `5 个选手位 + 1 个教练位`，比赛时必须全部上场 +- 备战席：`8 格` +- 连胜/连败计数：`0` +- 本局独立事件档案编号 + +开赛时系统从低费池为每位玩家确定性分配 5 名首发选手，保证第一场 +比赛就是完整的五人阵容。后续选秀用于替换、强化和升星，而不是解锁 +上场人口。 + +## 4. 选手池设计 + +### 4.1 选手来源 + +只使用“近期非常出名”的职业选手。第一期池子优先覆盖以下俱乐部与明星： + +- Spirit +- Vitality +- Falcons +- G2 +- FaZe +- NAVI +- MOUZ +- The MongolZ +- Astralis +- Liquid +- FURIA +- Eternal Fire + +选手数据不是对真实比赛数据的一比一复刻，而是基于公众认知做游戏化映射。 +首发版本固定维护 `60` 名棋子，运行期不在线抓取或改写数值。 + +### 4.2 费用层级 + +选手分为 `1~5 费`，越高费越稀有，基础数值和成长更强。 + +示例分层： + +- `5 费`：donk、ZywOo +- `4 费`：m0NESY、NiKo、ropz、frozen +- `3 费`：apEX、b1t、xertioN、w0nderful、malbsMd +- `2 费`：辅助型或中坚选手 +- `1 费`：前期过渡与羁绊补齐牌 + +正式实现时应维护一份固定的选手静态表，不允许运行期随意漂移。 + +### 4.3 定位 + +只保留三个核心定位： + +- `步枪手` +- `狙击手` +- `指挥` + +定位不是羁绊，而是战斗中的属性倾向与技能触发来源。 + +### 4.4 羁绊 + +只保留两类羁绊： + +1. `战队羁绊` +2. `国家羁绊` + +战队羁绊可来自： + +- 当前主战队； +- 历史待过的关键战队； +- 每名选手最多挂 `3` 条战队标签，避免过度泛化。 + +国家羁绊只有 `1` 条。 + +## 5. 选手数值 + +### 5.1 基础数值轴 + +为避免面板过宽，第一期只展示三个主数值： + +- `火力` +- `稳定` +- `残局` + +其中指挥位额外拥有隐藏或显式的 `指挥` 修正值，用于影响全队。 + +### 5.2 数值范围 + +- 1 费：`52~68` +- 2 费：`60~76` +- 3 费：`68~84` +- 4 费：`78~92` +- 5 费：`88~98` + +要求同费位内部也要有明显差异，不能所有 4 费都只有一两点波动。 + +### 5.3 派生战斗权重 + +可先采用以下简单模型： + +`单卡战力 = 火力 * 0.45 + 稳定 * 0.30 + 残局 * 0.25 + 星级修正 + 羁绊修正 + 教练修正` + +若定位是狙击手，可给火力更高上限； +若定位是指挥，可给团队修正更高权重。 + +### 5.4 属性与定位克制 + +战斗不能只比较总战力后掷骰子。三项主属性形成闭环： + +- `稳定克火力`：稳定阵容能吸收强攻的第一波冲击； +- `火力克残局`：前期人数优势会压缩残局发挥空间； +- `残局克稳定`：高残局能力能打破按部就班的默认局面。 + +三个定位也形成轻量闭环： + +- `狙击手克步枪手` +- `步枪手克指挥` +- `指挥克狙击手` + +玩家每回合可选择 `/战术 火力|稳定|残局`。战术选择会强化对应属性， +并在克制对方战术时获得额外修正。属性、定位和战术的总克制修正上限 +应明显高于随机波动，让调阵和判断对手比运气更重要。 + +## 6. 升星规则 + +- `3 张同名 = 2 星` +- `9 张同名 = 3 星` + +升星收益建议： + +- 2 星：基础战力乘 `1.35` +- 3 星：基础战力乘 `1.80` + +升星后保留原有羁绊与定位，不改变标签。 + +## 7. 教练系统 + +教练位不是普通战斗单位，但始终与 5 名选手一起上场。 + +### 7.1 教练等级作用 + +教练只提供额外的全队战术加成，不控制上场人数，也不占用选手位。 + +建议加成： + +- 1 级：全队战力 `+2%` +- 2 级：全队战力 `+4%` +- 3 级：全队战力 `+6%` +- 4 级：全队战力 `+8%` +- 5 级：全队战力 `+10%` + +### 7.2 教练升级 + +玩家可花金币升级教练经验。 + +建议升级费用为 `6 / 10 / 16 / 24` 金币。升级只提高上述战术加成。 +高费选手的商店概率改为随赛事回合推进，不再绑定教练等级。 + +## 8. 每轮 6 人选秀 + +这是本模式的核心。 + +### 8.1 固定规则 + +- 每轮只展示 `6` 名候选选手。 +- 这 `6` 人对所有玩家公开可见。 +- 每位玩家始终保有 5 名首发；购买的新选手进入备战席或与同名选手合成。 +- 调整阵容时必须用备战席选手替换首发，不能让首发少于 5 人。 +- 第一版采用“公开同池、各自购买”的简化方案： + - 所有人看到同一组 6 人； + - 某个玩家选择后，不会把该候选从其他玩家界面里抢空； + - 但底层共享总卡池仍然会影响后续刷新概率。 +- 每名候选都必须展示完整数据： + - 编号 `1~6` + - 选手名 + - 费用 + - 定位 + - 国家 + - 主战队 + - 历史战队 + - 主数值 +- 玩家通过编号选择，不输入复杂自然语言。 + +### 8.2 选秀指令 + +建议指令面如下： + +```text +/自走棋开始 +/自走棋报名 +/自走棋加AI战队 +/自走棋状态 +/选1 +/选2 +/选3 +/选4 +/选5 +/选6 +/刷新 +/锁定 +/升级教练 +/战术 火力|稳定|残局 +/上阵 备战编号 [首发位] +/下阵 首发位 备战编号 +/出售 备战编号 +/阵容 +/赛程 +/结束回合 +``` + +中文是主命令，英文别名可以后补。 + +### 8.3 刷新与锁定 + +- 每轮默认展示一组 6 人。 +- 玩家可支付金币执行刷新。 +- 玩家可锁定本轮 6 人，使下一轮仍保留同一组候选。 + +### 8.4 顺位 + +- 开局首轮选秀顺位随机。 +- 后续普通回合不再做蛇形公共抢位，而是每位玩家各自面对同一组 6 人池。 +- 如果后续想加竞技性，可在第二阶段引入“共享卡池抢空”机制。 + +第一期建议先做“公开同池、各自结算”，避免群聊交互过重。 + +## 9. 回合流程 + +建议一整回合拆成五段： + +1. `回合开始播报` +2. `6 人选秀 / 刷新 / 升级 / 调阵` +3. `PvE 或 PvP 结算` +4. `回合结算与经济发放` +5. `剧情归档` + +## 10. PvE 设计 + +PvE 不叫“打野怪”，统一包装成赛事内活动。 + +### 10.1 PvE 类型 + +- 训练赛 +- 媒体日 +- 热身赛 +- 赞助挑战 +- 地图专项训练 + +### 10.2 PvE 作用 + +PvE 的目标不是淘汰人，而是给资源： + +- 金币 +- 装备 +- 战术卡 +- 教练经验 +- 临时 buff + +### 10.3 PvE 结算逻辑 + +PvE 对手是模板化敌方阵容或阈值型事件，不需要复杂 AI。 + +示例： + +`玩家总战力 >= 目标战力` 则获胜，否则失败。 + +这样实现简单，也利于稳定复盘。 + +## 11. PvP 赛制 + +### 11.1 赛事包装 + +整局赛事名称固定为： + +`第 n 届 IEM 盐田 Major` + +### 11.2 按参赛人数生成赛制 + +- `2 人`：直接进入决赛。 +- `3 支队伍`：进行完整单循环，前 2 名进入决赛。 +- `4 人`：不打小组赛，按报名种子直接进行 `1v4`、`2v3` 半决赛。 +- `5~8 人`：进行完整小组单循环，每两支队伍只交手一次，前 4 名晋级。 + +奇数队伍的小组比赛日会有一队轮空，但不会获得胜场。 +小组赛每场采用 `BO3`。 + +### 11.3 小组排名 + +小组赛首先按胜场排序；同胜场依次比较： + +1. 相互交手结果； +2. 当前完整五人阵容战力； +3. 剩余生命值； +4. 固定种子编号。 + +所有配对在开赛时一次性生成并落档，不允许运行中根据结果重排赛程。 + +### 11.4 淘汰赛 + +- 四强固定为 `小组第 1 vs 第 4`、`第 2 vs 第 3`。 +- 3 人赛制由小组第 1、2 名直接进入决赛。 +- 4 人赛制使用报名种子 `1v4`、`2v3`。 +- 2 人赛制直接进入决赛。 +- 半决赛胜者进入总决赛，决出本届冠军。 +- 半决赛和总决赛均采用 `BO5`。 + +## 12. 战斗结算 + +### 12.1 目标 + +战斗要做到： + +- 本地确定性； +- 单轮可解释； +- 支持战报； +- 不依赖在线模型。 + +### 12.2 简化模型 + +建议第一期采用“阵容总分 + 对位修正”的轻量模型。 + +#### 步骤 1：计算阵容基础战力 + +`阵容基础战力 = 5 名首发选手战力总和` + +#### 步骤 2：加上团队修正 + +团队修正来源： + +- 战队羁绊 +- 国家羁绊 +- 教练等级提供的全队百分比加成 +- 装备/战术卡 +- 连胜或特殊事件 buff + +#### 步骤 3：计算克制修正 + +依次计算： + +1. `稳定 > 火力 > 残局 > 稳定` 的属性对位； +2. `狙击手 > 步枪手 > 指挥 > 狙击手` 的定位对位； +3. 双方本回合主动选择的战术对位。 + +克制结果必须进入结构化战报，明确展示双方修正值。 + +#### 步骤 4：加入小幅波动 + +只允许 `-2% ~ +2%` 的确定性哈希波动。单次运气不能覆盖成型阵容的 +属性、定位和战术克制优势。 + +#### 步骤 5：计算比分 + +单图采用 `MR15`： + +- 常规时间先到 `16` 分获胜； +- 常规时间 `15-15` 进入加时； +- 加时可产生 `19-17`，多轮加时可产生 `22-20` 等比分； +- 每次加时除结构化单图结果外，必须额外生成一段加时剧情并落档。 + +小组赛按 BO3 先赢 2 图，淘汰赛按 BO5 先赢 3 图。系列赛战报必须展示 +大比分、每张图比分、加时次数和克制修正。 + +## 13. 剧情系统与落档 + +这是本模式的强制要求。 + +### 13.1 归档原则 + +必须先结算，后写剧情。 + +即： + +1. 固化数值结果； +2. 存档回合事实； +3. 基于事实生成文字赛报； +4. 再把文字赛报作为附加档案保存。 + +### 13.2 必须落档的节点 + +- 开局报名完成 +- 每轮 6 人选秀结果 +- 每次刷新和关键升星 +- 每次 PvE 事件结果 +- 每场 PvP 结果 +- 小组赛排名、晋级和淘汰节点 +- 淘汰赛对阵与结果 +- 冠军总结 + +### 13.3 档案结构 + +建议至少保存两类记录： + +1. `结构化事件档` +2. `文本剧情档` + +结构化事件档用于回放和调试； +文本剧情档用于玩家阅读。 + +示例： + +```json +{ + "run_id": "major_0001", + "round": 4, + "phase": "group_2", + "event_type": "pvp_result", + "player_a": "Jacky", + "player_b": "Alice", + "winner": "Jacky", + "score_band": "close_win", + "roster_a_power": 418, + "roster_b_power": 403 +} +``` + +## 14. 数据持久化设计 + +建议新增以下 SQLite 表。 + +### 14.1 对局主表 + +`autochess_runs` + +- run_id +- group_fingerprint +- season_number +- state +- created_at +- updated_at + +### 14.2 玩家状态表 + +`autochess_players` + +- run_id +- person_key +- display_name +- hp +- gold +- coach_level +- coach_xp +- wins +- losses +- streak +- board_json +- bench_json +- shop_json +- locked_shop + +### 14.3 选秀记录表 + +`autochess_shop_events` + +- run_id +- round_number +- person_key +- visible_pool_json +- action_type +- action_payload_json +- created_at + +### 14.4 战斗结果表 + +`autochess_battles` + +- run_id +- round_number +- phase +- battle_type +- player_a +- player_b +- result_json +- created_at + +### 14.5 剧情归档表 + +`autochess_story_logs` + +- run_id +- round_number +- phase +- story_type +- story_text +- facts_json +- source_event_id +- created_at + +### 14.6 幂等与重放表 + +`autochess_action_replay` + +- run_id +- person_key +- round_number +- event_id +- action_kind +- response_text +- created_at + +## 15. 回放与幂等 + +因为群聊环境会出现重复事件、OCR 重读和命令重放，必须保证： + +- 同一事件只执行一次； +- 同一回合不能重复发奖或重复扣款； +- 同一段剧情不能重复写多份版本； +- 回合失败时整段事务回滚。 + +建议沿用现有本地游戏的幂等思路，为自走棋单独增加 action replay 表。 + +## 16. 指令交互风格 + +建议所有回复都遵循现有小游戏风格： + +`用户名 #局号：内容` + +例如： + +```text +Jacky #3：第 4 回合开始，你当前 32 金币，教练 3 级。 +本轮选秀如下： +1. donk 5费 步枪手 Spirit 俄罗斯 火力96 稳定88 残局86 +2. apEX 3费 指挥 Vitality 法国 火力68 稳定81 指挥95 +... +发送“选1”到“选6”，或发送“刷新”“锁定”“升级教练”。 +``` + +### 16.1 Taffy 主持状态机 + +Taffy 必须承担完整赛事主持，而不是只在结算时出现。 + +开赛时必须包含： + +- `欢迎大家来到第 n 届 IEM 盐田 Major` +- 本届参赛者和种子 +- 当前人数对应的完整赛制 +- 小组赛 BO3、淘汰赛 BO5、单图 MR15 和加时规则 +- 当前回合内容 +- 第一位需要准备的玩家 + +每回合按真人玩家生成确定性的准备顺序。AI 自动准备，不占真人操作轮次。 +只有当前玩家可以购买、刷新、选择战术和调整阵容。提交后必须明确交棒： + +```text +Taffy：Jacky，请做好赛前准备；完成后发送 /结束回合。 +Taffy：Jacky 准备完成！到 Alice 做准备了。 +Taffy：Alice 准备完成！全员准备完毕，主持人可以结算。 +``` + +下一回合轮换首位准备者，避免固定由同一个人总是先操作。 + +以下节点都必须由 Taffy 主持并落档： + +- 报名开启和新选手加入 +- AI 测试战队加入 +- 赛制与种子公布 +- 每位玩家开始/完成准备 +- 每轮开始和结算 +- 每场 BO3/BO5 对阵与结果 +- 每张加时地图 +- 小组排名和前四晋级 +- 1v4、2v3 半决赛 +- 总决赛和冠军颁奖 + +## 17. 实现顺序建议 + +建议按三个阶段落地。 + +### 第一阶段：最小可玩版 + +- 建局 / 报名 / 状态 +- 每轮 6 人选秀 +- 买卡、卖卡、升星 +- 教练等级 +- 简化 PvE +- 简化 PvP +- 基础剧情归档 + +### 第二阶段：赛事化增强 + +- 小组单循环完整配对 +- 1v4、2v3 淘汰赛树 +- 羁绊与装备系统细化 +- 战报模板优化 + +### 第三阶段：内容扩展 + +- 在固定 60 人首发池基础上增加新赛季棋子 +- 更多俱乐部和国家 +- 特殊赛事事件 +- 更细的剧情风格与本届 Major 年鉴 + +## 18. 第一版明确不做的内容 + +为保证能上线，第一版先不做： + +- 实时地图微操 +- 复杂技能动画 +- 过多职业分类 +- 在线爬取实时战队数据 +- 大量开放文本输入解析 + +## 19. 最终结论 + +这套模式最适合按“强运营、轻操作、重赛报”路线实现。 + +最重要的三个锚点已经确定： + +1. `每轮固定 6 人公开选秀` +2. `PvE 负责资源，PvP 负责小组单循环与淘汰赛` +3. `每一段剧情都必须落档` + +如果按这个设计推进，实现出来的成品会更像一场可回放、可复盘、 +可连载的本地 Major 模拟器，而不是一个只有数值结算的普通抽卡游戏。 diff --git a/extensions/jwebot/cs-major-autochess/README.md b/extensions/jwebot/cs-major-autochess/README.md new file mode 100644 index 0000000..10c9384 --- /dev/null +++ b/extensions/jwebot/cs-major-autochess/README.md @@ -0,0 +1,30 @@ +# cs-major-autochess + +Author: `jwebot` + +CS自走棋 + +Help: + +- `/自走棋开始|报名` + +Registered commands: + +- `/自走棋` +- `/cs自走棋` +- `/autochess` +- `/选` +- `/刷新` +- `/锁定` +- `/战术` +- `/升级教练` +- `/上阵` +- `/下阵` +- `/出售` +- `/阵容` +- `/赛程` +- `/结束回合` + +The adapter has no direct Ledger, GUI, environment, shell, or network access. +It delegates to the reviewed `cs-major-autochess` Kernel service profile +and returns a structured `ReplyPlan`. diff --git a/extensions/jwebot/cs-major-autochess/plugin.toml b/extensions/jwebot/cs-major-autochess/plugin.toml new file mode 100644 index 0000000..7ff0e19 --- /dev/null +++ b/extensions/jwebot/cs-major-autochess/plugin.toml @@ -0,0 +1,14 @@ +id = "cs-major-autochess" +author = "jwebot" +description = "CS自走棋" +help = ["/自走棋开始|报名"] +version = "1.0.0" +api_version = "1" +entrypoint = "cs_major_autochess:plugin" +type = "command" + +[capabilities] +commands = ["/自走棋", "/cs自走棋", "/autochess", "/选", "/刷新", "/锁定", "/战术", "/升级教练", "/上阵", "/下阵", "/出售", "/阵容", "/赛程", "/结束回合"] +storage = false +network_profiles = [] +service_profiles = ["cs-major-autochess"] diff --git a/extensions/jwebot/cs-major-autochess/src/cs_major_autochess.py b/extensions/jwebot/cs-major-autochess/src/cs_major_autochess.py new file mode 100644 index 0000000..6e3371e --- /dev/null +++ b/extensions/jwebot/cs-major-autochess/src/cs_major_autochess.py @@ -0,0 +1,37 @@ +"""Generated reviewed adapter for cs-major-autochess.""" + +from jwebot_sdk import Event, ExtensionResult, ReplyPlan, SingleGroupContext + + +PROFILE = 'cs-major-autochess' +AUTHOR = 'jwebot' +DESCRIPTION = 'CS自走棋' +HELP_ENTRIES = ('/自走棋开始|报名',) +COMMANDS = ('/自走棋', '/cs自走棋', '/autochess', '/选', '/刷新', '/锁定', '/战术', '/升级教练', '/上阵', '/下阵', '/出售', '/阵容', '/赛程', '/结束回合') + + +def _handler(context: SingleGroupContext, command: str): + def handle(event: Event, arguments: str) -> ExtensionResult: + response = context.services.call( + PROFILE, + operation="execute", + payload={ + "command": command, + "arguments": arguments, + "event": event.to_mapping(), + }, + ) + text = response.get("text") + if not isinstance(text, str) or not text.strip(): + raise ValueError("local service returned invalid text") + return ExtensionResult( + reply=ReplyPlan.text(text), + metadata={"local_extension": PROFILE, "author": AUTHOR}, + ) + + return handle + + +def plugin(context: SingleGroupContext) -> None: + for command in COMMANDS: + context.commands.register(command, _handler(context, command)) diff --git a/extensions/jwebot/cs-major-autochess/tests/test_plugin.py b/extensions/jwebot/cs-major-autochess/tests/test_plugin.py new file mode 100644 index 0000000..aec2120 --- /dev/null +++ b/extensions/jwebot/cs-major-autochess/tests/test_plugin.py @@ -0,0 +1,29 @@ +import sys +import unittest +from pathlib import Path +from unittest import mock + + +SOURCE = Path(__file__).resolve().parents[1] / "src" +sys.path.insert(0, str(SOURCE)) + +from cs_major_autochess import AUTHOR, DESCRIPTION, HELP_ENTRIES, plugin + + +class PluginTests(unittest.TestCase): + def test_registers_declared_commands_and_author(self): + context = mock.Mock() + plugin(context) + registered = [ + call.args[0] + for call in context.commands.register.call_args_list + ] + self.assertEqual(AUTHOR, 'jwebot') + self.assertEqual(DESCRIPTION, 'CS自走棋') + self.assertEqual(HELP_ENTRIES, ('/自走棋开始|报名',)) + self.assertIn('/自走棋', registered) + self.assertEqual(len(registered), 14) + + +if __name__ == "__main__": + unittest.main() diff --git a/packages/jwebot-config/pyproject.toml b/packages/jwebot-config/pyproject.toml index c6dbe83..f78539a 100644 --- a/packages/jwebot-config/pyproject.toml +++ b/packages/jwebot-config/pyproject.toml @@ -4,7 +4,7 @@ build-backend = "setuptools.build_meta" [project] name = "jwebot-config" -version = "0.6.3a0+1015" +version = "0.6.4a0+1015" description = "Profile-based configuration compiler for JWeBot" requires-python = ">=3.9" dependencies = [ diff --git a/packages/jwebot-config/src/jwebot_config/__init__.py b/packages/jwebot-config/src/jwebot_config/__init__.py index 81f0acc..394bdbb 100644 --- a/packages/jwebot-config/src/jwebot_config/__init__.py +++ b/packages/jwebot-config/src/jwebot_config/__init__.py @@ -36,4 +36,4 @@ __all__ = [ "write_default_profile", ] -__version__ = "0.6.3a0+1015" +__version__ = "0.6.4a0+1015" diff --git a/packages/jwebot-extension-host/pyproject.toml b/packages/jwebot-extension-host/pyproject.toml index d3c2445..7c310b1 100644 --- a/packages/jwebot-extension-host/pyproject.toml +++ b/packages/jwebot-extension-host/pyproject.toml @@ -4,12 +4,12 @@ build-backend = "setuptools.build_meta" [project] name = "jwebot-extension-host" -version = "0.6.3a0+1015" +version = "0.6.4a0+1015" description = "Single-release, single-invocation process host for JWeBot extensions" requires-python = ">=3.9" dependencies = [ - "jwebot-sdk==0.6.3a0+1015", - "jwebot-kernel==0.6.3a0+1015", + "jwebot-sdk==0.6.4a0+1015", + "jwebot-kernel==0.6.4a0+1015", ] [project.scripts] diff --git a/packages/jwebot-extension-host/src/jwebot_extension_host/__init__.py b/packages/jwebot-extension-host/src/jwebot_extension_host/__init__.py index e441828..a8d0c52 100644 --- a/packages/jwebot-extension-host/src/jwebot_extension_host/__init__.py +++ b/packages/jwebot-extension-host/src/jwebot_extension_host/__init__.py @@ -16,4 +16,4 @@ __all__ = [ "process_request", ] -__version__ = "0.6.3a0+1015" +__version__ = "0.6.4a0+1015" diff --git a/packages/jwebot-kernel/pyproject.toml b/packages/jwebot-kernel/pyproject.toml index e6c23b6..eeb0187 100644 --- a/packages/jwebot-kernel/pyproject.toml +++ b/packages/jwebot-kernel/pyproject.toml @@ -4,13 +4,13 @@ build-backend = "setuptools.build_meta" [project] name = "jwebot-kernel" -version = "0.6.3a0+1015" +version = "0.6.4a0+1015" description = "Stable routing and extension authority for JWeBot" requires-python = ">=3.9" dependencies = [ - "jwebot-config==0.6.3a0+1015", - "jwebot-sdk==0.6.3a0+1015", - "jwebot-official-extensions==0.6.3a0+1015", + "jwebot-config==0.6.4a0+1015", + "jwebot-sdk==0.6.4a0+1015", + "jwebot-official-extensions==0.6.4a0+1015", "tomli>=2; python_version < '3.11'", ] diff --git a/packages/jwebot-kernel/src/jwebot_kernel/__init__.py b/packages/jwebot-kernel/src/jwebot_kernel/__init__.py index c60bf2f..4b91e1a 100644 --- a/packages/jwebot-kernel/src/jwebot_kernel/__init__.py +++ b/packages/jwebot-kernel/src/jwebot_kernel/__init__.py @@ -107,4 +107,4 @@ __all__ = [ "verify_installed_release_signature", ] -__version__ = "0.6.3a0+1015" +__version__ = "0.6.4a0+1015" diff --git a/packages/jwebot-kernel/src/jwebot_kernel/edge_runtime.py b/packages/jwebot-kernel/src/jwebot_kernel/edge_runtime.py index c219be3..fadf01d 100644 --- a/packages/jwebot-kernel/src/jwebot_kernel/edge_runtime.py +++ b/packages/jwebot-kernel/src/jwebot_kernel/edge_runtime.py @@ -1070,6 +1070,71 @@ class Ledger: ON group_lottery_entries( group_fingerprint, round_number, created_at, person_key ); + CREATE TABLE IF NOT EXISTS autochess_runs ( + group_fingerprint TEXT NOT NULL, + run_id TEXT NOT NULL, + season_number INTEGER NOT NULL, + state TEXT NOT NULL, + host_person_key TEXT NOT NULL, + state_json TEXT NOT NULL, + created_at TEXT NOT NULL, + updated_at TEXT NOT NULL, + PRIMARY KEY (group_fingerprint, run_id) + ); + CREATE UNIQUE INDEX IF NOT EXISTS autochess_one_active_run + ON autochess_runs(group_fingerprint) + WHERE state IN ('registration', 'active'); + CREATE TABLE IF NOT EXISTS autochess_action_replay ( + group_fingerprint TEXT NOT NULL, + event_id TEXT NOT NULL, + run_id TEXT NOT NULL, + person_key TEXT NOT NULL, + command TEXT NOT NULL, + response_text TEXT NOT NULL, + created_at TEXT NOT NULL, + PRIMARY KEY (group_fingerprint, event_id) + ); + CREATE TABLE IF NOT EXISTS autochess_story_logs ( + group_fingerprint TEXT NOT NULL, + run_id TEXT NOT NULL, + story_id TEXT NOT NULL, + round_number INTEGER NOT NULL, + phase TEXT NOT NULL, + story_type TEXT NOT NULL, + story_text TEXT NOT NULL, + facts_json TEXT NOT NULL, + source_event_id TEXT NOT NULL, + created_at TEXT NOT NULL, + PRIMARY KEY (group_fingerprint, run_id, story_id), + FOREIGN KEY (group_fingerprint, run_id) + REFERENCES autochess_runs(group_fingerprint, run_id) + ON DELETE CASCADE + ); + CREATE TABLE IF NOT EXISTS autochess_battles ( + group_fingerprint TEXT NOT NULL, + run_id TEXT NOT NULL, + battle_id TEXT NOT NULL, + round_number INTEGER NOT NULL, + phase TEXT NOT NULL, + battle_type TEXT NOT NULL, + player_a TEXT NOT NULL, + player_b TEXT NOT NULL, + result_json TEXT NOT NULL, + source_event_id TEXT NOT NULL, + created_at TEXT NOT NULL, + PRIMARY KEY (group_fingerprint, run_id, battle_id), + FOREIGN KEY (group_fingerprint, run_id) + REFERENCES autochess_runs(group_fingerprint, run_id) + ON DELETE CASCADE + ); + CREATE INDEX IF NOT EXISTS autochess_story_run_round + ON autochess_story_logs( + group_fingerprint, run_id, round_number, story_id + ); + CREATE INDEX IF NOT EXISTS autochess_battle_run_round + ON autochess_battles( + group_fingerprint, run_id, round_number, battle_id + ); CREATE INDEX IF NOT EXISTS events_group_created ON events(group_fingerprint, created_at); CREATE INDEX IF NOT EXISTS outbox_due @@ -3634,6 +3699,250 @@ class Ledger: self.connection.execute("ROLLBACK") raise + def _autochess_action_replay( + self, + group: str, + event_id: str, + person_key: str, + command: str, + ) -> str | None: + previous = self.connection.execute( + """ + SELECT person_key, command, response_text + FROM autochess_action_replay + WHERE group_fingerprint = ? AND event_id = ? + """, + (group, event_id), + ).fetchone() + if previous is None: + return None + if previous["person_key"] != person_key or previous["command"] != command: + raise LocalCommandError("同一事件不能变更自走棋玩家或指令") + return str(previous["response_text"]) + + def _load_autochess_run( + self, + group: str, + ) -> tuple[sqlite3.Row | None, dict[str, Any] | None]: + row = self.connection.execute( + """ + SELECT * FROM autochess_runs + WHERE group_fingerprint = ? + ORDER BY + CASE WHEN state IN ('registration', 'active') THEN 0 ELSE 1 END, + season_number DESC + LIMIT 1 + """, + (group,), + ).fetchone() + if row is None: + return None, None + try: + state = json.loads(str(row["state_json"])) + except json.JSONDecodeError as error: + raise LocalCommandError("自走棋状态无效") from error + if not isinstance(state, dict): + raise LocalCommandError("自走棋状态无效") + return row, state + + @staticmethod + def _validate_autochess_state(state: dict[str, Any]) -> None: + status = state.get("status") + if status not in {"registration", "active", "completed", "cancelled"}: + raise LocalCommandError("自走棋赛事状态无效") + if ( + not isinstance(state.get("run_id"), str) + or not state["run_id"] + or type(state.get("season_number")) is not int + or int(state["season_number"]) < 1 + or not isinstance(state.get("host_person_key"), str) + or not state["host_person_key"] + or not isinstance(state.get("players"), dict) + ): + raise LocalCommandError("自走棋赛事结构无效") + if status in {"active", "completed"}: + players = state["players"].values() + if any( + not isinstance(player, dict) + or not isinstance(player.get("board"), list) + or len(player["board"]) != 5 + for player in players + ): + raise LocalCommandError("自走棋比赛必须保持五人首发") + + def apply_autochess_action( + self, + group: str, + sender: str, + event_id: str, + command: str, + action: Callable[ + [dict[str, Any] | None, dict[str, str], int], + dict[str, Any], + ], + ) -> str: + """Commit autochess state, replay, stories and battles atomically.""" + person = self._player_identity(group, sender) + person_key = str(person["person_key"]) + with self.lock: + self.connection.execute("BEGIN IMMEDIATE") + try: + replay = self._autochess_action_replay( + group, + event_id, + person_key, + command, + ) + if replay is not None: + self.connection.execute("COMMIT") + return replay + stored, state = self._load_autochess_run(group) + next_season = int( + self.connection.execute( + """ + SELECT COALESCE(MAX(season_number), 0) + 1 + FROM autochess_runs + WHERE group_fingerprint = ? + """, + (group,), + ).fetchone()[0] + ) + result = action(state, person, next_season) + if not isinstance(result, dict): + raise LocalCommandError("自走棋动作结果无效") + response = result.get("response") + new_state = result.get("state") + stories = result.get("stories", []) + battles = result.get("battles", []) + if not isinstance(response, str) or not response.strip(): + raise LocalCommandError("自走棋响应无效") + if not isinstance(stories, list) or not isinstance(battles, list): + raise LocalCommandError("自走棋归档无效") + timestamp = utc_now() + run_id = str(stored["run_id"]) if stored is not None else "" + if new_state is not None: + if not isinstance(new_state, dict): + raise LocalCommandError("自走棋状态无效") + self._validate_autochess_state(new_state) + run_id = str(new_state["run_id"]) + season_number = int(new_state["season_number"]) + existing = self.connection.execute( + """ + SELECT created_at FROM autochess_runs + WHERE group_fingerprint = ? AND run_id = ? + """, + (group, run_id), + ).fetchone() + created_at = ( + str(existing["created_at"]) + if existing is not None + else timestamp + ) + self.connection.execute( + """ + INSERT INTO autochess_runs ( + group_fingerprint, run_id, season_number, state, + host_person_key, state_json, created_at, updated_at + ) VALUES (?, ?, ?, ?, ?, ?, ?, ?) + ON CONFLICT(group_fingerprint, run_id) DO UPDATE SET + season_number = excluded.season_number, + state = excluded.state, + host_person_key = excluded.host_person_key, + state_json = excluded.state_json, + updated_at = excluded.updated_at + """, + ( + group, + run_id, + season_number, + str(new_state["status"]), + str(new_state["host_person_key"]), + canonical_json(new_state), + created_at, + timestamp, + ), + ) + for item in stories: + if ( + not isinstance(item, dict) + or not isinstance(item.get("story_id"), str) + or not isinstance(item.get("text"), str) + or not isinstance(item.get("facts", {}), dict) + ): + raise LocalCommandError("自走棋剧情归档无效") + self.connection.execute( + """ + INSERT INTO autochess_story_logs ( + group_fingerprint, run_id, story_id, + round_number, phase, story_type, story_text, + facts_json, source_event_id, created_at + ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?) + """, + ( + group, + run_id, + str(item["story_id"]), + int(item.get("round_number", 0)), + str(item.get("phase", "")), + str(item.get("story_type", "")), + str(item["text"]), + canonical_json(item.get("facts", {})), + event_id, + timestamp, + ), + ) + for item in battles: + if ( + not isinstance(item, dict) + or not isinstance(item.get("battle_id"), str) + or not isinstance(item.get("result"), dict) + ): + raise LocalCommandError("自走棋战斗归档无效") + self.connection.execute( + """ + INSERT INTO autochess_battles ( + group_fingerprint, run_id, battle_id, + round_number, phase, battle_type, player_a, + player_b, result_json, source_event_id, created_at + ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) + """, + ( + group, + run_id, + str(item["battle_id"]), + int(item.get("round_number", 0)), + str(item.get("phase", "")), + str(item.get("battle_type", "")), + str(item.get("player_a", "")), + str(item.get("player_b", "")), + canonical_json(item["result"]), + event_id, + timestamp, + ), + ) + self.connection.execute( + """ + INSERT INTO autochess_action_replay ( + group_fingerprint, event_id, run_id, person_key, + command, response_text, created_at + ) VALUES (?, ?, ?, ?, ?, ?, ?) + """, + ( + group, + event_id, + run_id, + person_key, + command, + response, + timestamp, + ), + ) + self.connection.execute("COMMIT") + return response + except BaseException: + self.connection.execute("ROLLBACK") + raise + def status(self) -> dict[str, Any]: with self.lock: events = self.connection.execute("SELECT COUNT(*) FROM events").fetchone()[0] diff --git a/packages/jwebot-official-extensions/pyproject.toml b/packages/jwebot-official-extensions/pyproject.toml index 11e0c3c..b33ff86 100644 --- a/packages/jwebot-official-extensions/pyproject.toml +++ b/packages/jwebot-official-extensions/pyproject.toml @@ -4,11 +4,11 @@ build-backend = "setuptools.build_meta" [project] name = "jwebot-official-extensions" -version = "0.6.3a0+1015" +version = "0.6.4a0+1015" description = "Reviewed local command adapters and broker implementation for JWeBot" requires-python = ">=3.9" dependencies = [ - "jwebot-sdk==0.6.3a0+1015", + "jwebot-sdk==0.6.4a0+1015", ] [tool.setuptools.packages.find] diff --git a/packages/jwebot-official-extensions/src/jwebot_official_extensions/__init__.py b/packages/jwebot-official-extensions/src/jwebot_official_extensions/__init__.py index ec7b657..b9eb31d 100644 --- a/packages/jwebot-official-extensions/src/jwebot_official_extensions/__init__.py +++ b/packages/jwebot-official-extensions/src/jwebot_official_extensions/__init__.py @@ -1,5 +1,17 @@ """Reviewed JWeBot local command adapters and broker implementation.""" +from .cs_autochess import ( + CARD_BY_ID, + COACH_BONUSES, + MAX_PLAYERS, + MIN_PLAYERS, + PLAYER_POOL, + CsMajorAutochess, + build_round_plan, + lineup_power, + matchup_modifier, + parse_autochess_action, +) from .descriptors import ( DIRECT_KERNEL_COMMANDS, DIRECT_KERNEL_COMMAND_KINDS, @@ -35,6 +47,9 @@ from .steam import ( ) __all__ = [ + "CARD_BY_ID", + "COACH_BONUSES", + "CsMajorAutochess", "DICE_CHALLENGE_TIMEOUT_SECONDS", "DEFAULT_STEAM_MIHOMO_PROXY_URL", "DIRECT_KERNEL_COMMANDS", @@ -43,8 +58,12 @@ __all__ = [ "LOCAL_COMMAND_ALIASES", "LOCAL_EXTENSIONS", "LOCAL_GAME_TIMEOUT_SECONDS", + "MAX_PLAYERS", "MAX_RPS_PLAYERS", + "MIN_PLAYERS", "OFFICIAL_AUTHOR", + "PLAYER_POOL", + "build_round_plan", "KernelCommandDescriptor", "LocalExtensionDescriptor", "LocalExtensionError", @@ -58,8 +77,11 @@ __all__ = [ "deterministic_int", "local_command_parts", "local_extension_command_token", + "lineup_power", + "matchup_modifier", + "parse_autochess_action", "extension_for_kind", "render_feature_list", ] -__version__ = "0.6.3a0+1015" +__version__ = "0.6.4a0+1015" diff --git a/packages/jwebot-official-extensions/src/jwebot_official_extensions/adapters/cs_major_autochess.py b/packages/jwebot-official-extensions/src/jwebot_official_extensions/adapters/cs_major_autochess.py new file mode 100644 index 0000000..6e3371e --- /dev/null +++ b/packages/jwebot-official-extensions/src/jwebot_official_extensions/adapters/cs_major_autochess.py @@ -0,0 +1,37 @@ +"""Generated reviewed adapter for cs-major-autochess.""" + +from jwebot_sdk import Event, ExtensionResult, ReplyPlan, SingleGroupContext + + +PROFILE = 'cs-major-autochess' +AUTHOR = 'jwebot' +DESCRIPTION = 'CS自走棋' +HELP_ENTRIES = ('/自走棋开始|报名',) +COMMANDS = ('/自走棋', '/cs自走棋', '/autochess', '/选', '/刷新', '/锁定', '/战术', '/升级教练', '/上阵', '/下阵', '/出售', '/阵容', '/赛程', '/结束回合') + + +def _handler(context: SingleGroupContext, command: str): + def handle(event: Event, arguments: str) -> ExtensionResult: + response = context.services.call( + PROFILE, + operation="execute", + payload={ + "command": command, + "arguments": arguments, + "event": event.to_mapping(), + }, + ) + text = response.get("text") + if not isinstance(text, str) or not text.strip(): + raise ValueError("local service returned invalid text") + return ExtensionResult( + reply=ReplyPlan.text(text), + metadata={"local_extension": PROFILE, "author": AUTHOR}, + ) + + return handle + + +def plugin(context: SingleGroupContext) -> None: + for command in COMMANDS: + context.commands.register(command, _handler(context, command)) diff --git a/packages/jwebot-official-extensions/src/jwebot_official_extensions/cs_autochess.py b/packages/jwebot-official-extensions/src/jwebot_official_extensions/cs_autochess.py new file mode 100644 index 0000000..1cf71e7 --- /dev/null +++ b/packages/jwebot-official-extensions/src/jwebot_official_extensions/cs_autochess.py @@ -0,0 +1,2169 @@ +"""Deterministic CS Major autochess rules for the reviewed local service.""" + +from __future__ import annotations + +import hashlib +import re +from typing import Any + + +MIN_PLAYERS = 2 +MAX_PLAYERS = 8 +AI_PERSON_KEY = "__local_ai_team__" +AI_DISPLAY_NAME = "AI 战队" +HOST_NAME = "Taffy" +STARTING_GOLD = 10 +STARTING_HP = 100 +STARTER_COUNT = 5 +BENCH_LIMIT = 8 +SHOP_SIZE = 6 +REFRESH_COST = 2 +COACH_UPGRADE_COSTS = {1: 6, 2: 10, 3: 16, 4: 24} +COACH_BONUSES = {1: 2, 2: 4, 3: 6, 4: 8, 5: 10} +COST_POOL_COPIES = {1: 18, 2: 15, 3: 12, 4: 9, 5: 7} +SHOP_ODDS = { + 1: {1: 70, 2: 25, 3: 5, 4: 0, 5: 0}, + 2: {1: 50, 2: 30, 3: 18, 4: 2, 5: 0}, + 3: {1: 30, 2: 35, 3: 25, 4: 9, 5: 1}, + 4: {1: 15, 2: 30, 3: 30, 4: 20, 5: 5}, + 5: {1: 5, 2: 20, 3: 30, 4: 30, 5: 15}, +} +TACTICS = ("火力", "稳定", "残局") +TACTIC_COUNTERS = { + "稳定": "火力", + "火力": "残局", + "残局": "稳定", +} +ATTRIBUTE_COUNTERS = ( + ("稳定", "火力"), + ("火力", "残局"), + ("残局", "稳定"), +) +ROLE_COUNTERS = { + "狙击手": "步枪手", + "步枪手": "指挥", + "指挥": "狙击手", +} +PVE_EVENTS = { + "pve_training": ("训练赛", 275, 4, 2), + "pve_media": ("媒体日挑战", 335, 6, 3), + "pve_sponsor": ("赞助挑战", 395, 8, 4), +} + + +def _card( + card_id: str, + name: str, + cost: int, + role: str, + nation: str, + team: str, + history: tuple[str, ...], + firepower: int, + stability: int, + clutch: int, + leadership: int = 0, +) -> dict[str, Any]: + return { + "id": card_id, + "name": name, + "cost": cost, + "role": role, + "nation": nation, + "team": team, + "teams": [team, *history][:3], + "firepower": firepower, + "stability": stability, + "clutch": clutch, + "leadership": leadership, + } + + +PLAYER_POOL = ( + _card("donk", "donk", 5, "步枪手", "俄罗斯", "Spirit", (), 98, 92, 91), + _card("sh1ro", "sh1ro", 4, "狙击手", "俄罗斯", "Spirit", ("Cloud9", "Gambit"), 91, 92, 90), + _card("chopper", "chopper", 3, "指挥", "俄罗斯", "Spirit", ("Vega",), 72, 83, 76, 94), + _card("zont1x", "zont1x", 1, "步枪手", "乌克兰", "Spirit", (), 65, 66, 61), + _card("magixx", "magixx", 1, "步枪手", "俄罗斯", "Spirit", (), 63, 67, 64), + _card("zywoo", "ZywOo", 5, "狙击手", "法国", "Vitality", (), 97, 96, 98), + _card("ropz", "ropz", 4, "步枪手", "爱沙尼亚", "Vitality", ("FaZe", "MOUZ"), 90, 92, 92), + _card("flamez", "flameZ", 3, "步枪手", "以色列", "Vitality", ("OG",), 84, 79, 77), + _card("apex", "apEX", 3, "指挥", "法国", "Vitality", ("EnVyUs", "G2"), 70, 82, 73, 97), + _card("mezii", "mezii", 1, "步枪手", "英国", "Vitality", ("fnatic", "Cloud9"), 66, 68, 65), + _card("m0nesy", "m0NESY", 4, "狙击手", "俄罗斯", "Falcons", ("G2", "NAVI"), 92, 89, 92), + _card("niko", "NiKo", 4, "步枪手", "波黑", "Falcons", ("G2", "FaZe"), 92, 88, 91), + _card("kyousuke", "kyousuke", 4, "步枪手", "俄罗斯", "Falcons", ("Spirit",), 92, 82, 84), + _card("magisk", "Magisk", 3, "步枪手", "丹麦", "Falcons", ("Vitality", "Astralis"), 80, 84, 84), + _card("teses", "TeSeS", 1, "步枪手", "丹麦", "Falcons", ("HEROIC",), 66, 65, 62), + _card("malbsmd", "malbsMd", 3, "步枪手", "危地马拉", "G2", ("M80",), 84, 78, 81), + _card("hunter", "huNter-", 3, "步枪手", "波黑", "G2", (), 79, 84, 82), + _card("heavygod", "HeavyGod", 1, "步枪手", "以色列", "G2", ("Cloud9", "OG"), 67, 66, 63), + _card("snax", "Snax", 2, "指挥", "波兰", "G2", ("GamerLegion", "Virtus.pro"), 62, 73, 72, 88), + _card("sunpayus", "SunPayus", 3, "狙击手", "西班牙", "G2", ("Falcons", "ENCE"), 82, 83, 84), + _card("frozen", "frozen", 4, "步枪手", "斯洛伐克", "FaZe", ("MOUZ",), 89, 91, 88), + _card("broky", "broky", 4, "狙击手", "拉脱维亚", "FaZe", (), 87, 88, 92), + _card("rain", "rain", 3, "步枪手", "挪威", "FaZe", (), 79, 82, 84), + _card("karrigan", "karrigan", 1, "指挥", "丹麦", "FaZe", ("MOUZ", "Astralis"), 54, 64, 60, 98), + _card("elige", "EliGE", 3, "步枪手", "美国", "FaZe", ("Complexity", "Liquid"), 84, 80, 79), + _card("b1t", "b1t", 3, "步枪手", "乌克兰", "NAVI", (), 84, 84, 83), + _card("w0nderful", "w0nderful", 3, "狙击手", "乌克兰", "NAVI", ("Spirit",), 84, 82, 84), + _card("jl", "jL", 3, "步枪手", "立陶宛", "NAVI", ("Apeks",), 82, 81, 84), + _card("aleksib", "Aleksib", 2, "指挥", "芬兰", "NAVI", ("G2", "ENCE"), 64, 76, 70, 95), + _card("im", "iM", 1, "步枪手", "罗马尼亚", "NAVI", ("GamerLegion",), 67, 63, 64), + _card("xertion", "xertioN", 3, "步枪手", "以色列", "MOUZ", (), 84, 78, 80), + _card("torzsi", "torzsi", 3, "狙击手", "匈牙利", "MOUZ", (), 82, 83, 84), + _card("jimpphat", "Jimpphat", 3, "步枪手", "芬兰", "MOUZ", (), 80, 84, 84), + _card("brollan", "Brollan", 1, "步枪手", "瑞典", "MOUZ", ("NIP", "fnatic"), 68, 64, 61), + _card("siuhy", "siuhy", 3, "指挥", "波兰", "Liquid", ("MOUZ", "GamerLegion"), 71, 82, 78, 95), + _card("senzu", "Senzu", 3, "步枪手", "蒙古", "The MongolZ", (), 84, 80, 83), + _card("910", "910", 1, "狙击手", "蒙古", "The MongolZ", (), 68, 65, 66), + _card("blitz", "bLitz", 2, "指挥", "蒙古", "The MongolZ", (), 68, 76, 74, 91), + _card("techno", "Techno", 1, "步枪手", "蒙古", "The MongolZ", (), 66, 67, 62), + _card("mzinho", "mzinho", 1, "步枪手", "蒙古", "The MongolZ", (), 64, 66, 65), + _card("device", "device", 4, "狙击手", "丹麦", "Astralis", ("NIP",), 89, 92, 92), + _card("stavn", "stavn", 3, "步枪手", "丹麦", "Astralis", ("HEROIC",), 83, 82, 81), + _card("jabbi", "jabbi", 3, "步枪手", "丹麦", "Astralis", ("HEROIC",), 80, 81, 80), + _card("hooxi", "HooXi", 2, "指挥", "丹麦", "Astralis", ("G2", "Copenhagen Flames"), 60, 71, 67, 94), + _card("staehr", "Staehr", 1, "步枪手", "丹麦", "Astralis", ("Sprout",), 68, 66, 64), + _card("naf", "NAF", 3, "步枪手", "加拿大", "Liquid", ("Renegades",), 80, 84, 84), + _card("twistzz", "Twistzz", 4, "指挥", "加拿大", "Liquid", ("FaZe",), 87, 90, 91, 92), + _card("nertz", "NertZ", 3, "步枪手", "以色列", "Liquid", ("HEROIC", "ENCE"), 84, 80, 82), + _card("ultimate", "ultimate", 2, "狙击手", "波兰", "Liquid", (), 74, 72, 73), + _card("skullz", "skullz", 1, "步枪手", "巴西", "Liquid", ("FURIA", "paiN"), 65, 67, 64), + _card("kscerato", "KSCERATO", 4, "步枪手", "巴西", "FURIA", (), 90, 88, 89), + _card("yuurih", "yuurih", 3, "步枪手", "巴西", "FURIA", (), 82, 81, 80), + _card("fallen", "FalleN", 3, "指挥", "巴西", "FURIA", ("Liquid", "MIBR"), 70, 80, 83, 93), + _card("molodoy", "molodoy", 2, "狙击手", "哈萨克斯坦", "FURIA", ("AMKAL",), 76, 70, 72), + _card("yekindar", "YEKINDAR", 3, "步枪手", "拉脱维亚", "FURIA", ("Liquid", "Virtus.pro"), 84, 74, 72), + _card("xantares", "XANTARES", 4, "步枪手", "土耳其", "Eternal Fire", ("BIG",), 92, 86, 85), + _card("woxic", "woxic", 3, "狙击手", "土耳其", "Eternal Fire", ("Cloud9", "MOUZ"), 83, 80, 82), + _card("maj3r", "MAJ3R", 2, "指挥", "土耳其", "Eternal Fire", ("Space Soldiers",), 68, 74, 71, 93), + _card("wicadia", "Wicadia", 2, "步枪手", "土耳其", "Eternal Fire", (), 76, 70, 68), + _card("jottaaa", "jottAAA", 1, "步枪手", "土耳其", "Eternal Fire", ("Sangal",), 67, 64, 63), +) +CARD_BY_ID = {card["id"]: card for card in PLAYER_POOL} + + +def _hash_int(seed: str) -> int: + return int.from_bytes( + hashlib.sha256(seed.encode("utf-8", "replace")).digest()[:8], + "big", + ) + + +def _round_robin_rounds(keys: list[str]) -> list[list[list[str]]]: + rotation: list[str | None] = list(keys) + if len(rotation) % 2: + rotation.append(None) + rounds: list[list[list[str]]] = [] + for _ in range(len(rotation) - 1): + pairs: list[list[str]] = [] + for index in range(len(rotation) // 2): + first = rotation[index] + second = rotation[-index - 1] + if first is not None and second is not None: + pairs.append([first, second]) + rounds.append(pairs) + rotation = [rotation[0], rotation[-1], *rotation[1:-1]] + return rounds + + +def build_round_plan(keys: list[str]) -> list[dict[str, Any]]: + if not MIN_PLAYERS <= len(keys) <= MAX_PLAYERS: + raise ValueError("参赛人数必须为 2～8 人") + plan: list[dict[str, Any]] = [ + { + "kind": "pve", + "phase": "pve_training", + "name": PVE_EVENTS["pve_training"][0], + } + ] + if len(keys) in {2, 4}: + for phase in ("pve_media", "pve_sponsor"): + plan.append( + { + "kind": "pve", + "phase": phase, + "name": PVE_EVENTS[phase][0], + } + ) + else: + group_rounds = _round_robin_rounds(keys) + media_after = max(1, (len(group_rounds) + 2) // 3) + sponsor_after = max( + media_after + 1, + (len(group_rounds) * 2 + 2) // 3, + ) + for day, pairs in enumerate(group_rounds, start=1): + item = { + "kind": "group", + "phase": f"group_{day}", + "name": f"小组赛第 {day} 轮", + "day": day, + "pairs": pairs, + "last_group_day": day == len(group_rounds), + } + plan.append(item) + if day == media_after: + plan.append( + { + "kind": "pve", + "phase": "pve_media", + "name": PVE_EVENTS["pve_media"][0], + } + ) + if day == sponsor_after: + plan.append( + { + "kind": "pve", + "phase": "pve_sponsor", + "name": PVE_EVENTS["pve_sponsor"][0], + } + ) + existing_pve = { + str(item["phase"]) + for item in plan + if item["kind"] == "pve" + } + for phase in ("pve_media", "pve_sponsor"): + if phase not in existing_pve: + plan.append( + { + "kind": "pve", + "phase": phase, + "name": PVE_EVENTS[phase][0], + } + ) + if len(keys) == 4: + plan.append( + { + "kind": "semifinal", + "phase": "semifinal", + "name": "淘汰赛半决赛（1v4、2v3）", + "pairs": [[keys[0], keys[3]], [keys[1], keys[2]]], + "pairing": "seed", + } + ) + elif len(keys) > 4: + plan.append( + { + "kind": "semifinal", + "phase": "semifinal", + "name": "淘汰赛半决赛（1v4、2v3）", + "pairs": [], + "pairing": "standings", + } + ) + plan.append( + { + "kind": "final", + "phase": "final", + "name": "总决赛", + } + ) + return plan + + +def _current_stage(state: dict[str, Any]) -> dict[str, Any]: + round_number = int(state["round_number"]) + plan = state.get("round_plan", []) + if not isinstance(plan, list) or not 1 <= round_number <= len(plan): + return {"kind": "complete", "phase": "complete", "name": "赛事结束"} + stage = plan[round_number - 1] + return stage if isinstance(stage, dict) else { + "kind": "complete", + "phase": "complete", + "name": "赛事结束", + } + + +def _host_text(text: str) -> str: + return text if text.startswith(f"{HOST_NAME}：") else f"{HOST_NAME}：{text}" + + +def _human_keys(state: dict[str, Any]) -> list[str]: + return [ + key + for key in state["seed_order"] + if not state["players"][key].get("is_ai") + ] + + +def _reset_preparation(state: dict[str, Any]) -> None: + human_keys = _human_keys(state) + if not human_keys: + state["preparation_order"] = [] + state["preparation_cursor"] = 0 + return + offset = (int(state["round_number"]) - 1) % len(human_keys) + order = [*human_keys[offset:], *human_keys[:offset]] + state["preparation_order"] = order + state["preparation_cursor"] = 0 + for key in state["join_order"]: + state["players"][key]["ready"] = bool( + state["players"][key].get("is_ai") + ) + + +def _current_preparer(state: dict[str, Any]) -> str | None: + order = state.get("preparation_order", []) + cursor = int(state.get("preparation_cursor", 0)) + if not isinstance(order, list) or not 0 <= cursor < len(order): + return None + return str(order[cursor]) + + +def _preparation_prompt(state: dict[str, Any]) -> str: + current = _current_preparer(state) + if current is None: + return _host_text("全员准备完毕，主持人可以发送 /自走棋结算。") + name = state["players"][current]["display_name"] + return _host_text( + f"{name}，请做好赛前准备；完成后发送 /结束回合。" + ) + + +def _format_description(state: dict[str, Any]) -> str: + team_count = len(state["join_order"]) + if state.get("ai_test_mode"): + return ( + "本届采用 3 人+1 AI 测试赛：报名种子 1v4、2v3 进行 BO5 " + "半决赛，胜者进入 BO5 总决赛" + ) + if team_count == 2: + return "本届两队在三轮赛前活动后直接进行 BO5 总决赛" + if team_count == 3: + return "本届三队进行 BO3 单循环，前两名进入 BO5 总决赛" + if team_count == 4: + return "本届按报名种子 1v4、2v3 进行 BO5 半决赛和总决赛" + return ( + f"本届 {team_count} 队进行 BO3 单循环，前四名按 1v4、2v3 " + "进入 BO5 淘汰赛" + ) + + +def _shop_tier(state: dict[str, Any]) -> int: + round_number = int(state["round_number"]) + total_rounds = max(1, len(state.get("round_plan", []))) + if total_rounds == 1: + return 5 + return min(5, 1 + (round_number - 1) * 5 // total_rounds) + + +def _new_unit(card_id: str, copies: int = 1) -> dict[str, Any]: + return { + "card_id": card_id, + "copies": copies, + "star": 3 if copies >= 9 else 2 if copies >= 3 else 1, + } + + +def _normalize_unit(unit: dict[str, Any]) -> None: + copies = int(unit.get("copies", 1)) + unit["copies"] = copies + unit["star"] = 3 if copies >= 9 else 2 if copies >= 3 else 1 + + +def _unit_power(unit: dict[str, Any]) -> float: + card = CARD_BY_ID[str(unit["card_id"])] + base = ( + card["firepower"] * 0.45 + + card["stability"] * 0.30 + + card["clutch"] * 0.25 + ) + if card["role"] == "指挥": + base += max(0, card["leadership"] - 70) * 0.12 + multiplier = {1: 1.0, 2: 1.35, 3: 1.80}[int(unit["star"])] + return base * multiplier + + +def lineup_power(player: dict[str, Any]) -> dict[str, Any]: + board = player.get("board", []) + if len(board) != STARTER_COUNT: + raise ValueError("比赛阵容必须正好有 5 名首发") + base = sum(_unit_power(unit) for unit in board) + teams: dict[str, int] = {} + nations: dict[str, int] = {} + axes = {"火力": 0.0, "稳定": 0.0, "残局": 0.0} + roles = {"步枪手": 0, "狙击手": 0, "指挥": 0} + for unit in board: + card = CARD_BY_ID[str(unit["card_id"])] + multiplier = {1: 1.0, 2: 1.35, 3: 1.80}[int(unit["star"])] + axes["火力"] += float(card["firepower"]) * multiplier + axes["稳定"] += float(card["stability"]) * multiplier + axes["残局"] += float(card["clutch"]) * multiplier + roles[str(card["role"])] += 1 + nations[card["nation"]] = nations.get(card["nation"], 0) + 1 + for team in card["teams"]: + teams[team] = teams.get(team, 0) + 1 + team_values = [] + for name, count in teams.items(): + bonus = 22 if count >= 5 else 15 if count >= 4 else 9 if count >= 3 else 4 if count >= 2 else 0 + if bonus: + team_values.append((bonus, count, name)) + team_values.sort(reverse=True) + team_bonus = sum(item[0] for item in team_values[:2]) + nation_values = [] + for name, count in nations.items(): + bonus = 18 if count >= 5 else 12 if count >= 4 else 7 if count >= 3 else 3 if count >= 2 else 0 + if bonus: + nation_values.append((bonus, count, name)) + nation_values.sort(reverse=True) + nation_bonus = nation_values[0][0] if nation_values else 0 + coach_bonus = COACH_BONUSES[int(player["coach_level"])] + total = base * (100 + team_bonus + nation_bonus + coach_bonus) / 100 + labels = [ + *(f"{name}{count}" for _, count, name in team_values[:2]), + *(f"{name}{count}" for _, count, name in nation_values[:1]), + f"教练+{coach_bonus}%", + ] + return { + "power": round(total, 1), + "base": round(base, 1), + "team_bonus": team_bonus, + "nation_bonus": nation_bonus, + "coach_bonus": coach_bonus, + "synergies": labels, + "axes": {name: round(value, 1) for name, value in axes.items()}, + "roles": roles, + } + + +def matchup_modifier( + player_a: dict[str, Any], + player_b: dict[str, Any], +) -> dict[str, Any]: + info_a = lineup_power(player_a) + info_b = lineup_power(player_b) + axes_a = dict(info_a["axes"]) + axes_b = dict(info_b["axes"]) + tactic_a = str(player_a.get("tactic", "稳定")) + tactic_b = str(player_b.get("tactic", "稳定")) + axes_a[tactic_a] *= 1.08 + axes_b[tactic_b] *= 1.08 + + def pressure( + own: dict[str, float], + opponent: dict[str, float], + ) -> float: + return sum( + max(0.0, own[strong] - opponent[weak]) + for strong, weak in ATTRIBUTE_COUNTERS + ) + + axis_net = pressure(axes_a, axes_b) - pressure(axes_b, axes_a) + axis_percent = max(-8.0, min(8.0, axis_net / 12.5)) + roles_a = info_a["roles"] + roles_b = info_b["roles"] + role_edge_a = sum( + int(roles_a[strong]) * int(roles_b[weak]) + for strong, weak in ROLE_COUNTERS.items() + ) + role_edge_b = sum( + int(roles_b[strong]) * int(roles_a[weak]) + for strong, weak in ROLE_COUNTERS.items() + ) + role_percent = max( + -5.0, + min(5.0, (role_edge_a - role_edge_b) * 0.8), + ) + tactic_percent = 0.0 + if TACTIC_COUNTERS[tactic_a] == tactic_b: + tactic_percent = 6.0 + elif TACTIC_COUNTERS[tactic_b] == tactic_a: + tactic_percent = -6.0 + total = max( + -15.0, + min(15.0, axis_percent + role_percent + tactic_percent), + ) + return { + "a_total_percent": round(total, 2), + "b_total_percent": round(-total, 2), + "axis_percent": round(axis_percent, 2), + "role_percent": round(role_percent, 2), + "tactic_percent": round(tactic_percent, 2), + "tactic_a": tactic_a, + "tactic_b": tactic_b, + } + + +def _draw_shop(state: dict[str, Any], seed: str) -> list[str]: + tier = _shop_tier(state) + odds = SHOP_ODDS[tier] + selected: list[str] = [] + for slot in range(SHOP_SIZE): + available_costs = [ + cost + for cost, weight in odds.items() + if weight > 0 + and any( + card["cost"] == cost + and int(state["pool"].get(card["id"], 0)) > 0 + and card["id"] not in selected + for card in PLAYER_POOL + ) + ] + if not available_costs: + break + total_weight = sum(odds[cost] for cost in available_costs) + roll = _hash_int(f"{seed}|cost|{slot}") % total_weight + chosen_cost = available_costs[-1] + for cost in available_costs: + if roll < odds[cost]: + chosen_cost = cost + break + roll -= odds[cost] + candidates = [ + card["id"] + for card in PLAYER_POOL + if card["cost"] == chosen_cost + and int(state["pool"].get(card["id"], 0)) > 0 + and card["id"] not in selected + ] + candidates.sort( + key=lambda card_id: _hash_int(f"{seed}|card|{slot}|{card_id}") + ) + selected.append(candidates[0]) + return selected + + +def _story( + state: dict[str, Any], + story_type: str, + text: str, + facts: dict[str, Any] | None = None, +) -> dict[str, Any]: + state["story_sequence"] = int(state.get("story_sequence", 0)) + 1 + hosted_text = _host_text(text) + item = { + "story_id": f"story_{state['story_sequence']:04d}", + "round_number": int(state.get("round_number", 0)), + "phase": str(state.get("phase", "registration")), + "story_type": story_type, + "text": hosted_text, + "facts": facts or {}, + } + state.setdefault("stories", []).append(item) + return item + + +def _battle_record( + state: dict[str, Any], + *, + battle_type: str, + player_a: str, + player_b: str, + result: dict[str, Any], +) -> dict[str, Any]: + state["battle_sequence"] = int(state.get("battle_sequence", 0)) + 1 + item = { + "battle_id": f"battle_{state['battle_sequence']:04d}", + "round_number": int(state["round_number"]), + "phase": str(state["phase"]), + "battle_type": battle_type, + "player_a": player_a, + "player_b": player_b, + "result": result, + } + state.setdefault("battles", []).append(item) + return item + + +def _result( + state: dict[str, Any] | None, + response: str, + *, + stories: list[dict[str, Any]] | None = None, + battles: list[dict[str, Any]] | None = None, +) -> dict[str, Any]: + return { + "state": state, + "response": response, + "stories": stories or [], + "battles": battles or [], + } + + +def _participant( + person_key: str, + display_name: str, + *, + is_ai: bool = False, +) -> dict[str, Any]: + return { + "person_key": person_key, + "display_name": display_name, + "hp": STARTING_HP, + "gold": STARTING_GOLD, + "coach_level": 1, + "board": [], + "bench": [], + "shop": [], + "shop_locked": False, + "ready": False, + "wins": 0, + "losses": 0, + "byes": 0, + "streak": 0, + "opponents": [], + "placement": None, + "tactic": "稳定", + "is_ai": is_ai, + } + + +def _assign_starters(state: dict[str, Any], player: dict[str, Any]) -> None: + candidates = [ + card["id"] + for card in PLAYER_POOL + if card["cost"] == 1 and int(state["pool"].get(card["id"], 0)) > 0 + ] + candidates.sort( + key=lambda card_id: _hash_int( + f"{state['run_id']}|starter|{player['person_key']}|{card_id}" + ) + ) + if len(candidates) < STARTER_COUNT: + raise ValueError("低费选手池不足，无法生成完整首发") + for card_id in candidates[:STARTER_COUNT]: + state["pool"][card_id] -= 1 + player["board"].append(_new_unit(card_id)) + + +def _player_prefix(state: dict[str, Any], player: dict[str, Any]) -> str: + return f"{player['display_name']} #{state['season_number']}：" + + +def _card_line(slot: int, card_id: str | None) -> str: + if card_id is None: + return f"{slot}. 已购买" + card = CARD_BY_ID[card_id] + history = [team for team in card["teams"] if team != card["team"]] + history_text = "/".join(history) if history else "无" + leadership = ( + f" 指挥{card['leadership']}" if card["role"] == "指挥" else "" + ) + return ( + f"{slot}. {card['name']}｜{card['cost']}费 {card['role']}｜" + f"{card['nation']}｜{card['team']}（历史 {history_text}）｜" + f"火{card['firepower']} 稳{card['stability']} 残{card['clutch']}" + f"{leadership}" + ) + + +def _shop_text(state: dict[str, Any], player: dict[str, Any]) -> str: + phase_name = str(_current_stage(state)["name"]) + power = lineup_power(player) + lock = "已锁" if player["shop_locked"] else "未锁" + lines = [ + ( + f"{_player_prefix(state, player)}第 {state['round_number']} 回合" + f" {phase_name}；金币 {player['gold']}；生命 {player['hp']}；" + f"教练 {player['coach_level']} 级；战术{player['tactic']}；" + f"阵容 {power['power']:.1f}；商店{lock}" + ) + ] + lines.extend( + _card_line(index, card_id) + for index, card_id in enumerate(player["shop"], start=1) + ) + lines.append("操作：/选1～/选6；/战术 火力|稳定|残局；/阵容；/结束回合") + return "\n".join(lines) + + +def _roster_entries(player: dict[str, Any]) -> list[tuple[str, int, dict[str, Any]]]: + return [ + *(("首发", index, unit) for index, unit in enumerate(player["board"], 1)), + *(("备战", index, unit) for index, unit in enumerate(player["bench"], 6)), + ] + + +def _lineup_text(state: dict[str, Any], player: dict[str, Any]) -> str: + power = lineup_power(player) + lines = [ + ( + f"{_player_prefix(state, player)}固定 5 人首发 + 教练；" + f"战力 {power['power']:.1f}（基础 {power['base']:.1f}）；" + f"战术 {player['tactic']}；羁绊 {'、'.join(power['synergies'])}；" + f"属性 火{power['axes']['火力']:.0f}/稳{power['axes']['稳定']:.0f}/" + f"残{power['axes']['残局']:.0f}" + ) + ] + for scope, index, unit in _roster_entries(player): + card = CARD_BY_ID[str(unit["card_id"])] + lines.append( + f"{index}. [{scope}] {card['name']} {unit['star']}星×{unit['copies']} " + f"{card['cost']}费 {card['role']} {card['team']}/{card['nation']} " + f"单卡{_unit_power(unit):.1f}" + ) + if not player["bench"]: + lines.append("备战席为空") + lines.append("替换：/上阵 备战编号 [首发位]；出售仅限备战席：/出售 编号") + return "\n".join(lines) + + +def _find_unit( + player: dict[str, Any], + index: int, +) -> tuple[str, int, dict[str, Any]] | None: + for scope, display_index, unit in _roster_entries(player): + if display_index == index: + internal = display_index - 1 if scope == "首发" else display_index - 6 + return scope, internal, unit + return None + + +def _merge_purchase(player: dict[str, Any], card_id: str) -> tuple[int, bool]: + for unit in [*player["board"], *player["bench"]]: + if unit["card_id"] == card_id: + old_star = int(unit["star"]) + unit["copies"] = int(unit["copies"]) + 1 + _normalize_unit(unit) + return int(unit["star"]), int(unit["star"]) > old_star + if len(player["bench"]) >= BENCH_LIMIT: + return 0, False + player["bench"].append(_new_unit(card_id)) + return 1, False + + +def _ai_opponent(state: dict[str, Any], ai_key: str) -> str | None: + stage = _current_stage(state) + pairs = [ + (str(pair[0]), str(pair[1])) + for pair in stage.get("pairs", []) + ] + if stage.get("kind") == "semifinal" and not pairs: + qualifiers = list(state.get("qualifiers") or _rankings(state)[:4]) + if len(qualifiers) == 4: + pairs = [ + (qualifiers[0], qualifiers[3]), + (qualifiers[1], qualifiers[2]), + ] + if stage.get("kind") == "final": + finalists = list(state.get("finalists", [])) + if len(finalists) != 2: + finalists = list(state.get("qualifiers", []))[:2] + if len(finalists) == 2: + pairs = [(finalists[0], finalists[1])] + for first, second in pairs: + if first == ai_key: + return second + if second == ai_key: + return first + return None + + +def _auto_manage_ai( + state: dict[str, Any], +) -> list[dict[str, Any]]: + stories: list[dict[str, Any]] = [] + for ai_key in state["join_order"]: + player = state["players"][ai_key] + if not player.get("is_ai"): + continue + actions: list[str] = [] + opponent_key = _ai_opponent(state, ai_key) + if opponent_key is not None: + opponent_tactic = str( + state["players"][opponent_key].get("tactic", "稳定") + ) + player["tactic"] = next( + tactic + for tactic, target in TACTIC_COUNTERS.items() + if target == opponent_tactic + ) + actions.append(f"战术{player['tactic']}") + else: + player["tactic"] = TACTICS[ + (int(state["round_number"]) - 1) % len(TACTICS) + ] + actions.append(f"战术{player['tactic']}") + + level = int(player["coach_level"]) + if level < 5: + upgrade_cost = COACH_UPGRADE_COSTS[level] + if int(player["gold"]) >= upgrade_cost + REFRESH_COST: + player["gold"] -= upgrade_cost + player["coach_level"] = level + 1 + actions.append(f"教练{level + 1}级") + + candidates = [] + owned = { + str(unit["card_id"]) + for unit in [*player["board"], *player["bench"]] + } + for slot, card_id in enumerate(player["shop"]): + if card_id is None: + continue + card = CARD_BY_ID[str(card_id)] + if ( + int(card["cost"]) <= int(player["gold"]) + and int(state["pool"].get(card_id, 0)) > 0 + and (card_id in owned or len(player["bench"]) < BENCH_LIMIT) + ): + candidates.append( + ( + card_id in owned, + int(card["cost"]), + int(card["firepower"]) + + int(card["stability"]) + + int(card["clutch"]), + -slot, + slot, + str(card_id), + ) + ) + if candidates: + *_, slot, card_id = max(candidates) + card = CARD_BY_ID[card_id] + player["gold"] -= int(card["cost"]) + state["pool"][card_id] -= 1 + star, _ = _merge_purchase(player, card_id) + player["shop"][slot] = None + actions.append(f"签下{card['name']}({star}星)") + + if player["bench"]: + strongest_index = max( + range(len(player["bench"])), + key=lambda index: _unit_power(player["bench"][index]), + ) + weakest_index = min( + range(len(player["board"])), + key=lambda index: _unit_power(player["board"][index]), + ) + if ( + _unit_power(player["bench"][strongest_index]) + > _unit_power(player["board"][weakest_index]) + ): + incoming = player["bench"][strongest_index] + outgoing = player["board"][weakest_index] + player["board"][weakest_index] = incoming + player["bench"][strongest_index] = outgoing + actions.append( + "换上" + + CARD_BY_ID[str(incoming["card_id"])]["name"] + ) + text = ( + f"{AI_DISPLAY_NAME} 自动运营：{'、'.join(actions)}；" + f"金币 {player['gold']}，五人首发战力 " + f"{lineup_power(player)['power']:.1f}。" + ) + stories.append( + _story( + state, + "ai_operation", + text, + { + "person_key": ai_key, + "actions": actions, + "gold": player["gold"], + "tactic": player["tactic"], + }, + ) + ) + return stories + + +MAP_POOL = ("Mirage", "Inferno", "Nuke", "Ancient", "Dust2") + + +def _map_score(margin: float, seed: str) -> tuple[str, str, int]: + if margin >= 0.18: + return "16-6", "碾压", 0 + if margin >= 0.11: + return "16-9", "明显优势", 0 + if margin >= 0.055: + return "16-12", "小胜", 0 + if margin >= 0.025: + return "16-14", "险胜", 0 + overtime_count = 1 + _hash_int(seed + "|overtime") % 2 + winner_score = 16 + overtime_count * 3 + return f"{winner_score}-{winner_score - 2}", "加时险胜", overtime_count + + +def _battle( + state: dict[str, Any], + key_a: str, + key_b: str, + seed: str, + *, + best_of: int, +) -> dict[str, Any]: + player_a = state["players"][key_a] + player_b = state["players"][key_b] + raw_a = lineup_power(player_a)["power"] + raw_b = lineup_power(player_b)["power"] + matchup = matchup_modifier(player_a, player_b) + target_maps = best_of // 2 + 1 + wins_a = 0 + wins_b = 0 + maps: list[dict[str, Any]] = [] + for map_index in range(1, best_of + 1): + fluct_a = _hash_int(f"{seed}|map|{map_index}|{key_a}") % 401 - 200 + fluct_b = _hash_int(f"{seed}|map|{map_index}|{key_b}") % 401 - 200 + adjusted_a = ( + raw_a + * (100 + float(matchup["a_total_percent"])) + / 100 + * (10_000 + fluct_a) + / 10_000 + ) + adjusted_b = ( + raw_b + * (100 + float(matchup["b_total_percent"])) + / 100 + * (10_000 + fluct_b) + / 10_000 + ) + if adjusted_a == adjusted_b: + adjusted_a += ( + 0.01 if _hash_int(f"{seed}|map|{map_index}|tie") % 2 == 0 else -0.01 + ) + winner_key = key_a if adjusted_a > adjusted_b else key_b + if winner_key == key_a: + wins_a += 1 + else: + wins_b += 1 + high = max(adjusted_a, adjusted_b) + low = min(adjusted_a, adjusted_b) + margin = (high - low) / max(high, 1) + score, band, overtime_count = _map_score( + margin, + f"{seed}|map|{map_index}", + ) + winner_name = state["players"][winner_key]["display_name"] + loser_key = key_b if winner_key == key_a else key_a + loser_name = state["players"][loser_key]["display_name"] + map_name = MAP_POOL[map_index - 1] + overtime_story = "" + if overtime_count: + overtime_story = ( + f"{map_name} 常规时间战成 15-15。" + f"{winner_name} 在第 {overtime_count} 轮加时抓住" + f"{loser_name} 的转点空隙，以 {score} 收下地图。" + ) + maps.append( + { + "map_index": map_index, + "map_name": map_name, + "winner_key": winner_key, + "winner": winner_name, + "loser_key": loser_key, + "loser": loser_name, + "score": score, + "band": band, + "overtime_count": overtime_count, + "overtime_story": overtime_story, + "adjusted_a": round(adjusted_a, 2), + "adjusted_b": round(adjusted_b, 2), + "fluctuation_a_bps": fluct_a, + "fluctuation_b_bps": fluct_b, + } + ) + if wins_a == target_maps or wins_b == target_maps: + break + winner_key, loser_key = ( + (key_a, key_b) if wins_a > wins_b else (key_b, key_a) + ) + winner_maps = max(wins_a, wins_b) + loser_maps = min(wins_a, wins_b) + sweep = loser_maps == 0 + damage = 18 if sweep and best_of == 5 else 15 if sweep else 10 + return { + "winner_key": winner_key, + "loser_key": loser_key, + "player_a": player_a["display_name"], + "player_b": player_b["display_name"], + "winner": state["players"][winner_key]["display_name"], + "loser": state["players"][loser_key]["display_name"], + "best_of": best_of, + "series_format": f"BO{best_of}", + "score": f"{winner_maps}-{loser_maps}", + "band": "横扫" if sweep else "鏖战", + "damage": damage, + "power_a": round(raw_a, 1), + "power_b": round(raw_b, 1), + "matchup": matchup, + "maps": maps, + } + + +def _apply_match_result( + state: dict[str, Any], + key_a: str, + key_b: str, + result: dict[str, Any], + *, + group_stage: bool, +) -> None: + winner_key = str(result["winner_key"]) + loser_key = str(result["loser_key"]) + winner = state["players"][winner_key] + loser = state["players"][loser_key] + loser["hp"] = max(0, int(loser["hp"]) - int(result["damage"])) + winner["streak"] = max(1, int(winner["streak"]) + 1) + loser["streak"] = min(-1, int(loser["streak"]) - 1) + if group_stage: + winner["wins"] = int(winner["wins"]) + 1 + loser["losses"] = int(loser["losses"]) + 1 + state.setdefault("group_results", []).append( + { + "winner_key": winner_key, + "loser_key": loser_key, + "score": str(result["score"]), + } + ) + for owner, opponent in ((key_a, key_b), (key_b, key_a)): + state["players"][owner].setdefault("opponents", []).append(opponent) + + +def _rankings(state: dict[str, Any]) -> list[str]: + keys = list(state["join_order"]) + seed_order = { + key: index + for index, key in enumerate(state.get("seed_order", keys), start=1) + } + wins_by_key = { + key: int(state["players"][key]["wins"]) + for key in keys + } + + def head_to_head_wins(key: str) -> int: + tied = { + other + for other in keys + if wins_by_key[other] == wins_by_key[key] + } + return sum( + 1 + for result in state.get("group_results", []) + if result.get("winner_key") == key + and result.get("loser_key") in tied + ) + + keys.sort( + key=lambda key: ( + -wins_by_key[key], + -head_to_head_wins(key), + -lineup_power(state["players"][key])["power"], + -int(state["players"][key]["hp"]), + seed_order.get(key, 999), + ) + ) + return keys + + +def _battle_story(result: dict[str, Any]) -> str: + map_scores = "、".join( + f"{item['map_name']} {item['score']}" + for item in result["maps"] + ) + matchup = result["matchup"] + return _host_text( + f"{result['winner']} 在 {result['series_format']} 以 {result['score']} " + f"击败 {result['loser']}（{map_scores}）；" + f"战术 {matchup['tactic_a']} 对 {matchup['tactic_b']}，" + f"{result['player_a']} 克制修正 {matchup['a_total_percent']:+.2f}%。" + ) + + +def _append_overtime_stories( + state: dict[str, Any], + result: dict[str, Any], + stories: list[dict[str, Any]], +) -> None: + for item in result["maps"]: + if int(item["overtime_count"]) <= 0: + continue + stories.append( + _story( + state, + "overtime", + str(item["overtime_story"]), + { + "map_name": item["map_name"], + "score": item["score"], + "overtime_count": item["overtime_count"], + "winner_key": item["winner_key"], + "loser_key": item["loser_key"], + }, + ) + ) + + +def _settle_pve( + state: dict[str, Any], + event_id: str, +) -> tuple[list[str], list[dict[str, Any]], list[dict[str, Any]]]: + stage = _current_stage(state) + event_name, target, win_gold, loss_gold = PVE_EVENTS[str(stage["phase"])] + lines: list[str] = [] + stories: list[dict[str, Any]] = [] + battles: list[dict[str, Any]] = [] + for key in state["join_order"]: + player = state["players"][key] + raw_power = lineup_power(player)["power"] + fluctuation = _hash_int( + f"{state['run_id']}|{state['round_number']}|{event_id}|{key}|pve" + ) % 401 - 200 + adjusted = raw_power * (10_000 + fluctuation) / 10_000 + won = adjusted >= target + reward = win_gold if won else loss_gold + player["gold"] = int(player["gold"]) + reward + result = { + "winner_key": key if won else "event", + "loser_key": "event" if won else key, + "winner": player["display_name"] if won else "赛事活动", + "loser": "赛事活动" if won else player["display_name"], + "score": "达标" if won else "未达标", + "band": "资源挑战", + "damage": 0, + "player_power": round(adjusted, 2), + "target_power": target, + "reward": reward, + "fluctuation_bps": fluctuation, + } + battles.append( + _battle_record( + state, + battle_type="pve", + player_a=key, + player_b="event", + result=result, + ) + ) + text = ( + f"{player['display_name']} {'完成' if won else '未完成'}" + f"{event_name}目标 {adjusted:.1f}/{target}，获得 {reward} 金币。" + ) + stories.append(_story(state, "pve_result", text, result)) + lines.append(_host_text(text)) + return lines, stories, battles + + +def _settle_group( + state: dict[str, Any], + event_id: str, +) -> tuple[list[str], list[dict[str, Any]], list[dict[str, Any]]]: + stage = _current_stage(state) + pairs = [ + (str(pair[0]), str(pair[1])) + for pair in stage.get("pairs", []) + ] + paired = {key for pair in pairs for key in pair} + bye = next( + (key for key in state["join_order"] if key not in paired), + None, + ) + lines: list[str] = [] + stories: list[dict[str, Any]] = [] + battles: list[dict[str, Any]] = [] + if bye is not None: + player = state["players"][bye] + player["byes"] = int(player["byes"]) + 1 + text = f"{player['display_name']} 本轮轮空休整，不增加小组胜场。" + stories.append(_story(state, "group_bye", text, {"person_key": bye})) + lines.append(text) + for index, (key_a, key_b) in enumerate(pairs, start=1): + result = _battle( + state, + key_a, + key_b, + f"{state['run_id']}|group|{stage['day']}|{event_id}|{index}", + best_of=3, + ) + _apply_match_result( + state, + key_a, + key_b, + result, + group_stage=True, + ) + battles.append( + _battle_record( + state, + battle_type="group_bo3", + player_a=key_a, + player_b=key_b, + result=result, + ) + ) + text = _battle_story(result) + stories.append(_story(state, "group_result", text, result)) + _append_overtime_stories(state, result, stories) + lines.append(text) + lines.extend( + _host_text(str(item["overtime_story"])) + for item in result["maps"] + if int(item["overtime_count"]) > 0 + ) + if stage.get("last_group_day"): + ranking = _rankings(state) + qualifier_count = 2 if len(ranking) == 3 else 4 + state["qualifiers"] = ranking[:qualifier_count] + names = "、".join( + state["players"][key]["display_name"] + for key in state["qualifiers"] + ) + text = f"小组单循环结束，晋级淘汰赛：{names}。" + stories.append( + _story( + state, + "group_complete", + text, + { + "ranking": ranking, + "qualifiers": list(state["qualifiers"]), + }, + ) + ) + lines.append(text) + return lines, stories, battles + + +def _settle_semifinal( + state: dict[str, Any], + event_id: str, +) -> tuple[list[str], list[dict[str, Any]], list[dict[str, Any]]]: + stage = _current_stage(state) + if stage.get("pairing") == "seed": + pairs = [ + (str(pair[0]), str(pair[1])) + for pair in stage.get("pairs", []) + ] + state["qualifiers"] = list(state["seed_order"]) + else: + qualifiers = list(state.get("qualifiers") or _rankings(state)[:4]) + pairs = [ + (qualifiers[0], qualifiers[3]), + (qualifiers[1], qualifiers[2]), + ] + lines: list[str] = [] + stories: list[dict[str, Any]] = [] + battles: list[dict[str, Any]] = [] + finalists: list[str] = [] + for index, (key_a, key_b) in enumerate(pairs, start=1): + result = _battle( + state, + key_a, + key_b, + f"{state['run_id']}|semifinal|{event_id}|{index}", + best_of=5, + ) + _apply_match_result( + state, + key_a, + key_b, + result, + group_stage=False, + ) + finalists.append(str(result["winner_key"])) + battles.append( + _battle_record( + state, + battle_type="semifinal_bo5", + player_a=key_a, + player_b=key_b, + result=result, + ) + ) + text = _battle_story(result) + stories.append(_story(state, "semifinal_result", text, result)) + _append_overtime_stories(state, result, stories) + lines.append(text) + lines.extend( + _host_text(str(item["overtime_story"])) + for item in result["maps"] + if int(item["overtime_count"]) > 0 + ) + state["finalists"] = finalists + return lines, stories, battles + + +def _settle_final( + state: dict[str, Any], + event_id: str, +) -> tuple[list[str], list[dict[str, Any]], list[dict[str, Any]]]: + finalists = list(state.get("finalists", [])) + if len(finalists) != 2: + if len(state["join_order"]) == 2: + finalists = list(state["seed_order"]) + else: + finalists = list(state.get("qualifiers") or _rankings(state)[:2])[:2] + key_a, key_b = finalists + result = _battle( + state, + key_a, + key_b, + f"{state['run_id']}|final|{event_id}", + best_of=5, + ) + _apply_match_result( + state, + key_a, + key_b, + result, + group_stage=False, + ) + champion_key = str(result["winner_key"]) + state["champion_key"] = champion_key + state["status"] = "completed" + battles = [ + _battle_record( + state, + battle_type="final_bo5", + player_a=key_a, + player_b=key_b, + result=result, + ) + ] + stories = [_story(state, "final_result", _battle_story(result), result)] + _append_overtime_stories(state, result, stories) + champion = state["players"][champion_key] + power = lineup_power(champion) + title = state["event_title"] + record = ( + f"小组赛 {champion['wins']}-{champion['losses']}" + if state.get("group_results") + else f"报名种子 #{champion.get('seed', 0)}" + ) + text = ( + f"{champion['display_name']} 夺得{title}冠军；最终阵容战力 " + f"{power['power']:.1f}，{record}。" + ) + stories.append( + _story( + state, + "champion", + text, + { + "champion_key": champion_key, + "power": power["power"], + "wins": champion["wins"], + "losses": champion["losses"], + }, + ) + ) + overtime_lines = [ + _host_text(str(item["overtime_story"])) + for item in result["maps"] + if int(item["overtime_count"]) > 0 + ] + return [ + _battle_story(result), + *overtime_lines, + _host_text(text), + ], stories, battles + + +def _settle_round( + state: dict[str, Any], + event_id: str, +) -> tuple[list[str], list[dict[str, Any]], list[dict[str, Any]]]: + ai_stories = _auto_manage_ai(state) + stage = _current_stage(state) + kind = str(stage["kind"]) + if kind == "pve": + lines, stories, battles = _settle_pve(state, event_id) + elif kind == "group": + lines, stories, battles = _settle_group(state, event_id) + elif kind == "semifinal": + lines, stories, battles = _settle_semifinal(state, event_id) + elif kind == "final": + lines, stories, battles = _settle_final(state, event_id) + else: + raise ValueError("赛事轮次无效") + stories = [*ai_stories, *stories] + round_intro = _host_text( + f"第 {state['round_number']} 回合{stage['name']}结算开始。" + ) + lines = [ + round_intro, + *(str(item["text"]) for item in ai_stories), + *lines, + ] + if state["status"] == "completed": + return lines, stories, battles + for key in state["join_order"]: + player = state["players"][key] + income = 5 + min(5, int(player["gold"]) // 10) + player["gold"] = int(player["gold"]) + income + player["ready"] = False + state["round_number"] = int(state["round_number"]) + 1 + next_stage = _current_stage(state) + state["phase"] = str(next_stage["phase"]) + _reset_preparation(state) + public_shop = _draw_shop( + state, + f"{state['run_id']}|round|{state['round_number']}|public", + ) + state["public_shop"] = public_shop + for key in state["join_order"]: + player = state["players"][key] + if not player["shop_locked"]: + player["shop"] = list(public_shop) + next_text = _host_text( + f"第 {state['round_number']} 回合{next_stage['name']}开始；" + "全员获得基础收入和利息。" + ) + stories.append(_story(state, "round_start", next_text, {"shop": public_shop})) + lines.append(next_text) + prompt = _preparation_prompt(state) + stories.append( + _story( + state, + "preparation_prompt", + prompt, + { + "person_key": _current_preparer(state), + "order": list(state["preparation_order"]), + }, + ) + ) + lines.append(prompt) + return lines, stories, battles + + +def _schedule_text(state: dict[str, Any]) -> str: + current = int(state.get("round_number", 0)) + if state["status"] == "registration": + joined = "、".join( + state["players"][key]["display_name"] for key in state["join_order"] + ) + return ( + f"{state['event_title']} 报名中；{len(state['join_order'])}/{MAX_PLAYERS} 人：" + f"{joined}。主持人发送 /自走棋开赛" + ) + ranking = _rankings(state) + stage = _current_stage(state) + stage_label = ( + "已结束" + if state["status"] == "completed" + else f"第 {current} 回合 {stage['name']}" + ) + lines = [ + f"{state['event_title']}｜{stage_label}" + ] + for index, key in enumerate(ranking, start=1): + player = state["players"][key] + marker = " *" if key in state.get("qualifiers", []) else "" + lines.append( + f"{index}. {player['display_name']} {player['wins']}-{player['losses']} " + f"HP{player['hp']} 战力{lineup_power(player)['power']:.1f} " + f"战术{player.get('tactic', '稳定')}{marker}" + ) + plan_text = " > ".join( + f"{index}{item['name']}" + for index, item in enumerate(state.get("round_plan", []), start=1) + ) + lines.append("赛程：" + plan_text) + return "\n".join(lines) + + +def _archive_text(state: dict[str, Any], page: int) -> str: + stories = list(state.get("stories", [])) + page_size = 7 + pages = max(1, (len(stories) + page_size - 1) // page_size) + page = min(max(1, page), pages) + start = (page - 1) * page_size + selected = stories[start : start + page_size] + lines = [ + f"{state['event_title']} 档案 {page}/{pages}；共 {len(stories)} 段剧情" + ] + for item in selected: + lines.append( + f"R{item['round_number']} {item['story_type']}：{item['text']}" + ) + if not selected: + lines.append("暂无剧情") + return "\n".join(lines) + + +def _parse_indices(value: str) -> list[int]: + value = value.strip() + if not value: + return [] + tokens = re.findall(r"\d{1,2}", value) + return [int(token) for token in tokens] + + +def parse_autochess_action(question: str) -> tuple[str, str]: + raw = question.lstrip() + if raw.startswith("/"): + raw = raw[1:].lstrip() + folded = raw.casefold() + for prefix in ("cs自走棋", "autochess", "自走棋"): + if folded.startswith(prefix.casefold()): + body = raw[len(prefix) :].strip() + compact = re.sub(r"\s+", "", body.casefold()) + if not compact: + return "status", "" + actions = ( + ("强制结算", "force_settle"), + ("结束回合", "ready"), + ("升级教练", "coach"), + ("加ai战队", "add_ai"), + ("当前状态", "status"), + ("开始", "create"), + ("报名", "join"), + ("开赛", "launch"), + ("商店", "status"), + ("状态", "status"), + ("刷新", "refresh"), + ("锁定", "lock"), + ("战术", "tactic"), + ("阵容", "lineup"), + ("赛程", "schedule"), + ("结算", "settle"), + ("档案", "archive"), + ("取消", "cancel"), + ("start", "create"), + ("join", "join"), + ("launch", "launch"), + ("addai", "add_ai"), + ("shop", "status"), + ("status", "status"), + ("refresh", "refresh"), + ("lock", "lock"), + ("tactic", "tactic"), + ("coach", "coach"), + ("lineup", "lineup"), + ("schedule", "schedule"), + ("ready", "ready"), + ("settle", "settle"), + ("archive", "archive"), + ("cancel", "cancel"), + ) + for token, action in actions: + if compact.startswith(token): + return action, compact[len(token) :] + if compact.startswith("选") or compact.startswith("buy"): + token = "选" if compact.startswith("选") else "buy" + return "buy", compact[len(token) :] + for token, action in (("上阵", "field"), ("下阵", "bench"), ("出售", "sell")): + if compact.startswith(token): + return action, body[len(token) :].strip() + return "unknown", body + direct = ( + ("升级教练", "coach"), + ("结束回合", "ready"), + ("刷新", "refresh"), + ("锁定", "lock"), + ("战术", "tactic"), + ("阵容", "lineup"), + ("赛程", "schedule"), + ("上阵", "field"), + ("下阵", "bench"), + ("出售", "sell"), + ("选", "buy"), + ) + for token, action in direct: + if folded.startswith(token.casefold()): + return action, raw[len(token) :].strip() + return "unknown", raw + + +class CsMajorAutochess: + """Command facade; persistence remains behind the Kernel Ledger service.""" + + def __init__(self, ledger: Any) -> None: + self.ledger = ledger + + @staticmethod + def rules() -> str: + return ( + "CS自走棋：/自走棋开始 创建报名；/自走棋报名；主持人 /自走棋开赛。\n" + "三名真人报名后，主持人可用 /自走棋加AI战队 开启 3 人+1 AI 测试模式。\n" + "每人固定 5 名首发 + 1 名教练，开赛即发完整首发；每轮公开 6 人商店。" + "/选1～/选6 购买；/阵容；/上阵 备战编号 [首发位]；/出售 备战编号。\n" + "教练只提供全队 +2%～+10% 加成；/升级教练；/刷新；/锁定。" + "/战术 火力|稳定|残局，稳定克火力、火力克残局、残局克稳定。\n" + "小组赛 BO3，半决赛/决赛 BO5；单图 MR15，15-15 进入加时。" + "玩家 /结束回合，主持人 /自走棋结算；/赛程；/自走棋档案 [页码]。" + ) + + def handle(self, row: dict[str, Any]) -> str: + action_kind, arguments = parse_autochess_action(str(row["question"])) + group = str(row["group_fingerprint"]) + sender = str(row["sender"]) + event_id = str(row["event_id"]) + command = f"autochess:{action_kind}:{arguments}" + + def execute( + state: dict[str, Any] | None, + person: dict[str, str], + next_season: int, + ) -> dict[str, Any]: + result = self._execute( + state, + person, + next_season, + action_kind, + arguments, + event_id, + ) + new_state = result.get("state") + response = result.get("response") + if isinstance(new_state, dict) and isinstance(response, str): + prefix = ( + f"{person['display_name']} " + f"#{new_state['season_number']}：" + ) + if not response.startswith(prefix): + result["response"] = prefix + response + return result + + return self.ledger.apply_autochess_action( + group, + sender, + event_id, + command, + execute, + ) + + def _execute( + self, + state: dict[str, Any] | None, + person: dict[str, str], + next_season: int, + action: str, + arguments: str, + event_id: str, + ) -> dict[str, Any]: + person_key = str(person["person_key"]) + display_name = str(person["display_name"]) + if action == "unknown": + return _result(state, self.rules()) + if action == "create": + if state is not None and state.get("status") in {"registration", "active"}: + return _result( + state, + f"{state['event_title']} 已在进行中；发送 /自走棋状态 查看", + ) + season = next_season + state = { + "schema_version": 1, + "run_id": f"major_{season:04d}", + "season_number": season, + "event_title": f"第 {season} 届 IEM 盐田 Major", + "status": "registration", + "host_person_key": person_key, + "host_display_name": display_name, + "round_number": 0, + "phase": "registration", + "players": { + person_key: _participant(person_key, display_name) + }, + "join_order": [person_key], + "pool": { + card["id"]: COST_POOL_COPIES[int(card["cost"])] + for card in PLAYER_POOL + }, + "public_shop": [], + "seed_order": [], + "round_plan": [], + "group_results": [], + "ai_test_mode": False, + "preparation_order": [], + "preparation_cursor": 0, + "qualifiers": [], + "finalists": [], + "champion_key": "", + "stories": [], + "battles": [], + "story_sequence": 0, + "battle_sequence": 0, + } + text = _host_text( + f"{state['event_title']} 开放报名；主持人 {display_name} 已自动加入。" + "支持 2～8 人；发送 /自走棋报名，主持人发送 /自走棋开赛" + ) + story = _story( + state, + "registration_open", + text, + {"host_person_key": person_key}, + ) + return _result(state, f"{display_name} #{season}：{text}", stories=[story]) + if state is None: + return _result(None, "当前没有自走棋赛事；发送 /自走棋开始") + if action == "join": + if state["status"] != "registration": + return _result(state, "本届赛事已停止报名") + if state.get("ai_test_mode"): + return _result( + state, + _host_text( + "3 人+1 AI 测试模式已经锁定参赛名单，不能再加入真人。" + ), + ) + if person_key in state["players"]: + return _result( + state, + f"{_player_prefix(state, state['players'][person_key])}你已报名", + ) + if len(state["join_order"]) >= MAX_PLAYERS: + return _result(state, f"本届赛事已满 {MAX_PLAYERS} 人") + state["players"][person_key] = _participant(person_key, display_name) + state["join_order"].append(person_key) + text = _host_text( + f"{display_name} 加入赛事，当前 {len(state['join_order'])}/{MAX_PLAYERS} 人。" + ) + story = _story( + state, + "player_joined", + text, + {"person_key": person_key}, + ) + return _result(state, f"{display_name} #{state['season_number']}：{text}", stories=[story]) + if action == "add_ai": + if state["status"] != "registration": + return _result(state, "AI 战队只能在报名阶段加入") + if person_key != state["host_person_key"]: + return _result( + state, + f"只有主持人 {state['host_display_name']} 可以加入 AI 战队", + ) + if AI_PERSON_KEY in state["players"]: + return _result(state, "3 人+1 AI 测试战队已经加入") + human_keys = [ + key + for key in state["join_order"] + if not state["players"][key].get("is_ai") + ] + if len(human_keys) != 3: + return _result( + state, + "3 人+1 AI 测试模式要求先有且仅有 3 名真人报名", + ) + state["players"][AI_PERSON_KEY] = _participant( + AI_PERSON_KEY, + AI_DISPLAY_NAME, + is_ai=True, + ) + state["join_order"].append(AI_PERSON_KEY) + state["ai_test_mode"] = True + text = _host_text( + "AI 战队已作为第 4 种子加入测试赛；" + "开赛后固定进行 1v4、2v3 半决赛，AI 将自动运营。" + ) + story = _story( + state, + "ai_team_joined", + text, + { + "person_key": AI_PERSON_KEY, + "seed": 4, + "test_mode": True, + }, + ) + return _result(state, text, stories=[story]) + if action == "launch": + if state["status"] != "registration": + return _result(state, "本届赛事已经开赛或结束") + if person_key != state["host_person_key"]: + return _result( + state, + f"只有主持人 {state['host_display_name']} 可以开赛", + ) + if len(state["join_order"]) < MIN_PLAYERS: + return _result(state, f"至少需要 {MIN_PLAYERS} 人才能开赛") + state["status"] = "active" + state["round_number"] = 1 + state["seed_order"] = list(state["join_order"]) + state["round_plan"] = build_round_plan(list(state["join_order"])) + state["phase"] = str(_current_stage(state)["phase"]) + for key in state["join_order"]: + _assign_starters(state, state["players"][key]) + for seed, key in enumerate(state["seed_order"], start=1): + state["players"][key]["seed"] = seed + _reset_preparation(state) + public_shop = _draw_shop(state, f"{state['run_id']}|round|1|public") + state["public_shop"] = public_shop + for key in state["join_order"]: + state["players"][key]["shop"] = list(public_shop) + names = "、".join( + state["players"][key]["display_name"] for key in state["join_order"] + ) + seeds = "、".join( + f"{state['players'][key]['seed']}号" + f"{state['players'][key]['display_name']}" + for key in state["seed_order"] + ) + text = _host_text( + f"欢迎大家来到{state['event_title']}！参赛者：{names}。" + f"种子顺序：{seeds}。{_format_description(state)}；" + "小组赛 BO3，淘汰赛 BO5，单图采用 MR15，" + "常规时间 15-15 进入加时。" + "每队已获得完整 5 人首发与常驻教练；" + f"共 {len(state['round_plan'])} 回合，第 1 回合训练赛开始。" + ) + story = _story( + state, + "tournament_started", + text, + { + "players": list(state["join_order"]), + "public_shop": public_shop, + "round_plan": list(state["round_plan"]), + }, + ) + preparation_story = _story( + state, + "preparation_prompt", + _preparation_prompt(state), + { + "person_key": _current_preparer(state), + "order": list(state["preparation_order"]), + }, + ) + response = ( + text + + "\n" + + _shop_text(state, state["players"][person_key]) + + "\n" + + _preparation_prompt(state) + ) + return _result( + state, + response, + stories=[story, preparation_story], + ) + player = state["players"].get(person_key) + if action == "archive": + page_values = _parse_indices(arguments) + return _result(state, _archive_text(state, page_values[0] if page_values else 1)) + if action == "schedule": + return _result(state, _schedule_text(state)) + if state["status"] == "registration": + return _result(state, _schedule_text(state)) + if state["status"] in {"completed", "cancelled"}: + if action in {"status", "lineup"}: + return _result(state, _schedule_text(state)) + return _result( + state, + f"{state['event_title']} 已{('结束' if state['status'] == 'completed' else '取消')}；" + "发送 /自走棋档案 查看记录，或 /自走棋开始 创建下一届", + ) + if player is None: + return _result(state, "你没有参加本届赛事，只能查看 /自走棋赛程") + if action == "status": + return _result( + state, + _shop_text(state, player) + "\n" + _preparation_prompt(state), + ) + if action == "lineup": + return _result(state, _lineup_text(state, player)) + preparation_actions = { + "buy", + "refresh", + "lock", + "tactic", + "coach", + "field", + "bench", + "sell", + "ready", + } + if action in preparation_actions: + current_preparer = _current_preparer(state) + if current_preparer is None: + return _result( + state, + _host_text( + "本回合全员已经准备完成，请等待主持人结算。" + ), + ) + if person_key != current_preparer: + current_name = state["players"][current_preparer]["display_name"] + return _result( + state, + _host_text( + f"当前轮到 {current_name} 做赛前准备；" + f"{display_name} 请等待交棒。" + ), + ) + if action == "tactic": + value = re.sub(r"\s+", "", arguments.casefold()) + aliases = { + "火力": "火力", + "fire": "火力", + "稳定": "稳定", + "stable": "稳定", + "残局": "残局", + "clutch": "残局", + } + if not value: + return _result( + state, + f"当前战术 {player['tactic']}；用法：/战术 火力|稳定|残局", + ) + tactic = aliases.get(value) + if tactic is None: + return _result(state, "战术只能选择：火力、稳定、残局") + player["tactic"] = tactic + text = ( + f"{display_name} 将本回合战术设为 {tactic}；" + "克制环为稳定>火力>残局>稳定。" + ) + story = _story( + state, + "tactic_selected", + text, + {"person_key": person_key, "tactic": tactic}, + ) + return _result(state, text, stories=[story]) + if action == "buy": + values = _parse_indices(arguments) + if len(values) != 1 or not 1 <= values[0] <= SHOP_SIZE: + return _result(state, "用法：/选1 到 /选6") + slot = values[0] - 1 + card_id = player["shop"][slot] if slot < len(player["shop"]) else None + if card_id is None: + return _result(state, "这个商店位置已经购买或没有候选") + card = CARD_BY_ID[str(card_id)] + if int(player["gold"]) < int(card["cost"]): + return _result( + state, + f"金币不足：{card['name']} 需要 {card['cost']}，当前 {player['gold']}", + ) + existing = any( + unit["card_id"] == card_id + for unit in [*player["board"], *player["bench"]] + ) + if not existing and len(player["bench"]) >= BENCH_LIMIT: + return _result(state, f"备战席已满 {BENCH_LIMIT} 格，请先出售") + if int(state["pool"].get(card_id, 0)) <= 0: + return _result(state, "共享卡池中的这名选手已被拿完，请刷新") + player["gold"] -= int(card["cost"]) + state["pool"][card_id] -= 1 + star, upgraded = _merge_purchase(player, str(card_id)) + player["shop"][slot] = None + text = ( + f"{display_name} 选中 {card['name']}，花费 {card['cost']} 金币；" + f"当前 {star} 星，余额 {player['gold']}。" + ) + if upgraded: + text += f"{card['name']} 完成升星。" + story = _story( + state, + "shop_purchase", + text, + { + "person_key": person_key, + "card_id": card_id, + "star": star, + "upgraded": upgraded, + "gold": player["gold"], + }, + ) + return _result(state, f"{_player_prefix(state, player)}{text}", stories=[story]) + if action == "refresh": + if int(player["gold"]) < REFRESH_COST: + return _result(state, f"刷新需要 {REFRESH_COST} 金币，当前 {player['gold']}") + player["gold"] -= REFRESH_COST + player["shop"] = _draw_shop( + state, + f"{state['run_id']}|{state['round_number']}|{event_id}|{person_key}|refresh", + ) + text = f"{display_name} 支付 {REFRESH_COST} 金币刷新商店。" + story = _story( + state, + "shop_refresh", + text, + {"person_key": person_key, "shop": list(player["shop"])}, + ) + return _result(state, _shop_text(state, player), stories=[story]) + if action == "lock": + player["shop_locked"] = not bool(player["shop_locked"]) + text = ( + f"{display_name} {'锁定' if player['shop_locked'] else '解锁'}了商店；" + + ("下一回合保留当前 6 人。" if player["shop_locked"] else "下一回合将自动换新。") + ) + story = _story( + state, + "shop_lock", + text, + {"person_key": person_key, "locked": player["shop_locked"]}, + ) + return _result(state, f"{_player_prefix(state, player)}{text}", stories=[story]) + if action == "coach": + level = int(player["coach_level"]) + if level >= 5: + return _result(state, "教练已满 5 级，全队战力加成 +10%") + cost = COACH_UPGRADE_COSTS[level] + if int(player["gold"]) < cost: + return _result(state, f"教练升级需要 {cost} 金币，当前 {player['gold']}") + player["gold"] -= cost + player["coach_level"] = level + 1 + bonus = COACH_BONUSES[level + 1] + text = ( + f"{display_name} 将教练升到 {level + 1} 级；" + f"全队加成 +{bonus}%，余额 {player['gold']}。" + ) + story = _story( + state, + "coach_upgrade", + text, + { + "person_key": person_key, + "coach_level": level + 1, + "coach_bonus": bonus, + }, + ) + return _result(state, f"{_player_prefix(state, player)}{text}", stories=[story]) + if action == "field": + values = _parse_indices(arguments) + if not values: + return _result(state, "用法：/上阵 备战编号 [首发位]，例如 /上阵 6 2") + found = _find_unit(player, values[0]) + if found is None or found[0] != "备战": + return _result(state, "第一个编号必须是 /阵容 中的备战席编号") + bench_index = found[1] + board_index = ( + values[1] - 1 + if len(values) >= 2 and 1 <= values[1] <= STARTER_COUNT + else min( + range(STARTER_COUNT), + key=lambda index: _unit_power(player["board"][index]), + ) + ) + incoming = player["bench"][bench_index] + outgoing = player["board"][board_index] + player["board"][board_index] = incoming + player["bench"][bench_index] = outgoing + in_name = CARD_BY_ID[str(incoming["card_id"])]["name"] + out_name = CARD_BY_ID[str(outgoing["card_id"])]["name"] + text = ( + f"{display_name} 用 {in_name} 替换首发 {board_index + 1} 的 {out_name}；" + "首发仍为完整 5 人。" + ) + story = _story( + state, + "lineup_change", + text, + { + "person_key": person_key, + "incoming": incoming["card_id"], + "outgoing": outgoing["card_id"], + "board_slot": board_index + 1, + }, + ) + return _result(state, f"{_player_prefix(state, player)}{text}", stories=[story]) + if action == "bench": + values = _parse_indices(arguments) + if len(values) < 2: + return _result( + state, + "固定五人阵容不能留空；用法：/下阵 首发位 备战编号，例如 /下阵 2 6", + ) + board_index = values[0] - 1 + found = _find_unit(player, values[1]) + if not 0 <= board_index < STARTER_COUNT or found is None or found[0] != "备战": + return _result(state, "首发位必须为 1～5，替补必须是 /阵容 中的备战编号") + bench_index = found[1] + incoming = player["bench"][bench_index] + outgoing = player["board"][board_index] + player["board"][board_index] = incoming + player["bench"][bench_index] = outgoing + text = ( + f"{display_name} 调整首发：{CARD_BY_ID[str(incoming['card_id'])]['name']} " + f"替换 {CARD_BY_ID[str(outgoing['card_id'])]['name']}。" + ) + story = _story(state, "lineup_change", text, {"person_key": person_key}) + return _result(state, f"{_player_prefix(state, player)}{text}", stories=[story]) + if action == "sell": + values = _parse_indices(arguments) + if len(values) != 1: + return _result(state, "用法：/出售 备战编号") + found = _find_unit(player, values[0]) + if found is None: + return _result(state, "没有这个阵容编号") + if found[0] == "首发": + return _result(state, "首发必须保持 5 人；先用 /上阵 备战编号 首发位 完成替换") + unit = player["bench"].pop(found[1]) + card = CARD_BY_ID[str(unit["card_id"])] + refund = int(card["cost"]) * int(unit["copies"]) + player["gold"] += refund + state["pool"][card["id"]] += int(unit["copies"]) + text = ( + f"{display_name} 出售 {card['name']} {unit['star']} 星×{unit['copies']}，" + f"返还 {refund} 金币；余额 {player['gold']}。" + ) + story = _story( + state, + "unit_sold", + text, + { + "person_key": person_key, + "card_id": card["id"], + "copies": unit["copies"], + "refund": refund, + }, + ) + return _result(state, f"{_player_prefix(state, player)}{text}", stories=[story]) + if action == "ready": + player["ready"] = True + state["preparation_cursor"] = ( + int(state["preparation_cursor"]) + 1 + ) + waiting = [ + state["players"][key]["display_name"] + for key in state["preparation_order"][ + int(state["preparation_cursor"]) : + ] + ] + next_preparer = _current_preparer(state) + if next_preparer is None: + text = _host_text( + f"{display_name} 准备完成！全员准备完毕，" + "主持人可以发送 /自走棋结算。" + ) + else: + next_name = state["players"][next_preparer]["display_name"] + text = _host_text( + f"{display_name} 准备完成！到 {next_name} 做准备了。" + f"{next_name}，请做好赛前准备；完成后发送 /结束回合。" + ) + story = _story( + state, + "round_ready", + text, + {"person_key": person_key, "waiting": waiting}, + ) + return _result(state, text, stories=[story]) + if action in {"settle", "force_settle"}: + if person_key != state["host_person_key"]: + return _result( + state, + f"只有主持人 {state['host_display_name']} 可以结算", + ) + waiting = [ + state["players"][key]["display_name"] + for key in state["join_order"] + if ( + not state["players"][key]["ready"] + and not state["players"][key].get("is_ai") + ) + ] + if waiting and action != "force_settle": + return _result( + state, + _host_text( + f"仍在等待：{'、'.join(waiting)}；" + "请按准备顺序完成，主持人也可用 /自走棋强制结算。" + ), + ) + lines, stories, battles = _settle_round(state, event_id) + return _result(state, "\n".join(lines), stories=stories, battles=battles) + if action == "cancel": + if person_key != state["host_person_key"]: + return _result( + state, + f"只有主持人 {state['host_display_name']} 可以取消赛事", + ) + state["status"] = "cancelled" + text = f"{state['event_title']} 已由主持人取消；已完成剧情仍保留在档案。" + story = _story(state, "tournament_cancelled", text) + return _result(state, text, stories=[story]) + return _result(state, self.rules()) diff --git a/packages/jwebot-official-extensions/src/jwebot_official_extensions/descriptors.py b/packages/jwebot-official-extensions/src/jwebot_official_extensions/descriptors.py index bbaaa04..3c2b984 100644 --- a/packages/jwebot-official-extensions/src/jwebot_official_extensions/descriptors.py +++ b/packages/jwebot-official-extensions/src/jwebot_official_extensions/descriptors.py @@ -61,6 +61,20 @@ LOCAL_COMMAND_ALIASES = { "人物": "person_info", "个人属性": "person_info", "person": "person_info", + "自走棋": "cs_autochess", + "cs自走棋": "cs_autochess", + "autochess": "cs_autochess", + "选": "cs_autochess", + "刷新": "cs_autochess", + "锁定": "cs_autochess", + "战术": "cs_autochess", + "升级教练": "cs_autochess", + "上阵": "cs_autochess", + "下阵": "cs_autochess", + "出售": "cs_autochess", + "阵容": "cs_autochess", + "赛程": "cs_autochess", + "结束回合": "cs_autochess", } @@ -141,6 +155,13 @@ LOCAL_EXTENSIONS = ( "game_rules", ), ), + LocalExtensionDescriptor( + extension_id="cs-major-autochess", + author=JWEBOT_AUTHOR, + description="CS自走棋", + help_entries=("/自走棋开始|报名",), + command_kinds=("cs_autochess",), + ), LocalExtensionDescriptor( extension_id="person-directory", author=OFFICIAL_AUTHOR, diff --git a/packages/jwebot-official-extensions/src/jwebot_official_extensions/local_commands.py b/packages/jwebot-official-extensions/src/jwebot_official_extensions/local_commands.py index cd6bf1c..d4c992e 100644 --- a/packages/jwebot-official-extensions/src/jwebot_official_extensions/local_commands.py +++ b/packages/jwebot-official-extensions/src/jwebot_official_extensions/local_commands.py @@ -14,6 +14,7 @@ from datetime import datetime, timezone from typing import Any, Callable from zoneinfo import ZoneInfo +from .cs_autochess import CsMajorAutochess from .descriptors import LOCAL_COMMAND_ALIASES, render_feature_list from .player_games import PlayerGames @@ -107,6 +108,7 @@ class LocalCommandEngine: self.profile_fact_is_safe = profile_fact_is_safe self.profile_expiry = profile_expiry self.players = PlayerGames(ledger, person_display_label) + self.autochess = CsMajorAutochess(ledger) def handle(self, row: dict[str, Any], event: dict[str, Any]) -> str: question = str(row["question"]) @@ -128,6 +130,8 @@ class LocalCommandEngine: "game_rules", }: return self.reply_cleaner(self.players.handle(row, kind, arguments)) + if kind == "cs_autochess": + return self.reply_cleaner(self.autochess.handle(row)) if kind == "dice": return self.reply_cleaner(self.dice(row, arguments)) if kind == "dice_challenge": diff --git a/packages/jwebot-sdk/pyproject.toml b/packages/jwebot-sdk/pyproject.toml index c661b86..0dc15b2 100644 --- a/packages/jwebot-sdk/pyproject.toml +++ b/packages/jwebot-sdk/pyproject.toml @@ -4,7 +4,7 @@ build-backend = "setuptools.build_meta" [project] name = "jwebot-sdk" -version = "0.6.3a0+1015" +version = "0.6.4a0+1015" description = "Typed, capability-limited extension API for JWeBot plugins" requires-python = ">=3.9" dependencies = [] diff --git a/packages/jwebot-sdk/src/jwebot_sdk/__init__.py b/packages/jwebot-sdk/src/jwebot_sdk/__init__.py index a79ddaa..1880bd8 100644 --- a/packages/jwebot-sdk/src/jwebot_sdk/__init__.py +++ b/packages/jwebot-sdk/src/jwebot_sdk/__init__.py @@ -38,4 +38,4 @@ __all__ = [ "SingleGroupContext", ] -__version__ = "0.6.3a0+1015" +__version__ = "0.6.4a0+1015" diff --git a/pyproject.toml b/pyproject.toml index 33809ac..1ed1c23 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -1,5 +1,5 @@ [tool.jwebot] -release = "alpha 0.6.3.1015" +release = "alpha 0.6.4.1015" production_deployments_enabled_by_default = false [tool.jwebot.workspace] diff --git a/tests/test_official_extension_inventory.py b/tests/test_official_extension_inventory.py index e81f353..5b3c9cf 100644 --- a/tests/test_official_extension_inventory.py +++ b/tests/test_official_extension_inventory.py @@ -83,6 +83,7 @@ class LocalExtensionInventoryTests(unittest.TestCase): "context-summary": OFFICIAL_AUTHOR, "person-directory": OFFICIAL_AUTHOR, "single-player-games": JWEBOT_AUTHOR, + "cs-major-autochess": JWEBOT_AUTHOR, "person-memory": OFFICIAL_AUTHOR, "server-status": OFFICIAL_AUTHOR, "jwebot-version": OFFICIAL_AUTHOR, diff --git a/tests/test_release_064.py b/tests/test_release_064.py new file mode 100644 index 0000000..9687c93 --- /dev/null +++ b/tests/test_release_064.py @@ -0,0 +1,508 @@ +"""Behavioral coverage for the CS Major autochess extension.""" + +from __future__ import annotations + +import copy +import json +import re +import tempfile +import threading +import unittest +from pathlib import Path +from unittest import mock + +from test_jwebot_edge import MODULE, config +from jwebot_official_extensions import cs_autochess as AUTOCHESS + + +class Release064AutochessTests(unittest.TestCase): + def setUp(self): + self.temp = tempfile.TemporaryDirectory() + self.root = Path(self.temp.name) + self.engine = MODULE.EdgeApplication(config(self.root)) + self.counter = 0 + + def tearDown(self): + self.engine.close() + self.temp.cleanup() + + def command( + self, + question: str, + *, + sender: str = "Alice", + event_id: str | None = None, + ) -> str: + self.counter += 1 + row = { + "question": question, + "sender": sender, + "group_fingerprint": "group-a", + "event_id": event_id or f"autochess-{self.counter}", + } + return self.engine.local.handle(row, {}) + + def state(self) -> dict: + row = self.engine.ledger.connection.execute( + """ + SELECT state_json FROM autochess_runs + WHERE group_fingerprint = 'group-a' + ORDER BY season_number DESC + LIMIT 1 + """ + ).fetchone() + self.assertIsNotNone(row) + return json.loads(row[0]) + + def save_state(self, state: dict) -> None: + self.engine.ledger.connection.execute( + """ + UPDATE autochess_runs SET state_json = ?, state = ? + WHERE group_fingerprint = 'group-a' AND run_id = ? + """, + ( + json.dumps(state, ensure_ascii=False, separators=(",", ":")), + state["status"], + state["run_id"], + ), + ) + + def launch(self, names: list[str]) -> dict: + self.command("/自走棋开始", sender=names[0]) + for name in names[1:]: + self.command("/自走棋报名", sender=name) + self.command("/自走棋开赛", sender=names[0]) + return self.state() + + def settle_to_completion(self) -> dict: + state = self.state() + while state["status"] == "active": + self.command("/自走棋强制结算") + state = self.state() + return state + + def test_static_pool_parser_and_attribute_ranges(self): + self.assertEqual(len(AUTOCHESS.PLAYER_POOL), 60) + self.assertEqual( + len({card["id"] for card in AUTOCHESS.PLAYER_POOL}), + 60, + ) + ranges = { + 1: (52, 68), + 2: (60, 76), + 3: (68, 84), + 4: (78, 92), + 5: (88, 98), + } + for card in AUTOCHESS.PLAYER_POOL: + with self.subTest(card=card["id"]): + lower, upper = ranges[card["cost"]] + for field in ("firepower", "stability", "clutch"): + self.assertGreaterEqual(card[field], lower) + self.assertLessEqual(card[field], upper) + self.assertIn(card["role"], {"步枪手", "狙击手", "指挥"}) + self.assertLessEqual(len(card["teams"]), 3) + if card["role"] == "指挥": + self.assertGreater(card["leadership"], 0) + self.assertEqual( + AUTOCHESS.parse_autochess_action("/自走棋开始"), + ("create", ""), + ) + self.assertEqual( + AUTOCHESS.parse_autochess_action("/选6"), + ("buy", "6"), + ) + self.assertEqual( + AUTOCHESS.parse_autochess_action("/战术残局"), + ("tactic", "残局"), + ) + + def test_round_plan_matches_player_count_without_duplicate_pairs(self): + for size in range(2, 9): + keys = [f"p{index}" for index in range(1, size + 1)] + plan = AUTOCHESS.build_round_plan(keys) + group_pairs = [ + tuple(sorted(pair)) + for stage in plan + if stage["kind"] == "group" + for pair in stage["pairs"] + ] + with self.subTest(size=size): + if size == 2: + self.assertFalse(group_pairs) + self.assertFalse( + any(stage["kind"] == "semifinal" for stage in plan) + ) + elif size == 4: + semifinal = next( + stage + for stage in plan + if stage["kind"] == "semifinal" + ) + self.assertEqual( + semifinal["pairs"], + [["p1", "p4"], ["p2", "p3"]], + ) + self.assertFalse(group_pairs) + else: + self.assertEqual( + len(group_pairs), + size * (size - 1) // 2, + ) + self.assertEqual(len(group_pairs), len(set(group_pairs))) + self.assertEqual(plan[-1]["kind"], "final") + + def test_launch_has_fixed_five_coach_shared_shop_and_complete_data(self): + response = self.command("/自走棋开始") + self.assertTrue(response.startswith("alice #1：")) + self.command("/自走棋报名", sender="Bob") + self.command("/自走棋报名", sender="Carol") + self.command("/自走棋报名", sender="Dave") + response = self.command("/自走棋开赛") + state = self.state() + + self.assertEqual(len(state["round_plan"]), 5) + self.assertEqual( + [stage["kind"] for stage in state["round_plan"]], + ["pve", "pve", "pve", "semifinal", "final"], + ) + shops = [] + for player in state["players"].values(): + self.assertEqual(len(player["board"]), 5) + self.assertEqual(player["coach_level"], 1) + self.assertEqual(player["tactic"], "稳定") + self.assertEqual(len(player["shop"]), 6) + shops.append(player["shop"]) + self.assertTrue(all(shop == shops[0] for shop in shops)) + for slot in range(1, 7): + line = next( + value + for value in response.splitlines() + if value.startswith(f"{slot}. ") + ) + self.assertRegex( + line, + r"\d费 (?:步枪手|狙击手|指挥).*历史 .*火\d+ 稳\d+ 残\d+", + ) + self.assertLessEqual(len(response), MODULE.MAX_REPLY_CHARS) + + def test_tactics_coach_and_lineup_changes_preserve_five_starters(self): + state = self.launch(["Alice", "Bob"]) + alice_key = state["seed_order"][0] + before_board = copy.deepcopy(state["players"][alice_key]["board"]) + before_power = AUTOCHESS.lineup_power( + state["players"][alice_key] + )["power"] + + tactic = self.command("/战术残局") + self.assertIn("战术设为 残局", tactic) + coach = self.command("/升级教练") + self.assertIn("教练升到 2 级", coach) + state = self.state() + alice = state["players"][alice_key] + self.assertEqual(alice["board"], before_board) + self.assertEqual(len(alice["board"]), 5) + self.assertGreater(AUTOCHESS.lineup_power(alice)["power"], before_power) + + starter_ids = {unit["card_id"] for unit in alice["board"]} + slot = next( + index + for index, card_id in enumerate(alice["shop"], start=1) + if card_id not in starter_ids + ) + self.command(f"/选{slot}") + self.assertIn("首发必须保持 5 人", self.command("/出售1")) + swapped = self.command("/上阵 6 2") + self.assertIn("首发仍为完整 5 人", swapped) + state = self.state() + self.assertEqual(len(state["players"][alice_key]["board"]), 5) + + def test_replay_debits_one_purchase_and_invalid_archive_rolls_back(self): + state = self.launch(["Alice", "Bob"]) + alice_key = state["seed_order"][0] + card_id = state["players"][alice_key]["shop"][0] + cost = AUTOCHESS.CARD_BY_ID[card_id]["cost"] + old_gold = state["players"][alice_key]["gold"] + old_pool = state["pool"][card_id] + replies: list[str] = [] + errors: list[BaseException] = [] + + def buy() -> None: + try: + replies.append( + self.command("/选1", event_id="same-autochess-buy") + ) + except BaseException as error: + errors.append(error) + + threads = [threading.Thread(target=buy) for _ in range(8)] + for thread in threads: + thread.start() + for thread in threads: + thread.join() + self.assertFalse(errors) + self.assertEqual(len(set(replies)), 1) + state = self.state() + self.assertEqual(state["players"][alice_key]["gold"], old_gold - cost) + self.assertEqual(state["pool"][card_id], old_pool - 1) + + before_json = self.engine.ledger.connection.execute( + """ + SELECT state_json FROM autochess_runs + WHERE group_fingerprint = 'group-a' AND run_id = ? + """, + (state["run_id"],), + ).fetchone()[0] + before_stories = self.engine.ledger.connection.execute( + "SELECT COUNT(*) FROM autochess_story_logs" + ).fetchone()[0] + + def invalid_archive(current, person, next_season): + del person, next_season + current["players"][alice_key]["gold"] += 999 + return { + "state": current, + "response": "invalid", + "stories": [{"bad": True}], + "battles": [], + } + + with self.assertRaisesRegex( + MODULE.LocalCommandError, + "剧情归档无效", + ): + self.engine.ledger.apply_autochess_action( + "group-a", + "Alice", + "invalid-autochess-archive", + "invalid", + invalid_archive, + ) + after_json = self.engine.ledger.connection.execute( + """ + SELECT state_json FROM autochess_runs + WHERE group_fingerprint = 'group-a' AND run_id = ? + """, + (state["run_id"],), + ).fetchone()[0] + after_stories = self.engine.ledger.connection.execute( + "SELECT COUNT(*) FROM autochess_story_logs" + ).fetchone()[0] + self.assertEqual(after_json, before_json) + self.assertEqual(after_stories, before_stories) + + def test_four_team_seed_bracket_is_bo5_with_mr15_scores(self): + state = self.launch(["Alice", "Bob", "Carol", "Dave"]) + seed_order = list(state["seed_order"]) + completed = self.settle_to_completion() + self.assertEqual(completed["status"], "completed") + self.assertTrue(completed["champion_key"]) + rows = self.engine.ledger.connection.execute( + """ + SELECT battle_type, player_a, player_b, result_json + FROM autochess_battles + WHERE battle_type != 'pve' + ORDER BY battle_id + """ + ).fetchall() + semifinals = [row for row in rows if row["battle_type"] == "semifinal_bo5"] + finals = [row for row in rows if row["battle_type"] == "final_bo5"] + self.assertEqual( + { + frozenset((row["player_a"], row["player_b"])) + for row in semifinals + }, + { + frozenset((seed_order[0], seed_order[3])), + frozenset((seed_order[1], seed_order[2])), + }, + ) + self.assertEqual(len(finals), 1) + for row in [*semifinals, *finals]: + result = json.loads(row["result_json"]) + self.assertEqual(result["series_format"], "BO5") + self.assertRegex(result["score"], r"^3-[0-2]$") + self.assertTrue(3 <= len(result["maps"]) <= 5) + for item in result["maps"]: + self.assertRegex(item["score"], r"^(?:16-(?:6|9|12|14)|19-17|22-20)$") + self.assertLessEqual(abs(item["fluctuation_a_bps"]), 200) + self.assertLessEqual(abs(item["fluctuation_b_bps"]), 200) + + def test_five_team_round_robin_bo3_then_top_four(self): + completed = self.launch(["Alice", "Bob", "Carol", "Dave", "Eve"]) + self.assertEqual( + sum( + len(stage.get("pairs", [])) + for stage in completed["round_plan"] + if stage["kind"] == "group" + ), + 10, + ) + completed = self.settle_to_completion() + self.assertEqual(len(completed["qualifiers"]), 4) + rows = self.engine.ledger.connection.execute( + """ + SELECT battle_type, player_a, player_b, result_json + FROM autochess_battles + WHERE battle_type != 'pve' + ORDER BY battle_id + """ + ).fetchall() + groups = [row for row in rows if row["battle_type"] == "group_bo3"] + self.assertEqual(len(groups), 10) + self.assertEqual( + len( + { + frozenset((row["player_a"], row["player_b"])) + for row in groups + } + ), + 10, + ) + for row in groups: + result = json.loads(row["result_json"]) + self.assertEqual(result["series_format"], "BO3") + self.assertRegex(result["score"], r"^2-[01]$") + self.assertEqual( + sum(row["battle_type"] == "semifinal_bo5" for row in rows), + 2, + ) + self.assertEqual( + sum(row["battle_type"] == "final_bo5" for row in rows), + 1, + ) + + def test_three_humans_plus_ai_has_taffy_host_and_turn_handoffs(self): + self.command("/自走棋开始") + self.command("/自走棋报名", sender="Bob") + self.assertIn( + "先有且仅有 3 名真人", + self.command("/自走棋加AI战队"), + ) + self.command("/自走棋报名", sender="Carol") + self.assertIn( + "只有主持人", + self.command("/自走棋加AI战队", sender="Bob"), + ) + self.assertIn("第 4 种子", self.command("/自走棋加AI战队")) + self.assertIn( + "已经锁定参赛名单", + self.command("/自走棋报名", sender="Dave"), + ) + launch = self.command("/自走棋开赛") + state = self.state() + + self.assertIn("Taffy：欢迎大家来到第 1 届 IEM 盐田 Major", launch) + self.assertIn("3 人+1 AI 测试赛", launch) + self.assertIn("1v4、2v3", launch) + self.assertIn("小组赛 BO3，淘汰赛 BO5", launch) + self.assertIn("常规时间 15-15 进入加时", launch) + self.assertIn("alice，请做好赛前准备", launch) + self.assertTrue(state["ai_test_mode"]) + self.assertEqual(state["seed_order"][-1], AUTOCHESS.AI_PERSON_KEY) + self.assertTrue( + state["players"][AUTOCHESS.AI_PERSON_KEY]["is_ai"] + ) + semifinal = state["round_plan"][-2] + self.assertEqual( + semifinal["pairs"], + [ + [state["seed_order"][0], AUTOCHESS.AI_PERSON_KEY], + [state["seed_order"][1], state["seed_order"][2]], + ], + ) + + self.assertIn( + "当前轮到 alice", + self.command("/选1", sender="Bob"), + ) + self.assertIn( + "alice 准备完成！到 bob 做准备了", + self.command("/结束回合"), + ) + self.assertIn( + "当前轮到 bob", + self.command("/结束回合", sender="Carol"), + ) + self.assertIn( + "bob 准备完成！到 carol 做准备了", + self.command("/结束回合", sender="Bob"), + ) + self.assertIn( + "全员准备完毕", + self.command("/结束回合", sender="Carol"), + ) + settled = self.command("/自走棋结算") + self.assertIn("Taffy：AI 战队 自动运营", settled) + self.assertIn("Taffy：bob，请做好赛前准备", settled) + state = self.state() + self.assertEqual( + state["preparation_order"][0], + state["seed_order"][1], + ) + self.assertEqual( + self.engine.ledger.connection.execute( + """ + SELECT COUNT(*) FROM autochess_story_logs + WHERE story_type = 'ai_operation' + """ + ).fetchone()[0], + 1, + ) + self.assertEqual( + self.engine.ledger.connection.execute( + """ + SELECT COUNT(*) FROM autochess_story_logs + WHERE story_text NOT LIKE 'Taffy：%' + """ + ).fetchone()[0], + 0, + ) + + def test_identical_final_creates_persisted_overtime_stories(self): + state = self.launch(["Alice", "Bob"]) + for _ in range(3): + self.command("/自走棋强制结算") + state = self.state() + self.assertEqual(AUTOCHESS._current_stage(state)["kind"], "final") + first, second = state["seed_order"] + state["players"][second]["board"] = copy.deepcopy( + state["players"][first]["board"] + ) + state["players"][first]["tactic"] = "稳定" + state["players"][second]["tactic"] = "稳定" + self.save_state(state) + + with mock.patch.object(AUTOCHESS, "_hash_int", return_value=0): + response = self.command( + "/自走棋强制结算", + event_id="forced-overtime-final", + ) + self.assertIn("19-17", response) + final = self.engine.ledger.connection.execute( + """ + SELECT result_json FROM autochess_battles + WHERE battle_type = 'final_bo5' + """ + ).fetchone() + result = json.loads(final["result_json"]) + self.assertEqual(result["score"], "3-0") + self.assertEqual( + [item["score"] for item in result["maps"]], + ["19-17", "19-17", "19-17"], + ) + overtime = self.engine.ledger.connection.execute( + """ + SELECT story_text, facts_json FROM autochess_story_logs + WHERE story_type = 'overtime' + ORDER BY story_id + """ + ).fetchall() + self.assertEqual(len(overtime), 3) + for row in overtime: + self.assertIn("常规时间战成 15-15", row["story_text"]) + self.assertEqual(json.loads(row["facts_json"])["overtime_count"], 1) + + +if __name__ == "__main__": + unittest.main()