107 lines
3.6 KiB
Python
107 lines
3.6 KiB
Python
from __future__ import annotations
|
||
from PWF.Convention.Runtime.Architecture import *
|
||
from PWF.Convention.Runtime.GlobalConfig import ConsoleFrontColor, ProjectConfig
|
||
|
||
from Plugins.WPSAPI import GuideEntry, WPSAPI
|
||
from Plugins.WPSConfigSystem import WPSConfigAPI
|
||
|
||
from .WPSRedPacketService import RedPacket, RedPacketType, WPSRedPacketService
|
||
|
||
logger: ProjectConfig = Architecture.Get(ProjectConfig)
|
||
|
||
|
||
class WPSRedPacketBase(WPSAPI):
|
||
"""红包插件公共基类，封装参数解析与输出模板。"""
|
||
@override
|
||
def wake_up(self) -> None:
|
||
pass
|
||
|
||
def dependencies(self) -> List[type]:
|
||
return [WPSAPI, WPSConfigAPI, WPSRedPacketService]
|
||
|
||
def is_enable_plugin(self) -> bool:
|
||
return True
|
||
|
||
def collect_command_entries(self) -> Sequence[GuideEntry]:
|
||
return ()
|
||
|
||
# region Helpers
|
||
|
||
def red_packet_service(self) -> WPSRedPacketService:
|
||
return Architecture.Get(WPSRedPacketService)
|
||
|
||
def config_api(self) -> WPSConfigAPI:
|
||
return Architecture.Get(WPSConfigAPI)
|
||
|
||
def parse_amount_and_slots(
|
||
self,
|
||
tokens: Sequence[str],
|
||
*,
|
||
require_slots: bool = True,
|
||
) -> Tuple[int, int, List[str]]:
|
||
service = self.red_packet_service()
|
||
remaining_tokens = list(tokens)
|
||
amount = service.get_default_amount()
|
||
slots = service.get_default_slots()
|
||
|
||
if remaining_tokens:
|
||
amount_token = remaining_tokens.pop(0)
|
||
if amount_token.isdigit():
|
||
amount = int(amount_token)
|
||
else:
|
||
remaining_tokens.insert(0, amount_token)
|
||
|
||
if require_slots and remaining_tokens:
|
||
slot_token = remaining_tokens.pop(0)
|
||
if slot_token.isdigit():
|
||
slots = int(slot_token)
|
||
else:
|
||
remaining_tokens.insert(0, slot_token)
|
||
elif not require_slots:
|
||
slots = 1
|
||
|
||
return amount, slots, remaining_tokens
|
||
|
||
def validate_amount_and_slots(self, amount: int, slots: int) -> None:
|
||
if amount <= 0:
|
||
raise ValueError("红包金额必须大于0")
|
||
if slots <= 0:
|
||
raise ValueError("红包份数必须大于0")
|
||
if amount < slots:
|
||
raise ValueError("红包金额不能小于份数")
|
||
|
||
def ensure_sufficient_points(self, user_id: int, amount: int) -> None:
|
||
points = self.config_api().get_user_points(user_id)
|
||
if points < amount:
|
||
raise ValueError("积分不足，无法发放红包")
|
||
|
||
def format_packet_message(self, packet: RedPacket, extra_hint: str | None = None) -> str:
|
||
type_label = {
|
||
RedPacketType.RANDOM: "手气红包",
|
||
RedPacketType.PASSWORD: "口令红包",
|
||
RedPacketType.EXCLUSIVE: "专属红包",
|
||
}.get(packet.packet_type, packet.packet_type.value)
|
||
lines = [
|
||
"# 🎁 红包已发出",
|
||
f"- 红包ID：`{packet.packet_id}`",
|
||
f"- 类型：{type_label}",
|
||
f"- 总额：{packet.total_amount} 分",
|
||
f"- 份数：{packet.total_slots}",
|
||
f"- 指令：`抢红包 {packet.packet_id}`",
|
||
]
|
||
if packet.packet_type == RedPacketType.EXCLUSIVE and packet.exclusive_username:
|
||
lines.append(f"- 专属对象：{packet.exclusive_username}")
|
||
if extra_hint:
|
||
lines.append(extra_hint)
|
||
return "\n".join(lines)
|
||
|
||
def format_error(self, message: str) -> str:
|
||
logger.Log("Warning", f"{ConsoleFrontColor.YELLOW}{message}{ConsoleFrontColor.RESET}")
|
||
return f"❌ {message}"
|
||
|
||
# endregion
|
||
|
||
|
||
__all__ = ["WPSRedPacketBase"]
|
||
|