338 lines
11 KiB
Python
338 lines
11 KiB
Python
"""PVP对战插件 - 回合制战斗"""
|
||
|
||
from __future__ import annotations
|
||
|
||
from typing import Optional, Sequence
|
||
|
||
from PWF.Convention.Runtime.Architecture import Architecture
|
||
from PWF.Convention.Runtime.GlobalConfig import ConsoleFrontColor, ProjectConfig
|
||
|
||
from Plugins.WPSAPI import GuideEntry
|
||
from Plugins.WPSConfigSystem import WPSConfigAPI
|
||
from .combat_plugin_base import WPSCombatBase
|
||
|
||
|
||
logger: ProjectConfig = ProjectConfig()
|
||
|
||
|
||
class WPSCombatBattle(WPSCombatBase):
|
||
"""PVP对战插件"""
|
||
|
||
def get_guide_subtitle(self) -> str:
|
||
return "玩家间回合制对战指令集"
|
||
|
||
def collect_command_entries(self) -> Sequence[GuideEntry]:
|
||
return (
|
||
GuideEntry(
|
||
title="挑战",
|
||
identifier="擂台 挑战 <目标用户>",
|
||
description="向指定玩家发起 PVP 挑战。",
|
||
metadata={"别名": "challenge"},
|
||
icon="⚔️",
|
||
details=[
|
||
{"type": "list", "items": ["不可挑战自己。", "挑战在超时前需对方接受，否则自动失效。"]},
|
||
],
|
||
),
|
||
GuideEntry(
|
||
title="接受挑战",
|
||
identifier="擂台 接受挑战 <挑战ID>",
|
||
description="接受待处理的挑战并初始化战斗。",
|
||
metadata={"别名": "accept"},
|
||
icon="✅",
|
||
details=[
|
||
{"type": "steps", "items": ["输入挑战列表中的 ID。", "系统创建战斗记录并通知双方。"]},
|
||
],
|
||
),
|
||
GuideEntry(
|
||
title="拒绝挑战",
|
||
identifier="擂台 拒绝挑战 <挑战ID>",
|
||
description="拒绝尚未开始的挑战请求。",
|
||
metadata={"别名": "reject"},
|
||
icon="🚫",
|
||
),
|
||
GuideEntry(
|
||
title="战斗动作",
|
||
identifier="擂台 战斗 <战斗ID> <技能名>",
|
||
description="在战斗中释放技能或执行普攻。",
|
||
metadata={"别名": "battle"},
|
||
icon="🌀",
|
||
details=[
|
||
{"type": "list", "items": ["技能冷却与效果可在 `技能列表` 中查看。", "战斗为回合制，按顺序执行。"]},
|
||
],
|
||
),
|
||
GuideEntry(
|
||
title="投降",
|
||
identifier="擂台 投降 <战斗ID>",
|
||
description="主动认输并结束当前战斗。",
|
||
metadata={"别名": "surrender"},
|
||
icon="🏳️",
|
||
),
|
||
)
|
||
|
||
def collect_guide_entries(self) -> Sequence[GuideEntry]:
|
||
return (
|
||
{
|
||
"title": "挑战生命周期",
|
||
"description": "挑战需在配置的超时前被接受，超时将自动失效。",
|
||
},
|
||
{
|
||
"title": "战斗指令",
|
||
"description": "战斗中按回合输入技能，系统根据属性与技能效果计算伤害。",
|
||
},
|
||
)
|
||
|
||
def is_enable_plugin(self) -> bool:
|
||
return True
|
||
|
||
def wake_up(self) -> None:
|
||
super().wake_up()
|
||
logger.Log(
|
||
"Info",
|
||
f"{ConsoleFrontColor.GREEN}WPSCombatBattle 插件已加载{ConsoleFrontColor.RESET}"
|
||
)
|
||
self.register_plugin("擂台")
|
||
|
||
# 启动超时检查（定期轮询）
|
||
# TODO: 使用时钟调度器定期检查超时
|
||
|
||
async def callback(self, message: str, chat_id: int, user_id: int) -> Optional[str]:
|
||
"""
|
||
处理PVP命令
|
||
|
||
命令格式：
|
||
- 挑战 <目标用户>
|
||
- 接受挑战 <挑战ID>
|
||
- 拒绝挑战 <挑战ID>
|
||
- 战斗 <战斗ID> <技能名>
|
||
- 投降 <战斗ID>
|
||
"""
|
||
message = self.parse_message_after_at(message).strip()
|
||
|
||
tokens = message.split()
|
||
|
||
if not tokens:
|
||
return await self.send_markdown_message(
|
||
self._help_message(),
|
||
chat_id,
|
||
user_id
|
||
)
|
||
|
||
command = tokens[0].lower()
|
||
|
||
if command in ["挑战", "challenge"]:
|
||
return await self._handle_challenge(chat_id, user_id, tokens[1:])
|
||
|
||
elif command in ["接受挑战", "accept"]:
|
||
return await self._handle_accept(chat_id, user_id, tokens[1:])
|
||
|
||
elif command in ["拒绝挑战", "reject"]:
|
||
return await self._handle_reject(chat_id, user_id, tokens[1:])
|
||
|
||
elif command in ["投降", "surrender"]:
|
||
return await self._handle_surrender(chat_id, user_id, tokens[1:])
|
||
|
||
else:
|
||
# 默认视为战斗动作
|
||
return await self._handle_battle_action(chat_id, user_id, tokens)
|
||
|
||
async def _handle_challenge(
|
||
self,
|
||
chat_id: int,
|
||
user_id: int,
|
||
args: list
|
||
) -> Optional[str]:
|
||
"""处理挑战命令"""
|
||
if not args:
|
||
return await self.send_markdown_message(
|
||
"❌ 请指定目标用户\n用法：`挑战 <目标用户名|用户ID>`",
|
||
chat_id,
|
||
user_id
|
||
)
|
||
target_identifier = args[0]
|
||
target_id, error_message = self._resolve_target_user(target_identifier)
|
||
if error_message:
|
||
return await self.send_markdown_message(error_message, chat_id, user_id)
|
||
if target_id is None:
|
||
return await self.send_markdown_message(
|
||
"❌ 未找到目标用户，请确认用户名或ID",
|
||
chat_id,
|
||
user_id
|
||
)
|
||
if target_id == user_id:
|
||
return await self.send_markdown_message(
|
||
"❌ 不能挑战自己",
|
||
chat_id,
|
||
user_id
|
||
)
|
||
service = self.service()
|
||
success, msg, challenge_id = service.create_pvp_challenge(user_id, target_id)
|
||
return await self.send_markdown_message(msg, chat_id, user_id)
|
||
|
||
async def _handle_accept(
|
||
self,
|
||
chat_id: int,
|
||
user_id: int,
|
||
args: list
|
||
) -> Optional[str]:
|
||
"""处理接受挑战"""
|
||
if not args:
|
||
return await self.send_markdown_message(
|
||
"❌ 请指定挑战ID\n用法：`接受挑战 <挑战ID>`",
|
||
chat_id,
|
||
user_id
|
||
)
|
||
|
||
try:
|
||
challenge_id = int(args[0])
|
||
except ValueError:
|
||
return await self.send_markdown_message(
|
||
"❌ 挑战ID格式错误",
|
||
chat_id,
|
||
user_id
|
||
)
|
||
|
||
service = self.service()
|
||
success, msg, battle_id = service.accept_challenge(challenge_id, user_id)
|
||
|
||
return await self.send_markdown_message(msg, chat_id, user_id)
|
||
|
||
async def _handle_reject(
|
||
self,
|
||
chat_id: int,
|
||
user_id: int,
|
||
args: list
|
||
) -> Optional[str]:
|
||
"""处理拒绝挑战"""
|
||
if not args:
|
||
return await self.send_markdown_message(
|
||
"❌ 请指定挑战ID\n用法：`拒绝挑战 <挑战ID>`",
|
||
chat_id,
|
||
user_id
|
||
)
|
||
|
||
try:
|
||
challenge_id = int(args[0])
|
||
except ValueError:
|
||
return await self.send_markdown_message(
|
||
"❌ 挑战ID格式错误",
|
||
chat_id,
|
||
user_id
|
||
)
|
||
|
||
service = self.service()
|
||
success, msg = service.reject_challenge(challenge_id, user_id)
|
||
|
||
return await self.send_markdown_message(msg, chat_id, user_id)
|
||
|
||
async def _handle_battle_action(
|
||
self,
|
||
chat_id: int,
|
||
user_id: int,
|
||
tokens: list
|
||
) -> Optional[str]:
|
||
"""处理战斗动作"""
|
||
if len(tokens) < 2:
|
||
return await self.send_markdown_message(
|
||
"❌ 命令格式错误\n用法：`战斗 <战斗ID> <技能名>`",
|
||
chat_id,
|
||
user_id
|
||
)
|
||
|
||
try:
|
||
battle_id = int(tokens[0])
|
||
except ValueError:
|
||
return await self.send_markdown_message(
|
||
"❌ 战斗ID格式错误",
|
||
chat_id,
|
||
user_id
|
||
)
|
||
|
||
skill_name = " ".join(tokens[1:])
|
||
|
||
service = self.service()
|
||
success, msg = service.execute_battle_action(battle_id, user_id, skill_name)
|
||
|
||
return await self.send_markdown_message(msg, chat_id, user_id)
|
||
|
||
async def _handle_surrender(
|
||
self,
|
||
chat_id: int,
|
||
user_id: int,
|
||
args: list
|
||
) -> Optional[str]:
|
||
"""处理投降"""
|
||
if not args:
|
||
return await self.send_markdown_message(
|
||
"❌ 请指定战斗ID\n用法：`投降 <战斗ID>`",
|
||
chat_id,
|
||
user_id
|
||
)
|
||
|
||
try:
|
||
battle_id = int(args[0])
|
||
except ValueError:
|
||
return await self.send_markdown_message(
|
||
"❌ 战斗ID格式错误",
|
||
chat_id,
|
||
user_id
|
||
)
|
||
|
||
service = self.service()
|
||
success, msg = service.surrender_battle(battle_id, user_id)
|
||
|
||
return await self.send_markdown_message(msg, chat_id, user_id)
|
||
|
||
def _help_message(self) -> str:
|
||
"""帮助信息"""
|
||
return """# ⚔️ PVP对战系统
|
||
**命令格式：**
|
||
- `擂台 挑战 <目标用户名|用户ID>`：发起PVP挑战
|
||
- `擂台 接受挑战 <挑战ID>`：接受挑战
|
||
- `擂台 拒绝挑战 <挑战ID>`：拒绝挑战
|
||
- `擂台 战斗 <战斗ID> <技能名>`：执行战斗动作
|
||
- `擂台 投降 <战斗ID>`：投降
|
||
|
||
**说明：**
|
||
- 挑战有效期15分钟，超时自动失效
|
||
- 回合制战斗，速度高者先手
|
||
- 胜者获得1000积分（或失败者全部积分）
|
||
- 超时未操作视为失败
|
||
- 可随时投降
|
||
|
||
**示例：**
|
||
- `擂台 挑战 玩家A`：向用户名为玩家A的用户发起挑战
|
||
- `擂台 挑战 12345`：向用户12345发起挑战
|
||
- `擂台 接受挑战 1`：接受挑战ID为1的挑战
|
||
- `擂台 战斗 1 攻击`：在战斗1中使用"攻击"技能
|
||
- `擂台 投降 1`：在战斗1中投降
|
||
"""
|
||
|
||
def _resolve_target_user(self, identifier: str) -> tuple[Optional[int], Optional[str]]:
|
||
text = identifier.strip()
|
||
if not text:
|
||
return None, "❌ 请指定目标用户"
|
||
try:
|
||
config_api: WPSConfigAPI = Architecture.Get(WPSConfigAPI)
|
||
except Exception as exc:
|
||
logger.Log(
|
||
"Error",
|
||
f"{ConsoleFrontColor.RED}获取 WPSConfigAPI 实例失败: {exc}{ConsoleFrontColor.RESET}",
|
||
)
|
||
return None, "❌ 系统繁忙，请稍后重试"
|
||
try:
|
||
resolved_id = config_api.find_user_id_by_username(text)
|
||
if resolved_id is not None:
|
||
return resolved_id, None
|
||
except Exception as exc:
|
||
logger.Log(
|
||
"Error",
|
||
f"{ConsoleFrontColor.RED}解析用户名 {text} 失败: {exc}{ConsoleFrontColor.RESET}",
|
||
)
|
||
return None, "❌ 系统繁忙，请稍后重试"
|
||
if text.isdigit():
|
||
return int(text), None
|
||
return None, None
|
||
|
||
|
||
__all__ = ["WPSCombatBattle"]
|