Files
JWeBot/packages/jwebot-official-extensions/src/jwebot_official_extensions/cs_autochess.py
T

3340 lines
127 KiB
Python
Raw Blame History

This file contains ambiguous Unicode characters
This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.
"""Deterministic CS Major autochess rules for the reviewed local service."""
from __future__ import annotations
import hashlib
import json
import re
from importlib import resources
from typing import Any
DATA_FILES = ("rules.json", "players.json", "coaches.json", "narratives.json")
DATA_ROOT = "data/cs_autochess"
MIN_NARRATIVE_FRAGMENTS = 120
MULTI_REPLY_MARKER = "\x1eJWEBOT_MULTI_REPLY_V1\x1e"
def _load_json_file(filename: str) -> dict[str, Any]:
if filename not in DATA_FILES:
raise ValueError(f"unknown CS autochess data file: {filename}")
path = resources.files(__package__).joinpath(DATA_ROOT, filename)
with path.open("r", encoding="utf-8") as handle:
payload = json.load(handle)
if not isinstance(payload, dict):
raise ValueError(f"{filename} must contain a JSON object")
if int(payload.get("schema_version", 0)) != 1:
raise ValueError(f"{filename} has unsupported schema_version")
return payload
def _mapping(value: Any, context: str) -> dict[str, Any]:
if not isinstance(value, dict):
raise ValueError(f"{context} must be an object")
return dict(value)
def _sequence(value: Any, context: str) -> list[Any]:
if not isinstance(value, list):
raise ValueError(f"{context} must be a list")
return list(value)
def _text(value: Any, context: str) -> str:
result = str(value).strip()
if not result:
raise ValueError(f"{context} must be a non-empty string")
return result
def _int_value(value: Any, context: str) -> int:
if isinstance(value, bool):
raise ValueError(f"{context} must be an integer")
try:
return int(value)
except (TypeError, ValueError) as exc:
raise ValueError(f"{context} must be an integer") from exc
def _float_value(value: Any, context: str) -> float:
if isinstance(value, bool):
raise ValueError(f"{context} must be a number")
try:
return float(value)
except (TypeError, ValueError) as exc:
raise ValueError(f"{context} must be a number") from exc
def _int_key_int_dict(value: Any, context: str) -> dict[int, int]:
return {
_int_value(key, f"{context}.key"): _int_value(item, f"{context}.{key}")
for key, item in _mapping(value, context).items()
}
def _str_key_int_dict(value: Any, context: str) -> dict[str, int]:
return {
_text(key, f"{context}.key"): _int_value(item, f"{context}.{key}")
for key, item in _mapping(value, context).items()
}
def _section(rules: dict[str, Any], key: str) -> dict[str, Any]:
if key not in rules:
raise ValueError(f"rules.json missing {key}")
return _mapping(rules[key], f"rules.{key}")
def _load_rules() -> dict[str, Any]:
rules = _load_json_file("rules.json")
for key in (
"tournament",
"ai",
"roster",
"economy",
"coach",
"shop",
"draft",
"tactics",
"battle",
"pve_events",
"special_events",
):
_section(rules, key)
return rules
def _load_player_pool() -> tuple[dict[str, Any], ...]:
payload = _load_json_file("players.json")
players = _sequence(payload.get("players"), "players.players")
required = {
"id",
"name",
"cost",
"role",
"nation",
"team",
"teams",
"firepower",
"stability",
"clutch",
"leadership",
}
seen: set[str] = set()
normalized: list[dict[str, Any]] = []
for index, raw_item in enumerate(players, start=1):
item = _mapping(raw_item, f"players[{index}]")
missing = sorted(required - item.keys())
if missing:
raise ValueError(f"players[{index}] missing {', '.join(missing)}")
card_id = _text(item["id"], f"players[{index}].id")
if card_id in seen:
raise ValueError(f"duplicate player id: {card_id}")
seen.add(card_id)
team = _text(item["team"], f"players[{index}].team")
teams = [
_text(team_name, f"players[{index}].teams")
for team_name in _sequence(item["teams"], f"players[{index}].teams")
]
if team not in teams:
teams.insert(0, team)
teams = [team, *[team_name for team_name in teams if team_name != team]][:3]
cost = _int_value(item["cost"], f"players[{index}].cost")
if not 1 <= cost <= 5:
raise ValueError(f"players[{index}].cost must be 1..5")
normalized.append(
{
"id": card_id,
"name": _text(item["name"], f"players[{index}].name"),
"cost": cost,
"role": _text(item["role"], f"players[{index}].role"),
"nation": _text(item["nation"], f"players[{index}].nation"),
"team": team,
"teams": teams,
"firepower": _int_value(item["firepower"], f"players[{index}].firepower"),
"stability": _int_value(item["stability"], f"players[{index}].stability"),
"clutch": _int_value(item["clutch"], f"players[{index}].clutch"),
"leadership": _int_value(item["leadership"], f"players[{index}].leadership"),
}
)
if len(normalized) < 100:
raise ValueError("CS autochess player pool must contain at least 100 cards")
return tuple(normalized)
def _load_coach_pool() -> tuple[dict[str, Any], ...]:
payload = _load_json_file("coaches.json")
coaches = _sequence(payload.get("coaches"), "coaches.coaches")
required = {"id", "name", "level", "style", "firepower", "stability", "clutch", "bonus"}
seen: set[str] = set()
normalized: list[dict[str, Any]] = []
for index, raw_item in enumerate(coaches, start=1):
item = _mapping(raw_item, f"coaches[{index}]")
missing = sorted(required - item.keys())
if missing:
raise ValueError(f"coaches[{index}] missing {', '.join(missing)}")
coach_id = _text(item["id"], f"coaches[{index}].id")
if coach_id in seen:
raise ValueError(f"duplicate coach id: {coach_id}")
seen.add(coach_id)
normalized.append(
{
"id": coach_id,
"name": _text(item["name"], f"coaches[{index}].name"),
"level": _int_value(item["level"], f"coaches[{index}].level"),
"style": _text(item["style"], f"coaches[{index}].style"),
"firepower": _int_value(item["firepower"], f"coaches[{index}].firepower"),
"stability": _int_value(item["stability"], f"coaches[{index}].stability"),
"clutch": _int_value(item["clutch"], f"coaches[{index}].clutch"),
"bonus": _int_value(item["bonus"], f"coaches[{index}].bonus"),
}
)
if len(normalized) < 10:
raise ValueError("CS autochess coach pool must contain at least 10 coaches")
return tuple(normalized)
def _load_narratives() -> dict[str, tuple[str, ...]]:
payload = _load_json_file("narratives.json")
fragments = _mapping(payload.get("fragments"), "narratives.fragments")
normalized: dict[str, tuple[str, ...]] = {}
for category, raw_items in fragments.items():
key = _text(category, "narratives category")
items = tuple(
_text(item, f"narratives.{key}")
for item in _sequence(raw_items, f"narratives.{key}")
)
if not items:
raise ValueError(f"narratives.{key} must not be empty")
normalized[key] = items
total = sum(len(items) for items in normalized.values())
if total < MIN_NARRATIVE_FRAGMENTS:
raise ValueError(
f"CS autochess needs at least {MIN_NARRATIVE_FRAGMENTS} narrative fragments"
)
return normalized
def _synergy_rule(value: Any, context: str) -> dict[str, Any]:
raw = _mapping(value, context)
thresholds = []
for index, item in enumerate(_sequence(raw.get("thresholds"), f"{context}.thresholds"), start=1):
rule = _mapping(item, f"{context}.thresholds[{index}]")
thresholds.append(
{
"min_count": _int_value(rule.get("min_count"), f"{context}.thresholds[{index}].min_count"),
"bonus": _int_value(rule.get("bonus"), f"{context}.thresholds[{index}].bonus"),
}
)
thresholds.sort(key=lambda item: int(item["min_count"]), reverse=True)
return {
"max_sources": _int_value(raw.get("max_sources"), f"{context}.max_sources"),
"thresholds": thresholds,
}
_RULES = _load_rules()
_TOURNAMENT_RULES = _section(_RULES, "tournament")
_AI_RULES = _section(_RULES, "ai")
_ROSTER_RULES = _section(_RULES, "roster")
_ECONOMY_RULES = _section(_RULES, "economy")
_COACH_RULES = _section(_RULES, "coach")
_SHOP_RULES = _section(_RULES, "shop")
_DRAFT_RULES = _section(_RULES, "draft")
_TACTIC_RULES = _section(_RULES, "tactics")
_BATTLE_RULES = _section(_RULES, "battle")
MIN_PLAYERS = _int_value(_TOURNAMENT_RULES.get("min_players"), "rules.tournament.min_players")
MAX_PLAYERS = _int_value(_TOURNAMENT_RULES.get("max_players"), "rules.tournament.max_players")
HOST_NAME = _text(_TOURNAMENT_RULES.get("host_name"), "rules.tournament.host_name")
EVENT_TITLE_TEMPLATE = _text(
_TOURNAMENT_RULES.get("event_title_template"),
"rules.tournament.event_title_template",
)
AI_PERSON_KEY = _text(_AI_RULES.get("primary_person_key"), "rules.ai.primary_person_key")
AI_DISPLAY_NAME = _text(_AI_RULES.get("primary_display_name"), "rules.ai.primary_display_name")
AI_SECOND_PERSON_KEY = _text(_AI_RULES.get("secondary_person_key"), "rules.ai.secondary_person_key")
AI_SECOND_DISPLAY_NAME = _text(_AI_RULES.get("secondary_display_name"), "rules.ai.secondary_display_name")
AI_POWER_MULTIPLIER = _float_value(_AI_RULES.get("power_multiplier"), "rules.ai.power_multiplier")
STARTING_GOLD = _int_value(_ECONOMY_RULES.get("starting_gold"), "rules.economy.starting_gold")
STARTING_HP = _int_value(_ECONOMY_RULES.get("starting_hp"), "rules.economy.starting_hp")
_ROUND_INCOME_RULES = _mapping(_ECONOMY_RULES.get("round_income"), "rules.economy.round_income")
ROUND_INCOME_BASE = _int_value(_ROUND_INCOME_RULES.get("base"), "rules.economy.round_income.base")
ROUND_INTEREST_DIVISOR = _int_value(_ROUND_INCOME_RULES.get("interest_divisor"), "rules.economy.round_income.interest_divisor")
ROUND_INTEREST_CAP = _int_value(_ROUND_INCOME_RULES.get("interest_cap"), "rules.economy.round_income.interest_cap")
_TRAINING_REWARDS = _mapping(_ECONOMY_RULES.get("training_rewards"), "rules.economy.training_rewards")
TRAINING_WIN_GOLD = _int_value(_TRAINING_REWARDS.get("win_gold"), "rules.economy.training_rewards.win_gold")
TRAINING_LOSS_GOLD = _int_value(_TRAINING_REWARDS.get("loss_gold"), "rules.economy.training_rewards.loss_gold")
TRAINING_BYE_GOLD = _int_value(_TRAINING_REWARDS.get("bye_gold"), "rules.economy.training_rewards.bye_gold")
_SERIES_REWARDS = _mapping(_ECONOMY_RULES.get("series_rewards"), "rules.economy.series_rewards")
GROUP_SERIES_WIN_GOLD = _int_value(_SERIES_REWARDS.get("group_win_gold"), "rules.economy.series_rewards.group_win_gold")
PLAYOFF_SERIES_WIN_GOLD = _int_value(_SERIES_REWARDS.get("playoff_win_gold"), "rules.economy.series_rewards.playoff_win_gold")
STARTER_COUNT = _int_value(_ROSTER_RULES.get("starter_count"), "rules.roster.starter_count")
BENCH_LIMIT = _int_value(_ROSTER_RULES.get("bench_limit"), "rules.roster.bench_limit")
STAR_MULTIPLIERS = {
_int_value(star, "rules.roster.star_multipliers.key"): _float_value(multiplier, f"rules.roster.star_multipliers.{star}")
for star, multiplier in _mapping(_ROSTER_RULES.get("star_multipliers"), "rules.roster.star_multipliers").items()
}
POWER_WEIGHTS = _mapping(_ROSTER_RULES.get("unit_power_weights"), "rules.roster.unit_power_weights")
POWER_FIREPOWER_WEIGHT = _float_value(POWER_WEIGHTS.get("firepower"), "rules.roster.unit_power_weights.firepower")
POWER_STABILITY_WEIGHT = _float_value(POWER_WEIGHTS.get("stability"), "rules.roster.unit_power_weights.stability")
POWER_CLUTCH_WEIGHT = _float_value(POWER_WEIGHTS.get("clutch"), "rules.roster.unit_power_weights.clutch")
POWER_LEADERSHIP_THRESHOLD = _int_value(POWER_WEIGHTS.get("leadership_threshold"), "rules.roster.unit_power_weights.leadership_threshold")
POWER_LEADERSHIP_SCALE = _float_value(POWER_WEIGHTS.get("leadership_scale"), "rules.roster.unit_power_weights.leadership_scale")
TEAM_SYNERGY_RULE = _synergy_rule(_ROSTER_RULES.get("team_synergy"), "rules.roster.team_synergy")
NATION_SYNERGY_RULE = _synergy_rule(_ROSTER_RULES.get("nation_synergy"), "rules.roster.nation_synergy")
COACH_UPGRADE_COSTS = _int_key_int_dict(_COACH_RULES.get("upgrade_costs"), "rules.coach.upgrade_costs")
COACH_BONUSES = _int_key_int_dict(_COACH_RULES.get("fallback_bonuses"), "rules.coach.fallback_bonuses")
SHOP_SIZE = _int_value(_SHOP_RULES.get("size"), "rules.shop.size")
REFRESH_COST = _int_value(_SHOP_RULES.get("refresh_cost"), "rules.shop.refresh_cost")
COST_POOL_COPIES = _int_key_int_dict(_SHOP_RULES.get("cost_pool_copies"), "rules.shop.cost_pool_copies")
SHOP_ODDS = {
_int_value(tier, "rules.shop.odds.tier"): _int_key_int_dict(costs, f"rules.shop.odds.{tier}")
for tier, costs in _mapping(_SHOP_RULES.get("odds"), "rules.shop.odds").items()
}
DRAFT_ROUNDS = _int_value(_DRAFT_RULES.get("rounds"), "rules.draft.rounds")
DRAFT_COACH_ROUND = _int_value(_DRAFT_RULES.get("coach_round"), "rules.draft.coach_round")
DRAFT_COACH_CHOICE_COUNT = _int_value(_DRAFT_RULES.get("coach_choice_count"), "rules.draft.coach_choice_count")
DRAFT_PLAYER_CHOICE_COUNT = _int_value(_DRAFT_RULES.get("player_choice_count"), "rules.draft.player_choice_count")
DRAFT_PLAYER_COPY_COUNT = _int_value(_DRAFT_RULES.get("player_copy_count"), "rules.draft.player_copy_count")
DRAFT_ROUND_TWO_PREFERRED_COSTS = tuple(
_int_value(item, "rules.draft.round_two_preferred_costs")
for item in _sequence(_DRAFT_RULES.get("round_two_preferred_costs"), "rules.draft.round_two_preferred_costs")
)
DRAFT_LATER_ROUND_PREFERRED_COSTS = tuple(
_int_value(item, "rules.draft.later_round_preferred_costs")
for item in _sequence(_DRAFT_RULES.get("later_round_preferred_costs"), "rules.draft.later_round_preferred_costs")
)
TACTICS = tuple(
_text(item, "rules.tactics.names")
for item in _sequence(_TACTIC_RULES.get("names"), "rules.tactics.names")
)
TACTIC_COUNTERS = {
_text(key, "rules.tactics.counters.key"): _text(value, f"rules.tactics.counters.{key}")
for key, value in _mapping(_TACTIC_RULES.get("counters"), "rules.tactics.counters").items()
}
ATTRIBUTE_COUNTERS = tuple(
(
_text(pair[0], "rules.tactics.attribute_counters.strong"),
_text(pair[1], "rules.tactics.attribute_counters.weak"),
)
for pair in _sequence(_TACTIC_RULES.get("attribute_counters"), "rules.tactics.attribute_counters")
)
ROLE_COUNTERS = {
_text(key, "rules.tactics.role_counters.key"): _text(value, f"rules.tactics.role_counters.{key}")
for key, value in _mapping(_TACTIC_RULES.get("role_counters"), "rules.tactics.role_counters").items()
}
TACTIC_AXIS_MULTIPLIER = _float_value(_TACTIC_RULES.get("selected_axis_multiplier"), "rules.tactics.selected_axis_multiplier")
TACTIC_AXIS_PERCENT_DIVISOR = _float_value(_TACTIC_RULES.get("axis_percent_divisor"), "rules.tactics.axis_percent_divisor")
TACTIC_AXIS_PERCENT_CAP = _float_value(_TACTIC_RULES.get("axis_percent_cap"), "rules.tactics.axis_percent_cap")
TACTIC_ROLE_PERCENT_SCALE = _float_value(_TACTIC_RULES.get("role_percent_scale"), "rules.tactics.role_percent_scale")
TACTIC_ROLE_PERCENT_CAP = _float_value(_TACTIC_RULES.get("role_percent_cap"), "rules.tactics.role_percent_cap")
TACTIC_COUNTER_PERCENT = _float_value(_TACTIC_RULES.get("tactic_counter_percent"), "rules.tactics.tactic_counter_percent")
TACTIC_TOTAL_PERCENT_CAP = _float_value(_TACTIC_RULES.get("total_percent_cap"), "rules.tactics.total_percent_cap")
MAP_POOL = tuple(
_text(item, "rules.battle.maps")
for item in _sequence(_BATTLE_RULES.get("maps"), "rules.battle.maps")
)
_BATTLE_DAMAGE = _mapping(_BATTLE_RULES.get("damage"), "rules.battle.damage")
BATTLE_BO5_SWEEP_DAMAGE = _int_value(_BATTLE_DAMAGE.get("bo5_sweep"), "rules.battle.damage.bo5_sweep")
BATTLE_SWEEP_DAMAGE = _int_value(_BATTLE_DAMAGE.get("sweep"), "rules.battle.damage.sweep")
BATTLE_NON_SWEEP_DAMAGE = _int_value(_BATTLE_DAMAGE.get("non_sweep"), "rules.battle.damage.non_sweep")
BATTLE_MR_ROUNDS = _int_value(_BATTLE_RULES.get("mr_rounds"), "rules.battle.mr_rounds")
BATTLE_TARGET_SCORE = _int_value(_BATTLE_RULES.get("target_score"), "rules.battle.target_score")
BATTLE_OVERTIME_START_SCORE = _int_value(_BATTLE_RULES.get("overtime_start_score"), "rules.battle.overtime_start_score")
BATTLE_OVERTIME_TARGET_SCORE = _int_value(_BATTLE_RULES.get("overtime_target_score"), "rules.battle.overtime_target_score")
BATTLE_OVERTIME_WIN_BY = _int_value(_BATTLE_RULES.get("overtime_win_by"), "rules.battle.overtime_win_by")
BATTLE_OVERTIME_SAFETY_CAP = _int_value(_BATTLE_RULES.get("overtime_safety_cap", 72), "rules.battle.overtime_safety_cap")
BATTLE_OVERTIME_SEGMENT_ROUNDS = _int_value(_BATTLE_RULES.get("overtime_segment_rounds"), "rules.battle.overtime_segment_rounds")
MATCH_BEST_OF = _str_key_int_dict(_BATTLE_RULES.get("match_best_of"), "rules.battle.match_best_of")
_ROUND_PROBABILITY_RULES = _mapping(_BATTLE_RULES.get("round_probability"), "rules.battle.round_probability")
ROUND_PROBABILITY_BASE = _float_value(_ROUND_PROBABILITY_RULES.get("base"), "rules.battle.round_probability.base")
ROUND_PROBABILITY_MIN = _float_value(_ROUND_PROBABILITY_RULES.get("min"), "rules.battle.round_probability.min")
ROUND_PROBABILITY_MAX = _float_value(_ROUND_PROBABILITY_RULES.get("max"), "rules.battle.round_probability.max")
ROUND_POWER_EDGE_CAP = _float_value(_ROUND_PROBABILITY_RULES.get("power_edge_cap"), "rules.battle.round_probability.power_edge_cap")
ROUND_POWER_EDGE_SCALE = _float_value(_ROUND_PROBABILITY_RULES.get("power_edge_scale"), "rules.battle.round_probability.power_edge_scale")
ROUND_PRESSURE_CAP = _float_value(_ROUND_PROBABILITY_RULES.get("pressure_cap"), "rules.battle.round_probability.pressure_cap")
ROUND_PRESSURE_SCALE = _float_value(_ROUND_PROBABILITY_RULES.get("pressure_scale"), "rules.battle.round_probability.pressure_scale")
ROUND_FIRST_HALF_SIDE_SWING = _float_value(_ROUND_PROBABILITY_RULES.get("first_half_side_swing"), "rules.battle.round_probability.first_half_side_swing")
ROUND_SECOND_HALF_SIDE_SWING = _float_value(_ROUND_PROBABILITY_RULES.get("second_half_side_swing"), "rules.battle.round_probability.second_half_side_swing")
ROUND_JITTER_MOD = _int_value(_ROUND_PROBABILITY_RULES.get("jitter_mod"), "rules.battle.round_probability.jitter_mod")
ROUND_JITTER_CENTER = _int_value(_ROUND_PROBABILITY_RULES.get("jitter_center"), "rules.battle.round_probability.jitter_center")
ROUND_JITTER_DIVISOR = _float_value(_ROUND_PROBABILITY_RULES.get("jitter_divisor"), "rules.battle.round_probability.jitter_divisor")
SCORE_BANDS = tuple(
{
"max_loser_score": _int_value(item.get("max_loser_score"), "rules.battle.score_bands.max_loser_score"),
"label": _text(item.get("label"), "rules.battle.score_bands.label"),
}
for item in (
_mapping(raw_item, "rules.battle.score_bands.item")
for raw_item in _sequence(_BATTLE_RULES.get("score_bands"), "rules.battle.score_bands")
)
)
PVE_EVENTS = {
key: (
_text(item.get("name"), f"rules.pve_events.{key}.name"),
_int_value(item.get("target"), f"rules.pve_events.{key}.target"),
_int_value(item.get("win_gold"), f"rules.pve_events.{key}.win_gold"),
_int_value(item.get("loss_gold"), f"rules.pve_events.{key}.loss_gold"),
)
for key, item in (
(str(name), _mapping(value, f"rules.pve_events.{name}"))
for name, value in _section(_RULES, "pve_events").items()
)
}
SPECIAL_EVENTS = {
key: (
_text(item.get("name"), f"rules.special_events.{key}.name"),
_int_value(item.get("target"), f"rules.special_events.{key}.target"),
_int_value(item.get("win_gold"), f"rules.special_events.{key}.win_gold"),
_int_value(item.get("loss_gold"), f"rules.special_events.{key}.loss_gold"),
)
for key, item in (
(str(name), _mapping(value, f"rules.special_events.{name}"))
for name, value in _section(_RULES, "special_events").items()
)
}
PLAYER_POOL = _load_player_pool()
CARD_BY_ID = {card["id"]: card for card in PLAYER_POOL}
COACH_POOL = _load_coach_pool()
COACH_BY_ID = {coach["id"]: coach for coach in COACH_POOL}
NARRATIVE_FRAGMENTS = _load_narratives()
def _hash_int(seed: str) -> int:
return int.from_bytes(
hashlib.sha256(seed.encode("utf-8", "replace")).digest()[:8],
"big",
)
def _pick_fragment(category: str, seed: str, **values: Any) -> str:
fragments = tuple(NARRATIVE_FRAGMENTS.get(category, ()))
if not fragments:
return ""
template = fragments[_hash_int(f"{category}|{seed}") % len(fragments)]
return template.format(**values)
def _narrative_inventory_size() -> int:
return sum(len(items) for items in NARRATIVE_FRAGMENTS.values())
def _pack_response_parts(parts: list[str]) -> str:
cleaned = [part.strip() for part in parts if part.strip()]
if len(cleaned) <= 1:
return cleaned[0] if cleaned else "没有可发送内容"
return MULTI_REPLY_MARKER + json.dumps(
cleaned,
ensure_ascii=False,
separators=(",", ":"),
)
def _unpack_response_parts(value: str) -> list[str]:
if not value.startswith(MULTI_REPLY_MARKER):
return [value]
try:
payload = json.loads(value[len(MULTI_REPLY_MARKER) :])
except json.JSONDecodeError:
return [value]
if not isinstance(payload, list):
return [value]
parts = [str(item).strip() for item in payload if str(item).strip()]
return parts or [value]
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],
}
]
group_rounds = _round_robin_rounds(keys)
challenge_after = 1 if len(group_rounds) <= 2 else 2
for day, pairs in enumerate(group_rounds, start=1):
plan.append(
{
"kind": "group",
"phase": f"group_{day}",
"name": f"小组赛第 {day} 轮",
"day": day,
"pairs": pairs,
"last_group_day": day == len(group_rounds),
}
)
if day == challenge_after:
phase = "pve_media" if len(keys) % 2 == 0 else "pve_sponsor"
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]]]
if len(keys) == 4
else [],
"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_variant") == "two_plus_two":
return (
"本届采用 2 人+2 AI 测试赛:报名种子 1v4、2v3 进行 BO5 "
"半决赛,胜者进入 BO5 总决赛"
)
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"] * POWER_FIREPOWER_WEIGHT
+ card["stability"] * POWER_STABILITY_WEIGHT
+ card["clutch"] * POWER_CLUTCH_WEIGHT
)
if card["role"] == "指挥":
base += (
max(0, card["leadership"] - POWER_LEADERSHIP_THRESHOLD)
* POWER_LEADERSHIP_SCALE
)
multiplier = STAR_MULTIPLIERS[int(unit["star"])]
return base * multiplier
def _synergy_values(
counts: dict[str, int],
rules: dict[str, Any],
) -> list[tuple[int, int, str]]:
values: list[tuple[int, int, str]] = []
thresholds = list(rules["thresholds"])
for name, count in counts.items():
bonus = 0
for rule in thresholds:
if count >= int(rule["min_count"]):
bonus = int(rule["bonus"])
break
if bonus:
values.append((bonus, count, name))
values.sort(reverse=True)
return values[: int(rules["max_sources"])]
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 = STAR_MULTIPLIERS[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 = _synergy_values(teams, TEAM_SYNERGY_RULE)
team_bonus = sum(item[0] for item in team_values)
nation_values = _synergy_values(nations, NATION_SYNERGY_RULE)
nation_bonus = sum(item[0] for item in nation_values)
coach_id = str(player.get("coach_id", ""))
coach = COACH_BY_ID.get(coach_id)
if coach is not None:
axes["火力"] += float(coach["firepower"])
axes["稳定"] += float(coach["stability"])
axes["残局"] += float(coach["clutch"])
coach_bonus = int(coach["bonus"])
coach_label = f"{coach['name']}+{coach_bonus}%"
else:
coach_level = int(player.get("coach_level") or 1)
coach_bonus = COACH_BONUSES.get(coach_level, 0)
coach_label = f"教练+{coach_bonus}%"
total = base * (100 + team_bonus + nation_bonus + coach_bonus) / 100
if player.get("is_ai"):
total *= AI_POWER_MULTIPLIER
labels = [
*(f"{name}{count}" for _, count, name in team_values[:2]),
*(f"{name}{count}" for _, count, name in nation_values[:1]),
coach_label,
]
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] *= TACTIC_AXIS_MULTIPLIER
axes_b[tactic_b] *= TACTIC_AXIS_MULTIPLIER
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(
-TACTIC_AXIS_PERCENT_CAP,
min(TACTIC_AXIS_PERCENT_CAP, axis_net / TACTIC_AXIS_PERCENT_DIVISOR),
)
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(
-TACTIC_ROLE_PERCENT_CAP,
min(
TACTIC_ROLE_PERCENT_CAP,
(role_edge_a - role_edge_b) * TACTIC_ROLE_PERCENT_SCALE,
),
)
tactic_percent = 0.0
if TACTIC_COUNTERS[tactic_a] == tactic_b:
tactic_percent = TACTIC_COUNTER_PERCENT
elif TACTIC_COUNTERS[tactic_b] == tactic_a:
tactic_percent = -TACTIC_COUNTER_PERCENT
total = max(
-TACTIC_TOTAL_PERCENT_CAP,
min(TACTIC_TOTAL_PERCENT_CAP, 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
story_text = str(text).strip()
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": story_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,
"team_name": f"Team {display_name}" if not is_ai else display_name,
"coach_id": "",
"coach_level": 0,
"coach_bonus": 0,
"coach_style": "",
"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"{player['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
def _round_probability(
adjusted_a: float,
adjusted_b: float,
seed: str,
round_index: int,
score_a: int,
score_b: int,
) -> float:
average = max(1.0, (adjusted_a + adjusted_b) / 2)
power_edge = max(
-ROUND_POWER_EDGE_CAP,
min(
ROUND_POWER_EDGE_CAP,
(adjusted_a - adjusted_b) / average * ROUND_POWER_EDGE_SCALE,
),
)
pressure = max(
-ROUND_PRESSURE_CAP,
min(ROUND_PRESSURE_CAP, (score_b - score_a) * ROUND_PRESSURE_SCALE),
)
side_swing = (
ROUND_FIRST_HALF_SIDE_SWING
if round_index <= BATTLE_MR_ROUNDS // 2
else ROUND_SECOND_HALF_SIDE_SWING
)
if round_index > BATTLE_MR_ROUNDS:
side_swing = 0.0
jitter = (
(_hash_int(f"{seed}|prob|{round_index}") % ROUND_JITTER_MOD)
- ROUND_JITTER_CENTER
) / ROUND_JITTER_DIVISOR
return max(
ROUND_PROBABILITY_MIN,
min(
ROUND_PROBABILITY_MAX,
ROUND_PROBABILITY_BASE + power_edge + pressure + side_swing + jitter,
),
)
def _simulate_map(
state: dict[str, Any],
key_a: str,
key_b: str,
seed: str,
map_index: int,
adjusted_a: float,
adjusted_b: float,
) -> dict[str, Any]:
score_a = 0
score_b = 0
rounds: list[str] = []
current_key = ""
current_run = 0
best_run_key = ""
best_run = 0
def record_round(round_index: int) -> None:
nonlocal score_a, score_b, current_key, current_run
nonlocal best_run_key, best_run
probability_a = _round_probability(
adjusted_a,
adjusted_b,
seed,
round_index,
score_a,
score_b,
)
roll = (_hash_int(f"{seed}|round|{round_index}") % 10_000) / 10_000
winner_key = key_a if roll < probability_a else key_b
if winner_key == key_a:
score_a += 1
else:
score_b += 1
rounds.append(winner_key)
if winner_key == current_key:
current_run += 1
else:
current_key = winner_key
current_run = 1
if current_run > best_run:
best_run = current_run
best_run_key = winner_key
for round_index in range(1, BATTLE_MR_ROUNDS + 1):
record_round(round_index)
if score_a >= BATTLE_TARGET_SCORE or score_b >= BATTLE_TARGET_SCORE:
break
round_index = len(rounds)
if (
score_a == BATTLE_OVERTIME_START_SCORE
and score_b == BATTLE_OVERTIME_START_SCORE
):
while (
max(score_a, score_b) < BATTLE_OVERTIME_TARGET_SCORE
or abs(score_a - score_b) < BATTLE_OVERTIME_WIN_BY
):
round_index += 1
record_round(round_index)
if round_index > BATTLE_OVERTIME_SAFETY_CAP:
break
winner_key = key_a if score_a > score_b else key_b
loser_key = key_b if winner_key == key_a else key_a
winner_score = score_a if winner_key == key_a else score_b
loser_score = score_b if winner_key == key_a else score_a
overtime_rounds = max(0, len(rounds) - BATTLE_MR_ROUNDS)
overtime_count = (
overtime_rounds + BATTLE_OVERTIME_SEGMENT_ROUNDS - 1
) // BATTLE_OVERTIME_SEGMENT_ROUNDS
if overtime_count:
band = "加时险胜"
else:
band = next(
str(item["label"])
for item in SCORE_BANDS
if loser_score <= int(item["max_loser_score"])
)
map_name = MAP_POOL[(map_index - 1) % len(MAP_POOL)]
winner_team = _team_name(state["players"][winner_key])
loser_team = _team_name(state["players"][loser_key])
best_run_team = (
_team_name(state["players"][best_run_key]) if best_run_key else winner_team
)
opening_team = _team_name(
state["players"][key_a if _hash_int(seed + "|open") % 2 == 0 else key_b]
)
actor = _key_player_name(state, winner_key, map_index)
pause = _pick_fragment("pauses", f"{seed}|pause")
run_text = (
_pick_fragment(
"runs",
f"{seed}|run",
team=best_run_team,
count=best_run,
)
if best_run >= 4
else pause
)
overtime_story = ""
if overtime_count:
overtime_story = (
f"{map_name} 常规时间战成 15-15。{winner_team} 在加时第 "
f"{overtime_count} 段抓住 {loser_team} 的转点空隙,以 "
f"{winner_score}-{loser_score} 收下地图。"
)
return {
"map_index": map_index,
"map_name": map_name,
"winner_key": winner_key,
"winner": state["players"][winner_key]["display_name"],
"loser_key": loser_key,
"loser": state["players"][loser_key]["display_name"],
"score": f"{winner_score}-{loser_score}",
"score_a": score_a,
"score_b": score_b,
"band": band,
"overtime_count": overtime_count,
"overtime_story": overtime_story,
"rounds_played": len(rounds),
"best_run_key": best_run_key,
"best_run": best_run,
"opening_text": _pick_fragment("openings", f"{seed}|opening", team=opening_team),
"run_text": run_text,
"pause_text": pause,
"clutch_text": _pick_fragment("clutches", f"{seed}|clutch", player=actor),
"economy_text": _pick_fragment("economy", f"{seed}|economy", team=loser_team),
"closer_text": _pick_fragment("closers", f"{seed}|closer", team=winner_team),
"adjusted_a": round(adjusted_a, 2),
"adjusted_b": round(adjusted_b, 2),
}
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
)
item = _simulate_map(
state,
key_a,
key_b,
f"{seed}|map|{map_index}",
map_index,
adjusted_a,
adjusted_b,
)
winner_key = str(item["winner_key"])
if winner_key == key_a:
wins_a += 1
else:
wins_b += 1
item["fluctuation_a_bps"] = fluct_a
item["fluctuation_b_bps"] = fluct_b
maps.append(item)
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 = (
BATTLE_BO5_SWEEP_DAMAGE
if sweep and best_of == 5
else BATTLE_SWEEP_DAMAGE
if sweep
else BATTLE_NON_SWEEP_DAMAGE
)
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"]
tactic_note = (
"战术克制明显"
if abs(float(matchup["a_total_percent"])) >= 5.5
else "战术拉扯接近"
)
return _host_text(
f"{result['winner']} 在 {result['series_format']} 以 {result['score']} "
f"击败 {result['loser']}({map_scores})。"
f"战术面是 {matchup['tactic_a']} 对 {matchup['tactic_b']},{tactic_note}。"
)
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=MATCH_BEST_OF["group"],
)
_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=MATCH_BEST_OF["semifinal"],
)
_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=MATCH_BEST_OF["final"],
)
_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 = ROUND_INCOME_BASE + min(
ROUND_INTEREST_CAP,
int(player["gold"]) // ROUND_INTEREST_DIVISOR,
)
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
def _coach_line(slot: int, coach_id: str) -> str:
coach = COACH_BY_ID[coach_id]
return (
f"{slot}. {coach['name']}|{coach['level']}级教练|{coach['style']}|"
f"火{coach['firepower']} 稳{coach['stability']} 残{coach['clutch']}|"
f"全队+{coach['bonus']}%"
)
def _team_name(player: dict[str, Any]) -> str:
if player.get("is_ai"):
return str(player["display_name"])
return f"Team {player['display_name']}"
def _seed_lines(state: dict[str, Any]) -> str:
return "、".join(
f"{state['players'][key]['seed']}号{state['players'][key]['display_name']}"
for key in state["seed_order"]
)
def _draft_order(state: dict[str, Any], round_number: int) -> list[str]:
order = list(state["seed_order"])
if round_number == 2:
order.reverse()
return order
def _draft_candidate_cards(state: dict[str, Any], round_number: int) -> list[str]:
cards = [
card["id"]
for card in PLAYER_POOL
if int(state["pool"].get(card["id"], 0)) > 0
]
cards.sort(
key=lambda card_id: _hash_int(
f"{state['run_id']}|draft|{round_number}|{card_id}"
)
)
selected: list[str] = []
preferred_costs = (
DRAFT_ROUND_TWO_PREFERRED_COSTS
if round_number == 2
else DRAFT_LATER_ROUND_PREFERRED_COSTS
)
for cost in preferred_costs:
for card_id in cards:
if (
card_id not in selected
and CARD_BY_ID[card_id]["cost"] == cost
and len(selected) < DRAFT_PLAYER_CHOICE_COUNT
):
selected.append(card_id)
if len(selected) >= DRAFT_PLAYER_CHOICE_COUNT:
break
for card_id in cards:
if len(selected) >= DRAFT_PLAYER_CHOICE_COUNT:
break
if card_id not in selected:
selected.append(card_id)
return selected[:DRAFT_PLAYER_CHOICE_COUNT]
def _start_draft_round(state: dict[str, Any], round_number: int) -> None:
draft_type = "coach" if round_number == DRAFT_COACH_ROUND else "player"
candidates = (
[coach["id"] for coach in COACH_POOL[:DRAFT_COACH_CHOICE_COUNT]]
if draft_type == "coach"
else _draft_candidate_cards(state, round_number)
)
state["phase"] = f"draft_{draft_type}_{round_number}"
state["draft"] = {
"round": round_number,
"type": draft_type,
"order": _draft_order(state, round_number),
"cursor": 0,
"candidates": candidates,
}
def _draft_round_title(draft: dict[str, Any]) -> str:
number = int(draft["round"])
return (
"第一轮选秀,本轮选教练"
if number == DRAFT_COACH_ROUND
else f"第 {number} 轮选秀,本轮选选手"
)
def _draft_board_part(state: dict[str, Any]) -> str:
draft = state["draft"]
order_names = "、".join(
state["players"][key]["display_name"]
for key in draft["order"]
)
lines = [
f"欢迎来到{_draft_round_title(draft)}。",
f"选择顺序:{order_names}",
]
if draft["type"] == "coach":
lines.extend(
_coach_line(index, coach_id)
for index, coach_id in enumerate(draft["candidates"], start=1)
)
lines.append(f"操作:轮到你时发送 /选1~/选{DRAFT_COACH_CHOICE_COUNT}。")
else:
lines.extend(
_card_line(index, card_id)
for index, card_id in enumerate(draft["candidates"], start=1)
)
lines.append("操作:轮到你时发送 /选1~/选6。")
return "\n".join(lines)
def _draft_prompt_part(state: dict[str, Any]) -> str:
draft = state["draft"]
cursor = int(draft.get("cursor", 0))
order = list(draft.get("order", []))
if cursor >= len(order):
return "本轮选秀已完成。"
key = str(order[cursor])
name = state["players"][key]["display_name"]
command = (
f"/选1~/选{DRAFT_COACH_CHOICE_COUNT}"
if draft["type"] == "coach"
else f"/选1~/选{DRAFT_PLAYER_CHOICE_COUNT}"
)
return f"现在轮到 @{name} 做出选择,使用 {command}。"
def _add_unit_to_roster(
player: dict[str, Any],
card_id: str,
*,
copies: int = 1,
) -> None:
unit = _new_unit(card_id, copies=copies)
if len(player["board"]) < STARTER_COUNT:
player["board"].append(unit)
else:
player["bench"].append(unit)
def _apply_draft_choice(
state: dict[str, Any],
person_key: str,
slot: int,
) -> str:
draft = state["draft"]
candidates = list(draft["candidates"])
player = state["players"][person_key]
if not 1 <= slot <= len(candidates):
raise ValueError("选秀编号无效")
choice_id = str(candidates[slot - 1])
if draft["type"] == "coach":
coach = COACH_BY_ID[choice_id]
player["coach_id"] = choice_id
player["coach_level"] = int(coach["level"])
player["coach_bonus"] = int(coach["bonus"])
player["coach_style"] = str(coach["style"])
text = (
f"{player['display_name']} 选择了教练 {coach['name']}!"
f"{coach['level']}级,{coach['style']},全队+{coach['bonus']}%。"
)
else:
card = CARD_BY_ID[choice_id]
draft_copies = DRAFT_PLAYER_COPY_COUNT
if int(state["pool"].get(choice_id, 0)) < draft_copies:
raise ValueError("这名选手已经被选走")
state["pool"][choice_id] -= draft_copies
_add_unit_to_roster(player, choice_id, copies=draft_copies)
text = (
f"{player['display_name']} 选择了 {card['name']}!"
f"{card['cost']}费 {card['role']},{card['team']}/{card['nation']},"
"以 2 星素材加入阵容。"
)
candidates.pop(slot - 1)
draft["candidates"] = candidates
draft["cursor"] = int(draft.get("cursor", 0)) + 1
return text
def _auto_pick_slot(state: dict[str, Any], ai_key: str) -> int:
draft = state["draft"]
candidates = list(draft["candidates"])
if draft["type"] == "coach":
values = [
(
int(COACH_BY_ID[str(coach_id)]["bonus"]),
-int(COACH_BY_ID[str(coach_id)]["level"]),
index,
)
for index, coach_id in enumerate(candidates, start=1)
]
return min(values)[2]
values = []
for index, card_id in enumerate(candidates, start=1):
card = CARD_BY_ID[str(card_id)]
values.append(
(
int(card["cost"]),
int(card["firepower"]) + int(card["stability"]) + int(card["clutch"]),
_hash_int(f"{state['run_id']}|ai|{ai_key}|{card_id}"),
index,
)
)
return min(values)[3]
def _auto_advance_draft(state: dict[str, Any]) -> list[str]:
parts: list[str] = []
while state.get("phase", "").startswith("draft_"):
draft = state["draft"]
order = list(draft["order"])
cursor = int(draft.get("cursor", 0))
if cursor >= len(order):
break
key = str(order[cursor])
if not state["players"][key].get("is_ai"):
break
slot = _auto_pick_slot(state, key)
parts.append(_apply_draft_choice(state, key, slot))
return parts
def _finish_draft_round_if_needed(state: dict[str, Any]) -> list[str]:
parts: list[str] = []
while state.get("phase", "").startswith("draft_"):
draft = state["draft"]
if int(draft.get("cursor", 0)) < len(draft.get("order", [])):
break
round_number = int(draft["round"])
if round_number < DRAFT_ROUNDS:
_start_draft_round(state, round_number + 1)
parts.append(_draft_board_part(state))
parts.extend(_auto_advance_draft(state))
continue
_ensure_full_rosters(state)
state["phase"] = "lineup_prep"
state.pop("draft", None)
_reset_ready(state, _human_keys(state))
parts.append(
"三轮选秀结束。每队已经整理为 5 名首发、1 名教练和 3 个备战席。"
)
parts.append(
"请各位玩家使用 /阵容 查看队伍,使用 /上阵 备战编号 [首发位] 调整首发;准备完成后发送 /结束回合。"
)
break
return parts
def _ensure_full_rosters(state: dict[str, Any]) -> None:
for key in state["join_order"]:
player = state["players"][key]
if not player.get("coach_id"):
fallback = COACH_POOL[-1] if player.get("is_ai") else COACH_POOL[0]
player["coach_id"] = fallback["id"]
player["coach_level"] = int(fallback["level"])
player["coach_bonus"] = int(fallback["bonus"])
player["coach_style"] = str(fallback["style"])
owned = {
str(unit["card_id"])
for unit in [*player.get("board", []), *player.get("bench", [])]
}
target_total = STARTER_COUNT + BENCH_LIMIT
while len(player["board"]) + len(player["bench"]) < target_total:
candidates = [
card["id"]
for card in PLAYER_POOL
if card["id"] not in owned and int(state["pool"].get(card["id"], 0)) > 0
]
candidates.sort(
key=lambda card_id: (
CARD_BY_ID[card_id]["cost"]
if player.get("is_ai")
else abs(int(CARD_BY_ID[card_id]["cost"]) - 2),
int(CARD_BY_ID[card_id]["cost"]),
_hash_int(f"{state['run_id']}|fill|{key}|{card_id}"),
)
)
if not candidates:
break
card_id = candidates[0]
state["pool"][card_id] -= 1
owned.add(card_id)
_add_unit_to_roster(player, card_id)
units = [*player.get("board", []), *player.get("bench", [])]
units.sort(key=_unit_power, reverse=True)
player["board"] = units[:STARTER_COUNT]
player["bench"] = units[STARTER_COUNT : STARTER_COUNT + BENCH_LIMIT]
player["shop"] = _draw_shop(
state,
f"{state['run_id']}|lineup|{key}",
)
def _reset_ready(state: dict[str, Any], keys: list[str]) -> None:
ready_keys = set(keys)
state["ready_keys"] = list(keys)
for key in state["join_order"]:
player = state["players"][key]
player["ready"] = bool(player.get("is_ai") or key not in ready_keys)
def _ready_waiting(state: dict[str, Any]) -> list[str]:
return [
key
for key in state.get("ready_keys", [])
if not state["players"][key].get("ready")
]
def _team_power_line(state: dict[str, Any], key: str) -> str:
player = state["players"][key]
power = lineup_power(player)
coach = COACH_BY_ID.get(str(player.get("coach_id", "")))
coach_text = (
f"{coach['name']} {coach['level']}级"
if coach is not None
else f"{player.get('coach_level', 1)}级"
)
return (
f"{player['display_name']} #{state['season_number']}:摩拉 {player['gold']};"
f"生命 {player['hp']};教练 {coach_text};战术{player['tactic']};"
f"阵容 {power['power']:.1f};商店{'已锁' if player['shop_locked'] else '未锁'}"
)
def _full_lineup_text(state: dict[str, Any], player: dict[str, Any]) -> str:
if len(player.get("board", [])) != STARTER_COUNT:
drafted = [
CARD_BY_ID[str(unit["card_id"])]["name"]
for unit in [*player.get("board", []), *player.get("bench", [])]
]
coach = COACH_BY_ID.get(str(player.get("coach_id", "")))
return "\n".join(
[
f"{player['display_name']} #{state['season_number']}:选秀中阵容",
f"教练:{coach['name'] if coach else '未选择'}",
"已选选手:" + ("、".join(drafted) if drafted else "暂无"),
]
)
return _lineup_text(state, player)
def _shop_status_text(state: dict[str, Any], player: dict[str, Any]) -> str:
if len(player.get("board", [])) != STARTER_COUNT:
return _full_lineup_text(state, player)
return _team_power_line(state, player) + "\n" + "\n".join(
_card_line(index, card_id)
for index, card_id in enumerate(player.get("shop", []), start=1)
) + "\n操作:/选1~/选6;/刷新;/锁定;/阵容;/战术 火力|稳定|残局;/结束回合"
def _training_schedule(state: dict[str, Any]) -> list[dict[str, Any]]:
rounds: list[dict[str, Any]] = []
keys = list(state["seed_order"])
for round_index, pairs in enumerate(_round_robin_rounds(keys), start=1):
paired = {key for pair in pairs for key in pair}
bye = next((key for key in keys if key not in paired), "")
rounds.append(
{
"round": round_index,
"pairs": [[str(pair[0]), str(pair[1])] for pair in pairs],
"bye": bye,
}
)
return rounds
def _ensure_training_schedule(state: dict[str, Any]) -> None:
if state.get("training_rounds"):
return
state["training_rounds"] = _training_schedule(state)
state["training_round_cursor"] = 0
def _training_pairing_line(state: dict[str, Any], round_item: dict[str, Any]) -> str:
rendered = [
f"{_team_name(state['players'][pair[0]])} vs {_team_name(state['players'][pair[1]])}"
for pair in round_item.get("pairs", [])
]
bye = str(round_item.get("bye") or "")
if bye:
rendered.append(f"{_team_name(state['players'][bye])} vs 本地陪练队")
return ";".join(rendered)
def _training_match_part(
state: dict[str, Any],
result: dict[str, Any],
round_number: int,
) -> str:
item = result["maps"][0]
winner_key = str(result["winner_key"])
loser_key = str(result["loser_key"])
winner = state["players"][winner_key]
loser = state["players"][loser_key]
return "\n".join(
[
(
f"训练赛第 {round_number} 轮:{_team_name(winner)} "
f"vs {_team_name(loser)}"
),
(
f"{item['map_name']}:{item['opening_text']}"
f"{item['run_text']}"
),
(
f"{item['clutch_text']}{item['closer_text']}"
f"{_team_name(winner)} 以 {item['score']} 拿下本图。"
),
(
f"摩拉:{_team_name(winner)} +{TRAINING_WIN_GOLD},"
f"{_team_name(loser)} +{TRAINING_LOSS_GOLD}。"
),
]
)
def _training_bye_part(state: dict[str, Any], key: str, round_number: int) -> str:
player = state["players"][key]
player["gold"] = int(player["gold"]) + TRAINING_BYE_GOLD
hook = _pick_fragment(
"training_hooks",
f"{state['run_id']}|training|bye|{round_number}|{key}",
)
return (
f"训练赛第 {round_number} 轮:{_team_name(player)} 与本地陪练队完成热身。\n"
f"{hook}\n摩拉:{_team_name(player)} +{TRAINING_BYE_GOLD}。"
)
def _settle_training_round(
state: dict[str, Any],
event_id: str,
) -> tuple[list[str], list[dict[str, Any]], list[dict[str, Any]]]:
_ensure_training_schedule(state)
rounds = list(state.get("training_rounds", []))
cursor = int(state.get("training_round_cursor", 0))
if cursor >= len(rounds):
state["phase"] = "training_complete_manage"
_prepare_new_shops(state, "training-complete")
_reset_ready(state, _human_keys(state))
return [
"训练赛已经全部结束。可以最后 D 牌、调阵容;所有真人发送 /结束回合 后进入小组赛。"
], [], []
round_item = rounds[cursor]
round_number = int(round_item["round"])
state["round_number"] = round_number
parts: list[str] = []
stories: list[dict[str, Any]] = []
battles: list[dict[str, Any]] = []
if cursor == 0:
intro = (
f"欢迎来到{state['event_title']}!你们已经到达酒店宾馆,"
f"训练赛共 {len(rounds)} 轮。训练赛不扣生命,只产出运营摩拉。"
)
parts.append(intro)
stories.append(_story(state, "training_open", intro))
hook = _pick_fragment(
"training_hooks",
f"{state['run_id']}|training|hook|{round_number}",
)
parts.append(
f"训练赛第 {round_number}/{len(rounds)} 轮对阵:"
f"{_training_pairing_line(state, round_item)}\n{hook}"
)
for match_index, pair in enumerate(round_item.get("pairs", []), start=1):
key_a, key_b = str(pair[0]), str(pair[1])
result = _battle(
state,
key_a,
key_b,
f"{state['run_id']}|training|{round_number}|{match_index}|{event_id}",
best_of=MATCH_BEST_OF["training"],
)
winner_key = str(result["winner_key"])
loser_key = str(result["loser_key"])
state["players"][winner_key]["gold"] = (
int(state["players"][winner_key]["gold"]) + TRAINING_WIN_GOLD
)
state["players"][loser_key]["gold"] = (
int(state["players"][loser_key]["gold"]) + TRAINING_LOSS_GOLD
)
result["training_rewards"] = {
winner_key: TRAINING_WIN_GOLD,
loser_key: TRAINING_LOSS_GOLD,
}
battle = _battle_record(
state,
battle_type="training_bo1",
player_a=key_a,
player_b=key_b,
result=result,
)
text = _training_match_part(state, result, round_number)
parts.append(text)
stories.append(_story(state, "training_result", text, result))
battles.append(battle)
bye = str(round_item.get("bye") or "")
if bye:
text = _training_bye_part(state, bye, round_number)
parts.append(text)
stories.append(_story(state, "training_bye", text, {"person_key": bye}))
state["training_round_cursor"] = cursor + 1
_prepare_new_shops(state, f"training-{round_number}")
_reset_ready(state, _human_keys(state))
if cursor + 1 < len(rounds):
state["phase"] = "training_manage"
parts.append(
"本轮训练结束。现在是运营窗口:/刷新 是 D 牌,/选1~/选6 是买店内选手,"
"/锁定 可保留商店,/上阵 可调整首发;所有真人发送 /结束回合 后进入下一轮训练。"
)
else:
state["phase"] = "training_complete_manage"
parts.append(
"训练赛全部结束。可以最后 D 牌、调阵容;所有真人发送 /结束回合 后进入小组赛。"
)
return parts, stories, battles
def _prepare_new_shops(state: dict[str, Any], reason: str) -> None:
for key in state["join_order"]:
player = state["players"][key]
if not player.get("shop_locked"):
player["shop"] = _draw_shop(
state,
f"{state['run_id']}|{reason}|{key}|{state.get('match_sequence', 0)}",
)
def _group_schedule(state: dict[str, Any]) -> list[dict[str, Any]]:
matches: list[dict[str, Any]] = []
for round_index, pairs in enumerate(_round_robin_rounds(list(state["seed_order"])), start=1):
paired = {key for pair in pairs for key in pair}
bye = next((key for key in state["seed_order"] if key not in paired), "")
for match_index, pair in enumerate(pairs, start=1):
matches.append(
{
"stage": "group",
"round": round_index,
"match": match_index,
"pair": [str(pair[0]), str(pair[1])],
"best_of": MATCH_BEST_OF["group"],
"bye": bye,
}
)
return matches
def _match_title(state: dict[str, Any], match: dict[str, Any]) -> str:
key_a, key_b = match["pair"]
player_a = state["players"][key_a]
player_b = state["players"][key_b]
if match["stage"] == "group":
return (
f"小组赛第 {match['round']} 轮第 {match['match']} 场:"
f"{_team_name(player_a)} vs {_team_name(player_b)}(BO{match['best_of']})"
)
label = "淘汰赛半决赛" if match["stage"] == "semifinal" else "总决赛"
return f"{label}:{_team_name(player_a)} vs {_team_name(player_b)}(BO{match['best_of']})"
def _match_ready_keys(state: dict[str, Any], match: dict[str, Any]) -> list[str]:
if match.get("stage") == "group":
return _human_keys(state)
return [
key
for key in match.get("pair", [])
if not state["players"][key].get("is_ai")
]
def _set_current_match(state: dict[str, Any], match: dict[str, Any]) -> list[str]:
state["current_match"] = match
state["phase"] = f"{match['stage']}_ready"
keys = _match_ready_keys(state, match)
_reset_ready(state, keys)
parts = [_match_title(state, match)]
if keys:
names = "、".join(state["players"][key]["display_name"] for key in keys)
parts.append(f"{names},请确认阵容;准备完成发送 /结束回合。")
else:
parts.append("本场为 AI 内战,将自动结算。")
return parts
def _start_group_parts(state: dict[str, Any]) -> list[str]:
state["group_matches"] = _group_schedule(state)
state["group_match_cursor"] = 0
state["match_sequence"] = 0
pair_lines = []
rounds = _round_robin_rounds(list(state["seed_order"]))
for round_index, pairs in enumerate(rounds, start=1):
rendered = []
paired = {key for pair in pairs for key in pair}
bye = next((key for key in state["seed_order"] if key not in paired), "")
for match_index, pair in enumerate(pairs, start=1):
rendered.append(
f"第 {round_index} 轮第 {match_index} 场 "
f"{_team_name(state['players'][pair[0]])} vs {_team_name(state['players'][pair[1]])}"
)
if bye:
rendered.append(f"{state['players'][bye]['display_name']} 轮空")
pair_lines.append(";".join(rendered))
parts = [
(
f"欢迎来到{state['event_title']}正赛小组赛!"
f"小组赛采用单循环 BO{MATCH_BEST_OF['group']}。"
),
"本阶段对阵:\n" + "\n".join(pair_lines),
]
if state["group_matches"]:
parts.extend(_set_current_match(state, state["group_matches"][0]))
return parts
def _key_player_name(state: dict[str, Any], key: str, offset: int) -> str:
board = state["players"][key].get("board", [])
if not board:
return state["players"][key]["display_name"]
unit = board[offset % len(board)]
return str(CARD_BY_ID[str(unit["card_id"])]["name"])
def _series_report_parts(
state: dict[str, Any],
match: dict[str, Any],
result: dict[str, Any],
) -> list[str]:
key_a, key_b = match["pair"]
matchup = result["matchup"]
parts = [
(
f"{_match_title(state, match)} 开始。"
f"战术面是 {matchup['tactic_a']} 对 {matchup['tactic_b']},"
"本场按逐回合模拟推进。"
)
]
wins: dict[str, int] = {key_a: 0, key_b: 0}
for index, item in enumerate(result["maps"], start=1):
winner_key = str(item["winner_key"])
loser_key = str(item["loser_key"])
wins[winner_key] += 1
lead_text = (
f"{_team_name(state['players'][winner_key])} 在系列赛中 "
f"{wins[winner_key]}-{wins[loser_key]} {'领先' if wins[winner_key] > wins[loser_key] else '追平'}"
)
parts.append(
f"图{index} {item['map_name']}:{item['opening_text']}"
f"{item['run_text']}{item['clutch_text']}"
f"{item['economy_text']}{item['closer_text']}"
f"{_team_name(state['players'][winner_key])} 以 {item['score']} 拿下地图;{lead_text}。"
)
winner_key = str(result["winner_key"])
loser_key = str(result["loser_key"])
reward = (
GROUP_SERIES_WIN_GOLD
if match["stage"] == "group"
else PLAYOFF_SERIES_WIN_GOLD
)
state["players"][winner_key]["gold"] = int(state["players"][winner_key]["gold"]) + reward
final_text = (
f"系列赛结束,{_team_name(state['players'][winner_key])} 以 {result['score']} 击败 "
f"{_team_name(state['players'][loser_key])},获得 {reward} 摩拉。"
"这笔钱可以马上用于 /刷新、/选1~/选6 或锁店等下一轮运营。"
)
parts[-1] = parts[-1] + "\n" + final_text
if len(parts) <= 5:
return parts
return [parts[0], parts[1], parts[2], "\n".join(parts[3:-1]), parts[-1]]
def _settle_current_match(
state: dict[str, Any],
event_id: str,
) -> tuple[list[str], list[dict[str, Any]], list[dict[str, Any]]]:
match = dict(state["current_match"])
key_a, key_b = match["pair"]
state["match_sequence"] = int(state.get("match_sequence", 0)) + 1
result = _battle(
state,
key_a,
key_b,
f"{state['run_id']}|{match['stage']}|{state['match_sequence']}|{event_id}",
best_of=int(match["best_of"]),
)
_apply_match_result(
state,
key_a,
key_b,
result,
group_stage=match["stage"] == "group",
)
battle_type = "group_bo3" if match["stage"] == "group" else f"{match['stage']}_bo5"
battle = _battle_record(
state,
battle_type=battle_type,
player_a=key_a,
player_b=key_b,
result=result,
)
parts = _series_report_parts(state, match, result)
story = _story(state, f"{match['stage']}_result", "\n".join(parts), result)
return parts, [story], [battle]
def _standings_part(state: dict[str, Any]) -> str:
lines = ["小组赛排名:"]
for index, key in enumerate(_rankings(state), start=1):
player = state["players"][key]
total = max(1, int(player["wins"]) + int(player["losses"]))
rate = int(player["wins"]) / total
lines.append(
f"{index}. {_team_name(player)} {player['wins']}胜{player['losses']}负 胜率{rate:.0%}"
)
return "\n".join(lines)
def _special_event_parts(
state: dict[str, Any],
event_id: str,
) -> tuple[list[str], list[dict[str, Any]]]:
event_key = "media" if _hash_int(state["run_id"] + event_id) % 2 == 0 else "sponsor"
event_kind, target, win_gold, loss_gold = SPECIAL_EVENTS[event_key]
parts = [f"{event_kind}突然插入。本届只会出现一次这种挑战;完成后继续小组赛。"]
stories: list[dict[str, Any]] = []
for key in state["join_order"]:
player = state["players"][key]
power = lineup_power(player)["power"]
won = power >= target
reward = win_gold if won else loss_gold
player["gold"] = int(player["gold"]) + reward
hook = _pick_fragment(
"challenges",
f"{state['run_id']}|{event_kind}|{event_id}|{key}",
)
text = (
f"{_team_name(player)} {'完成' if won else '勉强应付'}{event_kind}:"
f"{hook}获得 {reward} 摩拉。"
)
parts.append(text)
stories.append(_story(state, "special_event", text, {"person_key": key}))
state["special_event_done"] = True
state["phase"] = "special_manage"
_prepare_new_shops(state, event_kind)
_reset_ready(state, _human_keys(state))
parts.append("挑战结束。可以继续 D 牌和调整阵容;所有真人发送 /结束回合 后继续小组赛。")
return parts, stories
def _playoff_matches_from_rankings(state: dict[str, Any]) -> list[dict[str, Any]]:
ranking = _rankings(state)
if len(ranking) >= 4:
qualifiers = ranking[:4]
state["qualifiers"] = qualifiers
return [
{
"stage": "semifinal",
"round": 1,
"match": 1,
"pair": [qualifiers[0], qualifiers[3]],
"best_of": MATCH_BEST_OF["semifinal"],
},
{
"stage": "semifinal",
"round": 1,
"match": 2,
"pair": [qualifiers[1], qualifiers[2]],
"best_of": MATCH_BEST_OF["semifinal"],
},
]
finalists = ranking[:2]
state["finalists"] = finalists
return [
{
"stage": "final",
"round": 1,
"match": 1,
"pair": finalists,
"best_of": MATCH_BEST_OF["final"],
}
]
def _playoff_pairings_part(state: dict[str, Any]) -> str:
lines = ["淘汰赛对阵:"]
for match in state.get("playoff_matches", []):
key_a, key_b = match["pair"]
seed_a = state["players"][key_a].get("seed", "?")
seed_b = state["players"][key_b].get("seed", "?")
label = "半决赛" if match["stage"] == "semifinal" else "总决赛"
lines.append(
f"{label}:{_team_name(state['players'][key_a])}({seed_a}) vs {_team_name(state['players'][key_b])}({seed_b})"
)
return "\n".join(lines)
def _finish_tournament_parts(state: dict[str, Any]) -> list[str]:
champion_key = str(state["champion_key"])
champion = state["players"][champion_key]
power = lineup_power(champion)
state["status"] = "completed"
state["phase"] = "completed"
return [
f"{state['event_title']} 全部比赛结束!",
(
f"冠军:{_team_name(champion)}。最终阵容战力 {power['power']:.1f},"
f"小组赛 {champion['wins']}胜{champion['losses']}负。"
),
"颁奖:MVP 给到本届决赛关键先生;所有赛报已经写入自走棋档案,可用 /自走棋档案 查看。",
]
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战队,"
"分别开启 2 人+2 AI 或 3 人+1 AI 测试模式。\n"
f"开赛后 {DRAFT_ROUNDS} 轮选秀:第一轮固定 {DRAFT_COACH_CHOICE_COUNT} 名教练,"
"后续轮次选选手。"
"选秀完成后每队 5 名首发、1 名教练、3 个备战席。\n"
"/选1~/选6;/阵容;/刷新;/锁定;/上阵 备战编号 [首发位];/出售 备战编号。"
"/战术 火力|稳定|残局,稳定克火力、火力克残局、残局克稳定。\n"
"训练赛按参赛人数生成多轮真实对阵,每轮结束后可 D 牌和调阵容;"
f"随后进入 BO{MATCH_BEST_OF['group']} 单循环小组赛,"
f"半决赛/决赛 BO{MATCH_BEST_OF['final']}。"
"所有需要推进的阶段由相关真人发送 /结束回合,满足条件后自动进入下一步。"
)
def handle(self, row: dict[str, Any]) -> str:
return "\n".join(self.handle_parts(row))
def handle_parts(self, row: dict[str, Any]) -> list[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,
)
parts = result.pop("response_parts", None)
if isinstance(parts, list):
result["response"] = _pack_response_parts(
[str(part) for part in parts]
)
return result
packed = self.ledger.apply_autochess_action(
group,
sender,
event_id,
command,
execute,
)
return _unpack_response_parts(packed)
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"])
def respond(
current: dict[str, Any] | None,
parts: list[str] | str,
*,
stories: list[dict[str, Any]] | None = None,
battles: list[dict[str, Any]] | None = None,
) -> dict[str, Any]:
normalized = [parts] if isinstance(parts, str) else parts
normalized = [str(part) for part in normalized if str(part).strip()]
if not normalized:
normalized = ["没有可发送内容"]
return {
"state": current,
"response": normalized[0],
"response_parts": normalized,
"stories": stories or [],
"battles": battles or [],
}
if action == "unknown":
return respond(state, self.rules())
if action == "create":
legacy_active = (
state is not None
and state.get("status") in {"registration", "active"}
and int(state.get("schema_version") or 0) < 3
)
if (
state is not None
and state.get("status") in {"registration", "active"}
and not legacy_active
):
return respond(
state,
f"{state['event_title']} 已在进行中;发送 /自走棋状态 查看",
)
season = next_season
state = {
"schema_version": 3,
"run_id": f"major_{season:04d}",
"season_number": season,
"event_title": EVENT_TITLE_TEMPLATE.format(season=season),
"status": "registration",
"host_person_key": person_key,
"host_display_name": display_name,
"round_number": 1,
"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,
"ai_test_variant": "",
"ready_keys": [],
"qualifiers": [],
"finalists": [],
"champion_key": "",
"stories": [],
"battles": [],
"story_sequence": 0,
"battle_sequence": 0,
"special_event_done": False,
"training_rounds": [],
"training_round_cursor": 0,
}
text = (
f"{state['event_title']} 开放报名;主持人 {display_name} 已自动加入。"
"支持 2~8 人;发送 /自走棋报名,主持人发送 /自走棋开赛。"
)
if legacy_active:
text = "已保留旧版未完成赛事档案,并开启新版赛季。" + text
story = _story(state, "registration_open", text, {"host_person_key": person_key})
return respond(state, text, stories=[story])
if state is None:
return respond(None, "当前没有自走棋赛事;发送 /自走棋开始")
if action == "join":
if state["status"] != "registration":
return respond(state, "本届赛事已停止报名")
if state.get("ai_test_mode"):
return respond(state, "AI 测试模式已经锁定参赛名单,不能再加入真人。")
if person_key in state["players"]:
return respond(state, f"{display_name} 已经报名。")
if len(state["join_order"]) >= MAX_PLAYERS:
return respond(state, f"本届赛事已满 {MAX_PLAYERS} 人")
state["players"][person_key] = _participant(person_key, display_name)
state["join_order"].append(person_key)
text = f"{display_name} 加入赛事,当前 {len(state['join_order'])}/{MAX_PLAYERS} 人。"
story = _story(state, "player_joined", text, {"person_key": person_key})
return respond(state, text, stories=[story])
if action == "add_ai":
if state["status"] != "registration":
return respond(state, "AI 战队只能在报名阶段加入")
if person_key != state["host_person_key"]:
return respond(state, f"只有主持人 {state['host_display_name']} 可以加入 AI 战队")
if any(player.get("is_ai") for player in state["players"].values()):
return respond(state, "AI 测试战队已经加入")
human_keys = [
key for key in state["join_order"]
if not state["players"][key].get("is_ai")
]
if len(human_keys) not in {2, 3}:
return respond(state, "AI 测试模式要求先有且仅有 2 名或 3 名真人报名")
if len(human_keys) == 2:
ai_teams = (
(AI_PERSON_KEY, "AI 战队 A"),
(AI_SECOND_PERSON_KEY, AI_SECOND_DISPLAY_NAME),
)
variant = "two_plus_two"
text = "AI 战队 A、AI 战队 B 已作为第 3、4 种子加入;2 人+2 AI 测试赛已锁定。"
else:
ai_teams = ((AI_PERSON_KEY, AI_DISPLAY_NAME),)
variant = "three_plus_one"
text = "AI 战队已作为第 4 种子加入;3 人+1 AI 测试赛已锁定。"
for ai_key, ai_name in ai_teams:
state["players"][ai_key] = _participant(ai_key, ai_name, is_ai=True)
state["join_order"].append(ai_key)
state["ai_test_mode"] = True
state["ai_test_variant"] = variant
story = _story(state, "ai_team_joined", text, {"variant": variant})
return respond(state, text, stories=[story])
if action == "launch":
if state["status"] != "registration":
return respond(state, "本届赛事已经开赛或结束")
if person_key != state["host_person_key"]:
return respond(state, f"只有主持人 {state['host_display_name']} 可以开赛")
if len(state["join_order"]) < MIN_PLAYERS:
return respond(state, f"至少需要 {MIN_PLAYERS} 人才能开赛")
state["status"] = "active"
state["seed_order"] = list(state["join_order"])
state["round_plan"] = build_round_plan(list(state["join_order"]))
for seed, key in enumerate(state["seed_order"], start=1):
state["players"][key]["seed"] = seed
_start_draft_round(state, 1)
names = "、".join(
state["players"][key]["display_name"] for key in state["join_order"]
)
mode = (
"2 人+2 AI 测试赛"
if state.get("ai_test_variant") == "two_plus_two"
else "3 人+1 AI 测试赛"
if state.get("ai_test_variant") == "three_plus_one"
else f"{len(state['join_order'])} 队赛"
)
parts = [
f"欢迎大家来到{state['event_title']}!",
f"参赛者:{names}。",
(
f"种子顺序:{_seed_lines(state)}。本届采用 {mode}:"
f"先进行 {DRAFT_ROUNDS} 轮选秀,随后按轮次训练赛、"
f"小组赛 BO{MATCH_BEST_OF['group']} 单循环,"
f"淘汰赛 BO{MATCH_BEST_OF['final']};单图 MR15,"
f"常规时间 {BATTLE_OVERTIME_START_SCORE}-{BATTLE_OVERTIME_START_SCORE} 进入加时。"
),
_draft_board_part(state),
]
parts.extend(_auto_advance_draft(state))
parts.extend(_finish_draft_round_if_needed(state))
if state.get("phase", "").startswith("draft_"):
parts.append(_draft_prompt_part(state))
story = _story(
state,
"tournament_started",
"\n".join(parts),
{"players": list(state["join_order"])},
)
return respond(state, parts, stories=[story])
if action == "archive":
page_values = _parse_indices(arguments)
return respond(state, _archive_text(state, page_values[0] if page_values else 1))
if action == "schedule":
return respond(state, _schedule_text(state))
if state["status"] == "registration":
return respond(state, _schedule_text(state))
if state["status"] in {"completed", "cancelled"}:
if action in {"status", "lineup"}:
return respond(state, _schedule_text(state))
return respond(state, f"{state['event_title']} 已结束;发送 /自走棋开始 创建下一届")
player = state["players"].get(person_key)
if player is None:
return respond(state, "你没有参加本届赛事,只能查看 /自走棋赛程")
if action == "status":
return respond(state, _shop_status_text(state, player))
if action == "lineup":
return respond(state, _full_lineup_text(state, player))
if state.get("phase", "").startswith("draft_"):
if action != "buy":
return respond(state, _draft_prompt_part(state))
draft = state["draft"]
order = list(draft["order"])
cursor = int(draft.get("cursor", 0))
if cursor >= len(order):
return respond(state, "本轮选秀已经完成,请稍等主持进入下一轮。")
current_key = str(order[cursor])
if current_key != person_key:
return respond(
state,
f"当前选择的是 @{state['players'][current_key]['display_name']},请等待。",
)
values = _parse_indices(arguments)
if len(values) != 1:
return respond(
state,
(
f"用法:/选1~/选{DRAFT_COACH_CHOICE_COUNT}(教练)"
f"或 /选1~/选{DRAFT_PLAYER_CHOICE_COUNT}(选手)"
),
)
try:
text = _apply_draft_choice(state, person_key, values[0])
except ValueError as error:
return respond(state, str(error))
parts = [text]
parts.extend(_auto_advance_draft(state))
parts.extend(_finish_draft_round_if_needed(state))
if state.get("phase", "").startswith("draft_"):
parts.append(_draft_prompt_part(state))
story = _story(state, "draft_pick", text, {"person_key": person_key})
return respond(state, parts, stories=[story])
if action == "tactic":
value = re.sub(r"\s+", "", arguments.casefold())
aliases = {
"火力": "火力",
"fire": "火力",
"稳定": "稳定",
"stable": "稳定",
"残局": "残局",
"clutch": "残局",
}
if not value:
return respond(state, f"当前战术 {player['tactic']};用法:/战术 火力|稳定|残局")
tactic = aliases.get(value)
if tactic is None:
return respond(state, "战术只能选择:火力、稳定、残局")
player["tactic"] = tactic
text = f"{display_name} 将战术设为 {tactic}。"
story = _story(state, "tactic_selected", text, {"person_key": person_key, "tactic": tactic})
return respond(state, text, stories=[story])
if action == "coach":
return respond(state, "教练现在由第一轮选秀决定,本届不能用摩拉升级教练。")
if action == "buy":
values = _parse_indices(arguments)
if len(values) != 1 or not 1 <= values[0] <= SHOP_SIZE:
return respond(state, "用法:/选1 到 /选6")
slot = values[0] - 1
card_id = player["shop"][slot] if slot < len(player.get("shop", [])) else None
if card_id is None:
return respond(state, "这个商店位置已经购买或没有候选")
card = CARD_BY_ID[str(card_id)]
if int(player["gold"]) < int(card["cost"]):
return respond(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 respond(state, f"备战席已满 {BENCH_LIMIT} 格,请先出售")
if int(state["pool"].get(card_id, 0)) <= 0:
return respond(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})
return respond(state, text, stories=[story])
if action == "refresh":
if int(player["gold"]) < REFRESH_COST:
return respond(state, f"刷新需要 {REFRESH_COST} 摩拉,当前 {player['gold']}")
player["gold"] -= REFRESH_COST
player["shop"] = _draw_shop(
state,
f"{state['run_id']}|{state.get('phase')}|{event_id}|{person_key}|refresh",
)
return respond(state, _shop_status_text(state, player))
if action == "lock":
player["shop_locked"] = not bool(player["shop_locked"])
return respond(state, f"{display_name} {'锁定' if player['shop_locked'] else '解锁'}了商店。")
if action in {"field", "bench"}:
values = _parse_indices(arguments)
if not values:
return respond(state, "用法:/上阵 备战编号 [首发位],例如 /上阵 6 2")
found = _find_unit(player, values[0])
if found is None or found[0] != "备战":
return respond(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
text = (
f"{display_name} 用 {CARD_BY_ID[str(incoming['card_id'])]['name']} "
f"替换首发 {board_index + 1} 的 {CARD_BY_ID[str(outgoing['card_id'])]['name']}。"
)
story = _story(state, "lineup_change", text, {"person_key": person_key})
return respond(state, text, stories=[story])
if action == "sell":
values = _parse_indices(arguments)
if len(values) != 1:
return respond(state, "用法:/出售 备战编号")
found = _find_unit(player, values[0])
if found is None:
return respond(state, "没有这个阵容编号")
if found[0] == "首发":
return respond(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"])
return respond(state, f"{display_name} 出售 {card['name']},返还 {refund} 摩拉。")
if action == "ready":
if person_key not in state.get("ready_keys", []):
if state.get("phase") in {"semifinal_ready", "final_ready"}:
return respond(state, f"{display_name} 已不在当前淘汰赛对阵中,不需要准备。")
return respond(state, "当前阶段不需要你确认 /结束回合。")
player["ready"] = True
waiting = _ready_waiting(state)
if waiting:
names = "、".join(state["players"][key]["display_name"] for key in waiting)
return respond(state, f"{display_name} 准备完成;仍在等待:{names}。")
phase = str(state.get("phase"))
stories: list[dict[str, Any]] = []
battles: list[dict[str, Any]] = []
if phase in {"lineup_prep", "training_manage"}:
parts, stories, battles = _settle_training_round(state, event_id)
return respond(state, parts, stories=stories, battles=battles)
if phase == "training_complete_manage":
parts = _start_group_parts(state)
return respond(state, parts)
if phase == "group_ready":
parts, stories, battles = _settle_current_match(state, event_id)
state["group_match_cursor"] = int(state.get("group_match_cursor", 0)) + 1
state["last_completed_group_round"] = int(state["current_match"]["round"])
state["phase"] = "group_manage"
_prepare_new_shops(state, "group")
_reset_ready(state, _human_keys(state))
parts.append("本场结束。可以 D 牌、调阵容;所有真人发送 /结束回合 后进入下一场。")
return respond(state, parts, stories=stories, battles=battles)
if phase == "group_manage":
cursor = int(state.get("group_match_cursor", 0))
matches = list(state.get("group_matches", []))
last_round = int(state.get("last_completed_group_round", 0))
next_round = int(matches[cursor]["round"]) if cursor < len(matches) else 999
if (
not state.get("special_event_done")
and last_round in {1, 2}
and next_round > last_round
):
parts, stories = _special_event_parts(state, event_id)
return respond(state, parts, stories=stories)
if cursor < len(matches):
parts = _set_current_match(state, matches[cursor])
return respond(state, parts)
state["playoff_matches"] = _playoff_matches_from_rankings(state)
state["playoff_cursor"] = 0
parts = [_standings_part(state), _playoff_pairings_part(state)]
if state["playoff_matches"]:
parts.extend(_set_current_match(state, state["playoff_matches"][0]))
return respond(state, parts)
if phase == "special_manage":
cursor = int(state.get("group_match_cursor", 0))
matches = list(state.get("group_matches", []))
if cursor < len(matches):
return respond(state, _set_current_match(state, matches[cursor]))
return respond(state, "挑战结束,但没有后续小组赛对阵。")
if phase in {"semifinal_ready", "final_ready"}:
parts, stories, battles = _settle_current_match(state, event_id)
match = dict(state["current_match"])
winner_key = str(
battles[0]["result"]["winner_key"]
if battles else state["current_match"]["pair"][0]
)
if match["stage"] == "final":
state["champion_key"] = winner_key
parts.extend(_finish_tournament_parts(state))
return respond(state, parts, stories=stories, battles=battles)
winners = list(state.get("semifinal_winners", []))
winners.append(winner_key)
state["semifinal_winners"] = winners
state["playoff_cursor"] = int(state.get("playoff_cursor", 0)) + 1
cursor = int(state["playoff_cursor"])
playoff_matches = list(state.get("playoff_matches", []))
if cursor < len(playoff_matches):
parts.extend(_set_current_match(state, playoff_matches[cursor]))
return respond(state, parts, stories=stories, battles=battles)
final_match = {
"stage": "final",
"round": 1,
"match": 1,
"pair": winners[:2],
"best_of": MATCH_BEST_OF["final"],
}
state["finalists"] = winners[:2]
parts.append(
f"半决赛全部结束,胜者进入 BO{MATCH_BEST_OF['final']} 总决赛。"
)
parts.extend(_set_current_match(state, final_match))
if not state.get("ready_keys"):
final_parts, final_stories, final_battles = _settle_current_match(
state,
event_id + "-auto-final",
)
final_winner = str(final_battles[0]["result"]["winner_key"])
state["champion_key"] = final_winner
final_parts.extend(_finish_tournament_parts(state))
parts.extend(final_parts)
stories.extend(final_stories)
battles.extend(final_battles)
return respond(state, parts, stories=stories, battles=battles)
return respond(state, "当前阶段已经准备完成,请等待下一步。")
if action in {"settle", "force_settle"}:
return respond(state, "0.6.6 起赛事由 /结束回合 自动推进,不再需要主持人手动结算。")
if action == "cancel":
if person_key != state["host_person_key"]:
return respond(state, f"只有主持人 {state['host_display_name']} 可以取消赛事")
state["status"] = "cancelled"
state["phase"] = "cancelled"
text = f"{state['event_title']} 已由主持人取消;已完成剧情仍保留在档案。"
story = _story(state, "tournament_cancelled", text)
return respond(state, text, stories=[story])
return respond(state, self.rules())