from __future__ import annotations from typing import List, Sequence, Tuple from PWF.Convention.Runtime.Architecture import Architecture 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): """红包插件公共基类,封装参数解析与输出模板。""" 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"]