1.新增一些配方2.新增红包系统3.WPSConfigSystem中新增用户名替换空白字符机制

.tasks/2025-11-17_1_red-envelope-system.md

@@ -0,0 +1,40 @@
+# 背景
+文件名：2025-11-17_1
+创建于：2025-11-17_19:43:26
+创建者：ASUS
+主分支：main
+任务分支：未创建
+Yolo模式：Off
+
+# 任务描述
+新增一个红包系统，可以赠送积分，并发起一些小游戏，包括猜谜红包、手气红包、专属红包、口令红包。
+
+# 项目概览
+WPS Bot 插件化系统，已存在战斗、炼金、花园等插件，需要在既有积分/背包体系下扩展红包玩法。
+
+# 分析
+已查阅 `Plugins/WPSAPI.py` 与 `PWF/CoreModules/plugin_interface.py`：所有插件继承 `WPSAPI`，通过 `register_plugin` 绑定命令，消息发送统一走 `send_markdown_message`，依赖通过 `Architecture.Get` 获取，数据库操作直接使用 `get_db().conn`。
+积分记录位于 `user_info` 表，由 `WPSConfigAPI` 暴露的 `adjust_user_points/adjust_user_points_sync` 读写；签到、商店、战斗、菜园等模块都以此接口为唯一积分入口，理由是保证同一逻辑下的数据一致性。
+背包、商店、战斗、菜园等复杂插件都采用“service+plugin”拆分与独立 `DatabaseModel` 定义；红包系统若要支持多种玩法，需要类似的 service 层管理红包生命周期、并通过 API 输出图鉴信息和命令帮助。
+
+需求补充：
+- 指令入口拆分：`红包|手气红包 <金额> <人数>`（缺省金额/人数分别由 `ProjectConfig` 读取，默认 100 / 4）、`口令红包 <金额> <人数> <口令>`、`专属红包 <金额> <user_id|用户名>`，统一通过 `抢红包 <红包ID> [tokens]` 领取，口令红包要求 tokens 与口令一致。猜谜红包暂缓实现。
+- 金额不足直接拒绝发出；红包不限时间，可并行存在且发送者也能参与抢夺；人数不设上限但需合理处理默认值与输入校验。
+- 红包发出后必须在返回消息中附带红包ID，供玩家在 `抢红包` 指令中引用；需要记录红包状态、剩余份额/金额、领取日志，以及区分不同玩法的附加约束（专属白名单、口令验证等）。
+
+# 提议的解决方案
+（待补充：完成调研后填写多个可选实现思路及其优缺点。）
+
+# 当前执行步骤："1. 研究需求与代码结构"
+
+# 任务进度
+2025-11-17_20:29:15
+- 已修改：`Plugins/WPSRedPacketService.py` `Plugins/WPSRedPacketBase.py` `Plugins/WPSRandomRedPacket.py` `Plugins/WPSPasswordRedPacket.py` `Plugins/WPSExclusiveRedPacket.py` `Plugins/WPSRedPacketClaim.py`
+- 更改：实现红包核心服务、三类发包插件与抢红包入口，完成数据库表定义、指令注册与积分结算逻辑。
+- 原因：支持手气/口令/专属红包以及统一的领取流程。
+- 阻碍因素：无
+- 状态：未确认
+
+# 最终审查
+
+

Plugins/WPSCombatSystem/combat_plugin_base.py

@@ -389,7 +389,7 @@ class WPSCombatBase(WPSAPI):
         # 5.1. 注册冒险种子到菜园系统
         self._register_adventure_seeds_to_garden()
 
-        self._register_legendary_alchemy_recipes()
+        self._register_alchemy_recipes()
 
         # 6. 恢复过期任务和超时战斗
         try:
@@ -606,31 +606,38 @@ class WPSCombatBase(WPSAPI):
                 f"{ConsoleFrontColor.YELLOW}注册冒险种子到菜园系统时出错: {e}{ConsoleFrontColor.RESET}"
             )
 
