ninemine/NewWPSBot
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

新增查看指令

Browse Source
This commit is contained in:
LiuBai095
2025-11-10 22:30:16 +08:00
parent f5b03422e4
commit aba445f438
9 changed files with 240 additions and 39 deletions
Download Patch File Download Diff File Expand all files Collapse all files

145
Plugins/WPSBackpackSystem.py
View File

@@ -2,7 +2,7 @@ from __future__ import annotations
from dataclasses import dataclass
from enum import Enum
from typing import Dict, List, Optional, Tuple, override
from typing import Dict, List, Optional, override
from PWF.Convention.Runtime.Architecture import Architecture
from PWF.Convention.Runtime.GlobalConfig import ConsoleFrontColor, ProjectConfig
@@ -41,6 +41,7 @@ class BackpackItemDefinition:
item_id: str
name: str
tier: BackpackItemTier
description: str
@dataclass(frozen=True)
@@ -72,13 +73,14 @@ class WPSBackpackSystem(WPSAPI):
def register_db_model(self):
from PWF.CoreModules.plugin_interface import DatabaseModel
return [
models = [
DatabaseModel(
table_name=self.ITEMS_TABLE,
column_defs={
"item_id": "TEXT PRIMARY KEY",
"name": "TEXT NOT NULL",
"tier": "TEXT NOT NULL",
"description": "TEXT NOT NULL DEFAULT ''",
},
),
DatabaseModel(
@@ -91,9 +93,12 @@ class WPSBackpackSystem(WPSAPI):
},
),
]
return models
@override
def wake_up(self) -> None:
db = get_db()
db.define_column(self.ITEMS_TABLE, "description", "TEXT NOT NULL DEFAULT ''")
logger.Log(
"Info",
f"{ConsoleFrontColor.GREEN}WPSBackpackSystem 插件已加载{ConsoleFrontColor.RESET}",
@@ -105,15 +110,17 @@ class WPSBackpackSystem(WPSAPI):
def _warm_item_cache(self) -> None:
cursor = get_db().conn.cursor()
cursor.execute(
f"SELECT item_id, name, tier FROM {self.ITEMS_TABLE}"
f"SELECT item_id, name, tier, description FROM {self.ITEMS_TABLE}"
)
rows = cursor.fetchall()
for row in rows:
tier = BackpackItemTier.from_string(row["tier"])
description = row["description"] or ""
self._item_cache[row["item_id"]] = BackpackItemDefinition(
item_id=row["item_id"],
name=row["name"],
tier=tier,
description=description,
)
# region 对外接口
@@ -123,22 +130,29 @@ class WPSBackpackSystem(WPSAPI):
item_id: str,
name: str,
tier: BackpackItemTier,
description: str,
) -> None:
if not item_id or not name:
raise ValueError("item_id and name must be provided")
cursor = get_db().conn.cursor()
cursor.execute(
f"""
INSERT INTO {self.ITEMS_TABLE} (item_id, name, tier)
VALUES (?, ?, ?)
INSERT INTO {self.ITEMS_TABLE} (item_id, name, tier, description)
VALUES (?, ?, ?, ?)
ON CONFLICT(item_id) DO UPDATE
SET name = excluded.name,
tier = excluded.tier
tier = excluded.tier,
description = excluded.description
""",
(item_id, name, tier.name),
(item_id, name, tier.name, description or ""),
)
get_db().conn.commit()
self._item_cache[item_id] = BackpackItemDefinition(item_id, name, tier)
self._item_cache[item_id] = BackpackItemDefinition(
item_id=item_id,
name=name,
tier=tier,
description=description or "",
)
def add_item(
self,
@@ -191,7 +205,7 @@ class WPSBackpackSystem(WPSAPI):
cursor = get_db().conn.cursor()
cursor.execute(
f"""
SELECT ui.item_id, ui.quantity, i.name, i.tier
SELECT ui.item_id, ui.quantity, i.name, i.tier, i.description
FROM {self.USER_ITEMS_TABLE} ui
JOIN {self.ITEMS_TABLE} i ON ui.item_id = i.item_id
WHERE ui.user_id = ? AND ui.quantity > 0
@@ -209,6 +223,7 @@ class WPSBackpackSystem(WPSAPI):
item_id=row["item_id"],
name=row["name"],
tier=BackpackItemTier.from_string(row["tier"]),
description=row["description"] or "",
)
except ValueError:
continue
@@ -228,7 +243,7 @@ class WPSBackpackSystem(WPSAPI):
return self._item_cache[item_id]
cursor = get_db().conn.cursor()
cursor.execute(
f"SELECT item_id, name, tier FROM {self.ITEMS_TABLE} WHERE item_id = ?",
f"SELECT item_id, name, tier, description FROM {self.ITEMS_TABLE} WHERE item_id = ?",
(item_id,),
)
row = cursor.fetchone()
@@ -238,6 +253,7 @@ class WPSBackpackSystem(WPSAPI):
item_id=row["item_id"],
name=row["name"],
tier=BackpackItemTier.from_string(row["tier"]),
description=row["description"] or "",
)
self._item_cache[item_id] = definition
return definition
@@ -301,5 +317,112 @@ class WPSBackpackSystem(WPSAPI):
return None
__all__ = ["WPSBackpackSystem", "BackpackItemTier", "BackpackItemDefinition"]
class WPSItemDescription(WPSAPI):
@override
def dependencies(self) -> List[type]:
return [WPSBackpackSystem]
@override
def is_enable_plugin(self) -> bool:
return True
def __init__(self) -> None:
super().__init__()
from Plugins.WPSCombatSystem.combat_models import ADVENTURE_SOUVENIRS
from Plugins.WPSGardenSystem.garden_models import GARDEN_CROPS
from Plugins.WPSGardenSystem.garden_service import GardenConfig
self._backpack: Optional[WPSBackpackSystem] = None
self._souvenir_prices: Dict[str, int] = {
item_id: sell_price
for item_id, (_, __, sell_price, ___) in ADVENTURE_SOUVENIRS.items()
}
self._garden_crops = GARDEN_CROPS
self._garden_sale_multiplier = GardenConfig.load().sale_multiplier
@override
def wake_up(self) -> None:
self._backpack = Architecture.Get(WPSBackpackSystem)
self.register_plugin("查看")
self.register_plugin("view")
async def callback(self, message: str, chat_id: int, user_id: int) -> Optional[str]:
payload = self.parse_message_after_at(message).strip()
if not payload:
return await self.send_markdown_message("❌ 指令格式：`查看 <物品ID或名称>`", chat_id, user_id)
definition = self._resolve_definition(payload)
if not definition:
return await self.send_markdown_message("❌ 未找到对应物品，请确认输入是否正确", chat_id, user_id)
sale_info = self._get_sale_info(definition.item_id)
markdown = self._format_markdown(definition, sale_info)
return await self.send_markdown_message(markdown, chat_id, user_id)
def _resolve_definition(self, identifier: str) -> Optional[BackpackItemDefinition]:
lowered = identifier.strip().lower()
db = get_db().conn.cursor()
db.execute(
f"""
SELECT item_id
FROM {WPSBackpackSystem.ITEMS_TABLE}
WHERE lower(item_id) = ? OR lower(name) = ?
LIMIT 1
""",
(lowered, lowered),
)
row = db.fetchone()
item_id = row["item_id"] if row else identifier.strip()
backpack = self._backpack or Architecture.Get(WPSBackpackSystem)
try:
return backpack._get_definition(item_id)
except Exception:
return None
def _get_sale_info(self, item_id: str) -> Optional[Dict[str, str]]:
if item_id in self._souvenir_prices:
price = self._souvenir_prices[item_id]
return {
"category": "战斗纪念品",
"price": f"{price} 分/个",
"note": "在战斗系统中出售可立即换取积分。",
}
for crop in self._garden_crops.values():
if crop.fruit_id == item_id:
price = crop.seed_price * self._garden_sale_multiplier
return {
"category": "菜园果实",
"price": f"{price} 分/个",
"note": "可通过 `菜园 售出 <果实> <数量>` 换取积分。",
}
return None
def _format_markdown(
self,
definition: BackpackItemDefinition,
sale_info: Optional[Dict[str, str]],
) -> str:
tier_label = definition.tier.to_markdown_label(definition.tier.display_name)
lines = [
"# 🔍 物品详情",
f"- 名称：{definition.name}",
f"- ID：`{definition.item_id}`",
f"- 稀有度：{tier_label}",
f"- 描述：{definition.description or '（暂无描述）'}",
]
if sale_info:
lines.append(f"- 分类：{sale_info['category']}")
lines.append(f"- 系统售价：{sale_info['price']}")
note = sale_info.get("note")
if note:
lines.append(f"- 提示：{note}")
return "\n".join(lines)
__all__ = [
"WPSBackpackSystem",
"BackpackItemTier",
"BackpackItemDefinition",
"WPSItemDescription",
]