-    def _register_legendary_alchemy_recipes(self) -> None:
+    def _register_alchemy_recipes(self) -> None:
-        """注册传说装备的炼金链条"""
+        """注册炼金配方"""
         alchemy: WPSAlchemyGame = Architecture.Get(WPSAlchemyGame)
 
         recipe_definitions = (
+            # 物品
+            (("alchemy_slag", "alchemy_slag", "alchemy_slag"), "combat_material_ore", "alchemy_slag", 0.34),
+            (("combat_potion_hp_small","combat_potion_hp_small","combat_potion_hp_small"), "combat_potion_hp_medium", "alchemy_slag", 0.75),
+            (("combat_potion_hp_medium","combat_potion_hp_medium","combat_potion_hp_medium"), "combat_potion_hp_large", "alchemy_slag", 0.75),
+            (("combat_material_ore","combat_material_ore","combat_material_gem"),"combat_material_crystal", "alchemy_ash", 0.75),
+            (("combat_material_gem","combat_material_gem","combat_material_crystal"),"combat_material_essence","alchemy_ash",0.75),
+            (("combat_material_crystal","combat_potion_atk","combat_potion_def"),"combat_material_essence","alchemy_ash",0.7),
             # 护甲链
-            (("combat_material_ore", "garden_wood_maple", "combat_armor_chain"), "combat_armor_plate", 0.70),
+            (("combat_material_ore", "garden_wood_maple", "combat_armor_chain"), "combat_armor_plate", SPARK_DUST_ITEM_ID, 0.70),
-            (("combat_material_gem", "combat_material_crystal", "combat_armor_plate"), "combat_armor_sentinel", 0.50),
+            (("combat_material_gem", "combat_material_crystal", "combat_armor_plate"), "combat_armor_sentinel", SPARK_DUST_ITEM_ID, 0.50),
-            (("combat_armor_sentinel", "garden_wood_sakura", "combat_material_crystal"), "combat_armor_dragonheart", 0.30),
+            (("combat_armor_sentinel", "garden_wood_sakura", "combat_material_crystal"), "combat_armor_dragonheart", SPARK_DUST_ITEM_ID, 0.30),
-            (("combat_armor_dragonheart", "combat_material_essence", "combat_armor_plate"), "combat_armor_guardian", 0.10),
+            (("combat_armor_dragonheart", "combat_material_essence", "combat_armor_plate"), "combat_armor_guardian", SPARK_DUST_ITEM_ID, 0.10),
             # 鞋子链
-            (("combat_material_ore", "garden_wood_ginkgo", "combat_boots_leather"), "combat_boots_rapid", 0.70),
+            (("combat_material_ore", "garden_wood_ginkgo", "combat_boots_leather"), "combat_boots_rapid", SPARK_DUST_ITEM_ID, 0.70),
-            (("combat_boots_rapid", "combat_material_gem", "combat_material_crystal"), "combat_boots_wind", 0.50),
+            (("combat_boots_rapid", "combat_material_gem", "combat_material_crystal"), "combat_boots_wind", SPARK_DUST_ITEM_ID, 0.50),
-            (("combat_boots_wind", "combat_material_crystal", "garden_wood_ginkgo"), "combat_boots_tempest", 0.30),
+            (("combat_boots_wind", "combat_material_crystal", "garden_wood_ginkgo"), "combat_boots_tempest", SPARK_DUST_ITEM_ID, 0.30),
-            (("combat_boots_tempest", "combat_material_essence", "garden_wine_maple"), "combat_boots_starlight", 0.10),
+            (("combat_boots_tempest", "combat_material_essence", "garden_wine_maple"), "combat_boots_starlight", SPARK_DUST_ITEM_ID, 0.10),
             # 饰品链
-            (("combat_accessory_ring_str", "combat_material_gem", "garden_wood_sakura"), "combat_accessory_barrier", 0.70),
+            (("combat_accessory_ring_str", "combat_material_gem", "garden_wood_sakura"), "combat_accessory_barrier", SPARK_DUST_ITEM_ID, 0.70),
-            (("combat_accessory_barrier", "combat_material_crystal", "garden_wood_sakura"), "combat_accessory_amulet", 0.50),
+            (("combat_accessory_barrier", "combat_material_crystal", "garden_wood_sakura"), "combat_accessory_amulet", SPARK_DUST_ITEM_ID, 0.50),
-            (("combat_accessory_amulet", "combat_material_crystal", "garden_wine_sakura"), "combat_accessory_sanctum", 0.30),
+            (("combat_accessory_amulet", "combat_material_crystal", "garden_wine_sakura"), "combat_accessory_sanctum", SPARK_DUST_ITEM_ID, 0.30),
-            (("combat_accessory_sanctum", "combat_material_essence", "combat_souvenir_relic"), "combat_accessory_aegis", 0.10),
+            (("combat_accessory_sanctum", "combat_material_essence", "combat_souvenir_relic"), "combat_accessory_aegis", SPARK_DUST_ITEM_ID, 0.10),
         )
 
-        for materials, success_item_id, success_rate in recipe_definitions:
+        for materials, success_item_id, success_rate, fail_item_id in recipe_definitions:
             try:
-                alchemy.register_recipe(materials, success_item_id, SPARK_DUST_ITEM_ID, success_rate)
+                alchemy.register_recipe(materials, success_item_id, fail_item_id, success_rate)
             except Exception as exc:
                 logger.Log(
                     "Warning",

Plugins/WPSConfigSystem.py

@@ -4,6 +4,7 @@ from PWF.Convention.Runtime.Config import *
 from PWF.Convention.Runtime.Architecture import Architecture
 from PWF.Convention.Runtime.GlobalConfig import ProjectConfig
 from datetime import datetime
+import re
 
 from PWF.CoreModules.plugin_interface import DatabaseModel, get_db
 
@@ -71,6 +72,7 @@ class WPSConfigAPI(WPSAPI):
 
     @override
     def wake_up(self) -> None:
+        self._sanitize_usernames()
         logger.Log("Info", f"{ConsoleFrontColor.GREEN}WPSConfigAPI 插件已加载{ConsoleFrontColor.RESET}")
         self.register_plugin("config")
         self.register_plugin("cfg")
@@ -203,6 +205,36 @@ class WPSConfigAPI(WPSAPI):
         value = record.get("username")
         return str(value) if value else f"user_{user_id}"
 
+    def _sanitize_usernames(self) -> None:
+        cursor = get_db().conn.cursor()
+        try:
+            cursor.execute("SELECT user_id, username FROM user_info")
+            rows = cursor.fetchall()
+        except Exception as exc:
+            logger.Log("Error", f"查询 user_info 失败: {exc}")
+            return
+
+        updated = 0
+        for row in rows:
+            user_id = row["user_id"]
+            username = row["username"] or ""
+            if not username or not re.search(r"\s", username):
+                continue
+            sanitized = re.sub(r"\s+", "_", username.strip())
+            if sanitized == username:
+                continue
+            try:
+                cursor.execute(
+                    "UPDATE user_info SET username = ? WHERE user_id = ?",
+                    (sanitized, user_id),
+                )
+                updated += 1
+            except Exception as exc:
+                logger.Log("Error", f"更新 user_id={user_id} 的用户名失败: {exc}")
+        if updated:
+            get_db().conn.commit()
+            logger.Log("Info", f"已替换 {updated} 条含空白字符的用户名")
+
     def find_user_id_by_username(self, username: str) -> Optional[int]:
         text = (username or "").strip()
         if not text:

Plugins/WPSExclusiveRedPacket.py

@@ -0,0 +1,86 @@
+from __future__ import annotations
+
+from typing import Optional
+
+from PWF.Convention.Runtime.Architecture import Architecture
+
+from Plugins.WPSAPI import GuideEntry
+from Plugins.WPSConfigSystem import WPSConfigAPI
+
+from .WPSRedPacketBase import WPSRedPacketBase
+
+
+class WPSExclusiveRedPacket(WPSRedPacketBase):
+    """专属红包入口插件"""
+
+    def get_guide_subtitle(self) -> str:
+        return "发送给指定用户的专属红包"
+
+    def collect_command_entries(self):
+        return (
+            GuideEntry(
+                title="专属红包",
+                identifier="专属红包 <金额> <user_id|用户名>",
+                description="只允许指定用户领取，该用户也需要使用 `抢红包` 指令领取。",
+            ),
+        )
+
+    def wake_up(self) -> None:
+        super().wake_up()
+        self.register_plugin("专属红包")
+
+    async def callback(self, message: str, chat_id: int, user_id: int) -> Optional[str]:
+        payload = self.parse_message_after_at(message).strip()
+        tokens = [token.strip() for token in payload.split() if token.strip()]
+        if len(tokens) < 2:
+            return await self.send_markdown_message(
+                self.format_error("指令格式：`专属红包 <金额> <用户ID|用户名>`"),
+                chat_id,
+                user_id,
+            )
+
+        service = self.red_packet_service()
+        amount_token = tokens[0]
+        target_token = tokens[1:]
+        try:
+            if not amount_token.isdigit():
+                raise ValueError("金额需要是正整数")
+            amount = int(amount_token)
+            target_identifier = " ".join(target_token).strip()
+            if not target_identifier:
+                raise ValueError("需要指定目标用户")
+            target_user_id = self._resolve_target_user(target_identifier)
+            if target_user_id is None:
+                raise ValueError("未找到目标用户")
+            target_username = self.config_api().get_user_name(target_user_id)
+
+            self.validate_amount_and_slots(amount, 1)
+            self.ensure_sufficient_points(user_id, amount)
+            packet = service.create_exclusive_packet(
+                chat_id,
+                user_id,
+                amount,
+                target_user_id,
+                target_username,
+            )
+            hint = f"- 专属对象：{target_username}（仅该用户可抢）"
+            return await self.send_markdown_message(
+                self.format_packet_message(packet, hint),
+                chat_id,
+                user_id,
+            )
+        except ValueError as exc:
+            return await self.send_markdown_message(self.format_error(str(exc)), chat_id, user_id)
+
+    def _resolve_target_user(self, identifier: str) -> Optional[int]:
+        text = identifier.strip()
+        if not text:
+            return None
+        if text.isdigit():
+            return int(text)
+        config_api: WPSConfigAPI = Architecture.Get(WPSConfigAPI)
+        return config_api.find_user_id_by_username(text)
+
+
+__all__ = ["WPSExclusiveRedPacket"]
+

Plugins/WPSPasswordRedPacket.py

@@ -0,0 +1,55 @@
+from __future__ import annotations
+
+from typing import Optional
+
+from Plugins.WPSAPI import GuideEntry
+
+from .WPSRedPacketBase import WPSRedPacketBase
+
+
+class WPSPasswordRedPacket(WPSRedPacketBase):
+    """口令红包入口插件"""
+
+    def get_guide_subtitle(self) -> str:
+        return "发布口令红包，需要准确口令才能领取"
+
+    def collect_command_entries(self):
+        return (
+            GuideEntry(
+                title="口令红包",
+                identifier="口令红包 <金额> <人数> <口令>",
+                description="设置好口令后，领取者需在抢红包时附带同样的口令。",
+            ),
+        )
+
+    def wake_up(self) -> None:
+        super().wake_up()
+        self.register_plugin("口令红包")
+
+    async def callback(self, message: str, chat_id: int, user_id: int) -> Optional[str]:
+        payload = self.parse_message_after_at(message).strip()
+        tokens = [token.strip() for token in payload.split() if token.strip()]
+        service = self.red_packet_service()
+        try:
+            amount, slots, remaining = self.parse_amount_and_slots(tokens)
+            if not remaining:
+                raise ValueError("需要提供口令参数")
+            password = " ".join(remaining)
+            self.validate_amount_and_slots(amount, slots)
+            self.ensure_sufficient_points(user_id, amount)
+            packet = service.create_password_packet(chat_id, user_id, amount, slots, password)
+            hint = (
+                f"- 提示：抢红包时使用 ``抢红包 {packet.packet_id} {password}`` "
+                "（口令必须完全一致）。"
+            )
+            return await self.send_markdown_message(
+                self.format_packet_message(packet, hint),
+                chat_id,
+                user_id,
+            )
+        except ValueError as exc:
+            return await self.send_markdown_message(self.format_error(str(exc)), chat_id, user_id)
+
+
+__all__ = ["WPSPasswordRedPacket"]
+

Plugins/WPSRandomRedPacket.py

@@ -0,0 +1,51 @@
+from __future__ import annotations
+
+from typing import Optional
+
+from Plugins.WPSAPI import GuideEntry
+
+from .WPSRedPacketBase import WPSRedPacketBase
+
+
+class WPSRandomRedPacket(WPSRedPacketBase):
+    """手气红包入口插件"""
+
+    def get_guide_subtitle(self) -> str:
+        return "发起手气红包，随机拆分积分给指定人数"
+
+    def collect_command_entries(self):
+        return (
+            GuideEntry(
+                title="红包",
+                identifier="红包 <金额> <人数>",
+                description="按照随机手气拆分积分，缺省金额/人数使用配置项。",
+                metadata={"别名": "手气红包"},
+            ),
+        )
+
+    def wake_up(self) -> None:
+        super().wake_up()
+        self.register_plugin("红包")
+        self.register_plugin("手气红包")
+
+    async def callback(self, message: str, chat_id: int, user_id: int) -> Optional[str]:
+        payload = self.parse_message_after_at(message).strip()
+        tokens = [token.strip() for token in payload.split() if token.strip()]
+        service = self.red_packet_service()
+        try:
+            amount, slots, _ = self.parse_amount_and_slots(tokens)
+            self.validate_amount_and_slots(amount, slots)
+            self.ensure_sufficient_points(user_id, amount)
+            packet = service.create_random_packet(chat_id, user_id, amount, slots)
+            hint = f"- 提示：使用 `抢红包 {packet.packet_id}` 抢夺份额。"
+            return await self.send_markdown_message(
+                self.format_packet_message(packet, hint),
+                chat_id,
+                user_id,
+            )
+        except ValueError as exc:
+            return await self.send_markdown_message(self.format_error(str(exc)), chat_id, user_id)
+
+
+__all__ = ["WPSRandomRedPacket"]
+

Plugins/WPSRedPacketBase.py

@@ -0,0 +1,106 @@
+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"]
+

Plugins/WPSRedPacketClaim.py

@@ -0,0 +1,56 @@
+from __future__ import annotations
+
+from typing import Optional, Sequence
+
+from Plugins.WPSAPI import GuideEntry
+
+from .WPSRedPacketBase import WPSRedPacketBase
+
+
+class WPSRedPacketClaim(WPSRedPacketBase):
+    """抢红包入口"""
+
+    def get_guide_subtitle(self) -> str:
+        return "通过红包ID领取积分奖励"
+
+    def collect_command_entries(self):
+        return (
+            GuideEntry(
+                title="抢红包",
+                identifier="抢红包 <红包ID> [口令]",
+                description="输入红包ID后领取剩余份额，口令红包需附带口令。",
+            ),
+        )
+
+    def wake_up(self) -> None:
+        super().wake_up()
+        self.register_plugin("抢红包")
+
+    async def callback(self, message: str, chat_id: int, user_id: int) -> Optional[str]:
+        payload = self.parse_message_after_at(message).strip()
+        tokens = [token.strip() for token in payload.split() if token.strip()]
+        if not tokens:
+            return await self.send_markdown_message(
+                self.format_error("指令格式：`抢红包 <红包ID> [口令]`"),
+                chat_id,
+                user_id,
+            )
+
+        packet_id = tokens[0]
+        extra_tokens: Sequence[str] = tokens[1:]
+        username = self.config_api().get_user_name(user_id)
+        service = self.red_packet_service()
+        success, message_text, _ = service.claim_packet(
+            chat_id,
+            packet_id,
+            user_id,
+            username,
+            extra_tokens,
+        )
+        if success:
+            return await self.send_markdown_message(message_text, chat_id, user_id)
+        return await self.send_markdown_message(message_text, chat_id, user_id)
+
+
+__all__ = ["WPSRedPacketClaim"]
+

Plugins/WPSRedPacketService.py

@@ -0,0 +1,405 @@
+from __future__ import annotations
+
+import random
+from dataclasses import dataclass
+from datetime import datetime, timezone
+from enum import Enum
+from typing import List, Optional, Sequence, Tuple
+from uuid import uuid4
+
+from PWF.Convention.Runtime.Architecture import Architecture
+from PWF.Convention.Runtime.GlobalConfig import ConsoleFrontColor, ProjectConfig
+from PWF.CoreModules.database import get_db
+from PWF.CoreModules.plugin_interface import DatabaseModel
+
+from Plugins.WPSAPI import GuideEntry, WPSAPI
+from Plugins.WPSConfigSystem import WPSConfigAPI
+
+logger: ProjectConfig = Architecture.Get(ProjectConfig)
+
+
+class RedPacketType(str, Enum):
+    RANDOM = "random"
+    PASSWORD = "password"
+    EXCLUSIVE = "exclusive"
+
+
+@dataclass(frozen=True)
+class RedPacket:
+    packet_id: str
+    owner_id: int
+    packet_type: RedPacketType
+    total_amount: int
+    total_slots: int
+    remaining_amount: int
+    remaining_slots: int
+    exclusive_user_id: Optional[int] = None
+    exclusive_username: Optional[str] = None
+    password: Optional[str] = None
+    created_at: str = ""
+
+
+@dataclass(frozen=True)
+class RedPacketClaim:
+    packet_id: str
+    user_id: int
+    username: str
+    amount: int
+    claimed_at: str
+
+
+class WPSRedPacketService(WPSAPI):
+    """红包核心服务，负责数据存储与积分结算。"""
+
+    PACKET_TABLE = "red_packets"
+    CLAIM_TABLE = "red_packet_claims"
+
+    def __init__(self) -> None:
+        super().__init__()
+        self._default_amount: int = int(logger.FindItem("red_packet_default_amount", 100))
+        self._default_slots: int = int(logger.FindItem("red_packet_default_slots", 4))
+        logger.SaveProperties()
+
+    def get_guide_subtitle(self) -> str:
+        return "统一管理红包生命周期与积分结算的核心服务"
+
+    def collect_command_entries(self) -> Sequence[GuideEntry]:
+        return ()
+
+    def collect_guide_entries(self) -> Sequence[GuideEntry]:
+        return (
+            {
+                "title": "默认配置",
+                "description": f"默认金额 {self._default_amount} 分，默认份数 {self._default_slots} 份，可通过 ProjectConfig 覆盖。",
+            },
+            {
+                "title": "积分结算",
+                "description": "发送红包时一次性扣除积分，领取时再发放给领取者，红包不支持退款或撤销。",
+            },
+        )
+
+    def dependencies(self) -> List[type]:
+        return [WPSAPI, WPSConfigAPI]
+
+    def is_enable_plugin(self) -> bool:
+        return True
+
+    def register_db_model(self):
+        return [
+            DatabaseModel(
+                table_name=self.PACKET_TABLE,
+                column_defs={
+                    "packet_id": "TEXT PRIMARY KEY",
+                    "owner_id": "INTEGER NOT NULL",
+                    "packet_type": "TEXT NOT NULL",
+                    "total_amount": "INTEGER NOT NULL",
+                    "total_slots": "INTEGER NOT NULL",
+                    "remaining_amount": "INTEGER NOT NULL",
+                    "remaining_slots": "INTEGER NOT NULL",
+                    "exclusive_user_id": "INTEGER",
+                    "exclusive_username": "TEXT",
+                    "password": "TEXT",
+                    "created_at": "TEXT NOT NULL",
+                },
+            ),
+            DatabaseModel(
+                table_name=self.CLAIM_TABLE,
+                column_defs={
+                    "claim_id": "INTEGER PRIMARY KEY AUTOINCREMENT",
+                    "packet_id": "TEXT NOT NULL",
+                    "user_id": "INTEGER NOT NULL",
+                    "username": "TEXT",
+                    "amount": "INTEGER NOT NULL",
+                    "claimed_at": "TEXT NOT NULL",
+                },
+            ),
+        ]
+
+    def wake_up(self) -> None:
+        logger.Log(
+            "Info",
+            f"{ConsoleFrontColor.GREEN}WPSRedPacketService 插件已加载{ConsoleFrontColor.RESET}",
+        )
+
+    # region 配置 / 公共接口
+
+    def get_default_amount(self) -> int:
+        return max(1, self._default_amount)
+
+    def get_default_slots(self) -> int:
+        return max(1, self._default_slots)
+
+    def generate_packet_id(self) -> str:
+        return uuid4().hex[:12]
+
+    def _config_api(self) -> WPSConfigAPI:
+        return Architecture.Get(WPSConfigAPI)
+
+    def _now(self) -> str:
+        return datetime.now(timezone.utc).isoformat()
+
+    # endregion
+
+    # region 发红包
+
+    def create_random_packet(
+        self,
+        chat_id: int,
+        owner_id: int,
+        total_amount: int,
+        total_slots: int,
+    ) -> RedPacket:
+        return self._create_packet(
+            chat_id=chat_id,
+            owner_id=owner_id,
+            total_amount=total_amount,
+            total_slots=total_slots,
+            packet_type=RedPacketType.RANDOM,
+        )
+
+    def create_password_packet(
+        self,
+        chat_id: int,
+        owner_id: int,
+        total_amount: int,
+        total_slots: int,
+        password: str,
+    ) -> RedPacket:
+        if not password:
+            raise ValueError("口令不能为空")
+        return self._create_packet(
+            chat_id=chat_id,
+            owner_id=owner_id,
+            total_amount=total_amount,
+            total_slots=total_slots,
+            packet_type=RedPacketType.PASSWORD,
+            password=password,
+        )
+
+    def create_exclusive_packet(
+        self,
+        chat_id: int,
+        owner_id: int,
+        total_amount: int,
+        target_user_id: int,
+        target_username: str,
+    ) -> RedPacket:
+        if target_user_id <= 0:
+            raise ValueError("目标用户ID无效")
+        return self._create_packet(
+            chat_id=chat_id,
+            owner_id=owner_id,
+            total_amount=total_amount,
+            total_slots=1,
+            packet_type=RedPacketType.EXCLUSIVE,
+            exclusive_user_id=target_user_id,
+            exclusive_username=target_username or f"user_{target_user_id}",
+        )
+
+    def _create_packet(
+        self,
+        *,
+        chat_id: int,
+        owner_id: int,
+        total_amount: int,
+        total_slots: int,
+        packet_type: RedPacketType,
+        password: Optional[str] = None,
+        exclusive_user_id: Optional[int] = None,
+        exclusive_username: Optional[str] = None,
+    ) -> RedPacket:
+        if total_amount <= 0:
+            raise ValueError("红包金额必须大于0")
+        if total_slots <= 0:
+            raise ValueError("红包份数必须大于0")
+        if total_amount < total_slots:
+            raise ValueError("红包金额不能小于份数")
+
+        self._deduct_points(owner_id, total_amount, packet_type, chat_id)
+
+        packet_id = self.generate_packet_id()
+        created_at = self._now()
+
+        cursor = get_db().conn.cursor()
+        cursor.execute(
+            f"""
+            INSERT INTO {self.PACKET_TABLE} (
+                packet_id, owner_id, packet_type,
+                total_amount, total_slots,
+                remaining_amount, remaining_slots,
+                exclusive_user_id, exclusive_username,
+                password, created_at
+            ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
+            """,
+            (
+                packet_id,
+                owner_id,
+                packet_type.value,
+                total_amount,
+                total_slots,
+                total_amount,
+                total_slots,
+                exclusive_user_id,
+                exclusive_username,
+                password,
+                created_at,
+            ),
+        )
+        get_db().conn.commit()
+        return RedPacket(
+            packet_id=packet_id,
+            owner_id=owner_id,
+            packet_type=packet_type,
+            total_amount=total_amount,
+            total_slots=total_slots,
+            remaining_amount=total_amount,
+            remaining_slots=total_slots,
+            exclusive_user_id=exclusive_user_id,
+            exclusive_username=exclusive_username,
+            password=password,
+            created_at=created_at,
+        )
+
+    def _deduct_points(
+        self,
+        owner_id: int,
+        total_amount: int,
+        packet_type: RedPacketType,
+        chat_id: int,
+    ) -> None:
+        config_api = self._config_api()
+        current_points = config_api.get_user_points(owner_id)
+        if current_points < total_amount:
+            raise ValueError("积分不足，无法发放红包")
+        config_api.adjust_user_points_sync(
+            owner_id,
+            -total_amount,
+            reason=f"发放{packet_type.value}红包（会话{chat_id}）",
+        )
+
+    # endregion
+
+    # region 抢红包
+
+    def claim_packet(
+        self,
+        chat_id: int,
+        packet_id: str,
+        user_id: int,
+        username: str,
+        tokens: Optional[Sequence[str]] = None,
+    ) -> Tuple[bool, str, Optional[int]]:
+        if not packet_id:
+            return False, "❌ 红包ID不能为空", None
+        conn = get_db().conn
+        cursor = conn.cursor()
+        try:
+            cursor.execute("BEGIN IMMEDIATE")
+            cursor.execute(
+                f"""
+                SELECT packet_id, owner_id, packet_type, total_amount, total_slots,
+                       remaining_amount, remaining_slots, exclusive_user_id,
+                       exclusive_username, password
+                FROM {self.PACKET_TABLE}
+                WHERE packet_id = ?
+                """,
+                (packet_id,),
+            )
+            row = cursor.fetchone()
+            if not row:
+                conn.rollback()
+                return False, "❌ 红包不存在或已失效", None
+
+            packet = self._row_to_packet(row)
+
+            if packet.remaining_slots <= 0 or packet.remaining_amount <= 0:
+                conn.rollback()
+                return False, "⚠️ 红包已经被抢完了", None
+
+            cursor.execute(
+                f"""
+                SELECT 1 FROM {self.CLAIM_TABLE}
+                WHERE packet_id = ? AND user_id = ?
+                """,
+                (packet_id, user_id),
+            )
+            if cursor.fetchone():
+                conn.rollback()
+                return False, "⚠️ 你已经抢过这个红包了", None
+
+            if packet.packet_type == RedPacketType.EXCLUSIVE:
+                if packet.exclusive_user_id != user_id:
+                    conn.rollback()
+                    return False, "❌ 这是专属红包，只有目标用户可以领取", None
+
+            if packet.packet_type == RedPacketType.PASSWORD:
+                provided = " ".join(tokens or []).strip()
+                if not provided:
+                    conn.rollback()
+                    return False, "❌ 这是口令红包，需要附带口令", None
+                if provided != (packet.password or ""):
+                    conn.rollback()
+                    return False, "❌ 口令不正确", None
+
+            grant_amount = self._allocate_amount(packet.remaining_amount, packet.remaining_slots)
+
+            cursor.execute(
+                f"""
+                UPDATE {self.PACKET_TABLE}
+                SET remaining_amount = remaining_amount - ?,
+                    remaining_slots = remaining_slots - 1
+                WHERE packet_id = ?
+                """,
+                (grant_amount, packet_id),
+            )
+            cursor.execute(
+                f"""
+                INSERT INTO {self.CLAIM_TABLE} (packet_id, user_id, username, amount, claimed_at)
+                VALUES (?, ?, ?, ?, ?)
+                """,
+                (packet_id, user_id, username, grant_amount, self._now()),
+            )
+            conn.commit()
+        except Exception as exc:
+            conn.rollback()
+            logger.Log("Error", f"{ConsoleFrontColor.RED}领取红包失败: {exc}{ConsoleFrontColor.RESET}")
+            return False, "❌ 领取红包时出现错误，请稍后再试", None
+
+        config_api = self._config_api()
+        config_api.adjust_user_points_sync(
+            user_id,
+            grant_amount,
+            reason=f"抢到红包 {packet_id}（会话{chat_id}）",
+        )
+
+        return True, f"✅ 领取成功，获得 {grant_amount} 分！", grant_amount
+
+    def _allocate_amount(self, remaining_amount: int, remaining_slots: int) -> int:
+        if remaining_slots <= 1:
+            return remaining_amount
+        min_per_slot = 1
+        max_available = remaining_amount - (remaining_slots - 1) * min_per_slot
+        avg = remaining_amount // remaining_slots
+        upper = max(min_per_slot, min(max_available, max(min_per_slot, avg * 2)))
+        return random.randint(min_per_slot, upper)
+
+    def _row_to_packet(self, row) -> RedPacket:
+        return RedPacket(
+            packet_id=row["packet_id"],
+            owner_id=row["owner_id"],
+            packet_type=RedPacketType(row["packet_type"]),
+            total_amount=row["total_amount"],
+            total_slots=row["total_slots"],
+            remaining_amount=row["remaining_amount"],
+            remaining_slots=row["remaining_slots"],
+            exclusive_user_id=row["exclusive_user_id"],
+            exclusive_username=row["exclusive_username"],
+            password=row["password"],
+            created_at="",
+        )
+
+    # endregion
+
+
+__all__ = ["WPSRedPacketService", "RedPacketType", "RedPacket", "RedPacketClaim"]
+