新增查看炼金配方的功能
This commit is contained in:
@@ -1,8 +1,9 @@
|
||||
from __future__ import annotations
|
||||
|
||||
import random
|
||||
from collections import defaultdict, Counter
|
||||
from dataclasses import dataclass
|
||||
from typing import Dict, List, Optional, Sequence, Tuple, override
|
||||
from typing import Dict, List, Optional, Sequence, Set, Tuple, override
|
||||
|
||||
from PWF.Convention.Runtime.Architecture import Architecture
|
||||
from PWF.Convention.Runtime.GlobalConfig import ConsoleFrontColor, ProjectConfig
|
||||
@@ -55,6 +56,9 @@ class WPSAlchemyGame(WPSAPI):
|
||||
def __init__(self) -> None:
|
||||
super().__init__()
|
||||
self._recipes: Dict[Tuple[str, str, str], AlchemyRecipe] = {}
|
||||
self._material_index: Dict[str, Set[Tuple[str, str, str]]] = defaultdict(set)
|
||||
self._success_index: Dict[str, Set[Tuple[str, str, str]]] = defaultdict(set)
|
||||
self._fail_index: Dict[str, Set[Tuple[str, str, str]]] = defaultdict(set)
|
||||
self._fortune_coeff = FORTUNE_COEFF
|
||||
|
||||
@override
|
||||
@@ -138,18 +142,81 @@ class WPSAlchemyGame(WPSAPI):
|
||||
clamped_rate = clamp01(base_success_rate)
|
||||
if clamped_rate != base_success_rate:
|
||||
raise ValueError("配方成功率必须位于 0~1 之间")
|
||||
sorted_key = tuple(sorted_materials)
|
||||
existing = self._recipes.get(sorted_key)
|
||||
if existing:
|
||||
self._unindex_recipe(sorted_key, existing)
|
||||
recipe = AlchemyRecipe(
|
||||
materials=sorted_materials,
|
||||
materials=sorted_key,
|
||||
success_item_id=success_item_id.strip(),
|
||||
fail_item_id=fail_item_id.strip() or self.ASH_ITEM_ID,
|
||||
base_success_rate=base_success_rate,
|
||||
)
|
||||
self._recipes[sorted_materials] = recipe
|
||||
self._recipes[sorted_key] = recipe
|
||||
self._index_recipe(sorted_key, recipe)
|
||||
logger.Log(
|
||||
"Info",
|
||||
f"{ConsoleFrontColor.CYAN}已注册炼金配方 {sorted_materials} -> {success_item_id} ({base_success_rate:.2%}){ConsoleFrontColor.RESET}",
|
||||
)
|
||||
|
||||
def _index_recipe(
|
||||
self, materials_key: Tuple[str, str, str], recipe: AlchemyRecipe
|
||||
) -> None:
|
||||
for material in recipe.materials:
|
||||
self._material_index[material].add(materials_key)
|
||||
self._success_index[recipe.success_item_id].add(materials_key)
|
||||
self._fail_index[recipe.fail_item_id].add(materials_key)
|
||||
|
||||
def _unindex_recipe(
|
||||
self, materials_key: Tuple[str, str, str], recipe: AlchemyRecipe
|
||||
) -> None:
|
||||
for material in recipe.materials:
|
||||
material_set = self._material_index.get(material)
|
||||
if material_set and materials_key in material_set:
|
||||
material_set.discard(materials_key)
|
||||
if not material_set:
|
||||
del self._material_index[material]
|
||||
success_set = self._success_index.get(recipe.success_item_id)
|
||||
if success_set and materials_key in success_set:
|
||||
success_set.discard(materials_key)
|
||||
if not success_set:
|
||||
del self._success_index[recipe.success_item_id]
|
||||
fail_set = self._fail_index.get(recipe.fail_item_id)
|
||||
if fail_set and materials_key in fail_set:
|
||||
fail_set.discard(materials_key)
|
||||
if not fail_set:
|
||||
del self._fail_index[recipe.fail_item_id]
|
||||
|
||||
def get_recipes_using_item(self, item_id: str) -> List[AlchemyRecipe]:
|
||||
if not item_id:
|
||||
return []
|
||||
material_keys = sorted(self._material_index.get(item_id, set()))
|
||||
return [self._recipes[key] for key in material_keys]
|
||||
|
||||
def get_recipes_producing_item(
|
||||
self, item_id: str
|
||||
) -> Dict[str, List[AlchemyRecipe]]:
|
||||
if not item_id:
|
||||
return {"success": [], "fail": []}
|
||||
success_keys = sorted(
|
||||
self._success_index.get(item_id, set()),
|
||||
key=lambda key: (
|
||||
-self._recipes[key].base_success_rate,
|
||||
self._recipes[key].materials,
|
||||
),
|
||||
)
|
||||
fail_keys = sorted(
|
||||
self._fail_index.get(item_id, set()),
|
||||
key=lambda key: (
|
||||
-self._recipes[key].base_success_rate,
|
||||
self._recipes[key].materials,
|
||||
),
|
||||
)
|
||||
return {
|
||||
"success": [self._recipes[key] for key in success_keys],
|
||||
"fail": [self._recipes[key] for key in fail_keys],
|
||||
}
|
||||
|
||||
async def callback(
|
||||
self, message: str, chat_id: int, user_id: int
|
||||
) -> Optional[str]:
|
||||
@@ -379,5 +446,176 @@ class WPSAlchemyGame(WPSAPI):
|
||||
)
|
||||
|
||||
|
||||
__all__ = ["WPSAlchemyGame"]
|
||||
|
||||
class WPSAlchemyRecipeLookup(WPSAPI):
|
||||
def __init__(self) -> None:
|
||||
super().__init__()
|
||||
self._alchemy: Optional[WPSAlchemyGame] = None
|
||||
self._backpack: Optional[WPSBackpackSystem] = None
|
||||
|
||||
def dependencies(self) -> List[type]:
|
||||
return [WPSAlchemyGame, WPSBackpackSystem]
|
||||
|
||||
def is_enable_plugin(self) -> bool:
|
||||
return True
|
||||
|
||||
def wake_up(self) -> None:
|
||||
self._alchemy = Architecture.Get(WPSAlchemyGame)
|
||||
self._backpack = Architecture.Get(WPSBackpackSystem)
|
||||
self.register_plugin("炼金配方")
|
||||
self.register_plugin("alchemy_recipe")
|
||||
logger.Log(
|
||||
"Info",
|
||||
f"{ConsoleFrontColor.GREEN}WPSAlchemyRecipeLookup 插件已加载{ConsoleFrontColor.RESET}",
|
||||
)
|
||||
|
||||
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(self._help_text(), chat_id, user_id)
|
||||
|
||||
backpack = self._backpack or Architecture.Get(WPSBackpackSystem)
|
||||
definition = self._resolve_definition(payload, backpack)
|
||||
if definition is None:
|
||||
return await self.send_markdown_message(
|
||||
f"❌ 未找到物品 `{payload}`，请确认输入的物品 ID 或名称。",
|
||||
chat_id,
|
||||
user_id,
|
||||
)
|
||||
|
||||
alchemy = self._alchemy or Architecture.Get(WPSAlchemyGame)
|
||||
material_recipes = alchemy.get_recipes_using_item(definition.item_id)
|
||||
produce_map = alchemy.get_recipes_producing_item(definition.item_id)
|
||||
success_recipes = produce_map["success"]
|
||||
fail_recipes = produce_map["fail"]
|
||||
|
||||
message_text = self._format_markdown(
|
||||
definition,
|
||||
material_recipes,
|
||||
success_recipes,
|
||||
fail_recipes,
|
||||
backpack,
|
||||
)
|
||||
return await self.send_markdown_message(message_text, chat_id, user_id)
|
||||
|
||||
def _resolve_definition(
|
||||
self, identifier: str, backpack: WPSBackpackSystem
|
||||
) -> Optional[BackpackItemDefinition]:
|
||||
lowered = identifier.strip().lower()
|
||||
cursor = get_db().conn.cursor()
|
||||
cursor.execute(
|
||||
f"""
|
||||
SELECT item_id
|
||||
FROM {WPSBackpackSystem.ITEMS_TABLE}
|
||||
WHERE lower(item_id) = ? OR lower(name) = ?
|
||||
LIMIT 1
|
||||
""",
|
||||
(lowered, lowered),
|
||||
)
|
||||
row = cursor.fetchone()
|
||||
item_id = row["item_id"] if row else identifier.strip()
|
||||
try:
|
||||
return backpack._get_definition(item_id) # noqa: SLF001
|
||||
except Exception:
|
||||
return None
|
||||
|
||||
def _format_markdown(
|
||||
self,
|
||||
target: BackpackItemDefinition,
|
||||
material_recipes: List[AlchemyRecipe],
|
||||
success_recipes: List[AlchemyRecipe],
|
||||
fail_recipes: List[AlchemyRecipe],
|
||||
backpack: WPSBackpackSystem,
|
||||
) -> str:
|
||||
lines = [
|
||||
f"# 🔍 炼金配方查询｜{target.name}",
|
||||
f"- 物品 ID：`{target.item_id}`",
|
||||
"---",
|
||||
]
|
||||
lines.append("## 作为配方材料")
|
||||
lines.extend(
|
||||
self._format_recipe_entries(material_recipes, backpack)
|
||||
or ["- 暂无记录"]
|
||||
)
|
||||
|
||||
lines.append("\n## 作为成功产物")
|
||||
lines.extend(
|
||||
self._format_recipe_entries(success_recipes, backpack, role="success")
|
||||
or ["- 暂无记录"]
|
||||
)
|
||||
|
||||
lines.append("\n## 作为失败产物")
|
||||
lines.extend(
|
||||
self._format_recipe_entries(fail_recipes, backpack, role="fail")
|
||||
or ["- 暂无记录"]
|
||||
)
|
||||
return "\n".join(lines)
|
||||
|
||||
def _format_recipe_entries(
|
||||
self,
|
||||
recipes: List[AlchemyRecipe],
|
||||
backpack: WPSBackpackSystem,
|
||||
*,
|
||||
role: str = "material",
|
||||
) -> List[str]:
|
||||
if not recipes:
|
||||
return []
|
||||
entries: List[str] = []
|
||||
for recipe in recipes:
|
||||
materials = self._summarize_materials(recipe, backpack)
|
||||
success_name = self._resolve_item_name(recipe.success_item_id, backpack)
|
||||
fail_name = self._resolve_item_name(recipe.fail_item_id, backpack)
|
||||
rate = f"{recipe.base_success_rate:.0%}"
|
||||
if role == "material":
|
||||
entry = (
|
||||
f"- 材料：{materials}｜成功产物：`{success_name}`｜"
|
||||
f"失败产物：`{fail_name}`｜成功率：{rate}"
|
||||
)
|
||||
elif role == "success":
|
||||
entry = (
|
||||
f"- 材料：{materials}｜成功率：{rate}｜"
|
||||
f"失败产物：`{fail_name}`"
|
||||
)
|
||||
else:
|
||||
entry = (
|
||||
f"- 材料：{materials}｜成功率：{rate}｜"
|
||||
f"成功产物：`{success_name}`"
|
||||
)
|
||||
entries.append(entry)
|
||||
return entries
|
||||
|
||||
def _summarize_materials(
|
||||
self, recipe: AlchemyRecipe, backpack: WPSBackpackSystem
|
||||
) -> str:
|
||||
counter = Counter(recipe.materials)
|
||||
parts: List[str] = []
|
||||
for item_id, count in sorted(counter.items()):
|
||||
name = self._resolve_item_name(item_id, backpack)
|
||||
if count == 1:
|
||||
parts.append(f"`{name}`")
|
||||
else:
|
||||
parts.append(f"`{name}` × {count}")
|
||||
return " + ".join(parts)
|
||||
|
||||
def _resolve_item_name(
|
||||
self, item_id: str, backpack: WPSBackpackSystem
|
||||
) -> str:
|
||||
try:
|
||||
definition = backpack._get_definition(item_id) # noqa: SLF001
|
||||
return definition.name
|
||||
except Exception:
|
||||
return item_id
|
||||
|
||||
def _help_text(self) -> str:
|
||||
return (
|
||||
"# ⚗️ 炼金配方查询帮助\n"
|
||||
"- `炼金配方 <物品ID>`\n"
|
||||
"- `炼金配方 <物品名称>`\n"
|
||||
"> 输入需要精确匹配注册物品，名称不区分大小写。"
|
||||
)
|
||||
|
||||
|
||||
__all__ = ["WPSAlchemyGame", "WPSAlchemyRecipeLookup"]
|
||||
|
||||
|
||||
Reference in New Issue
Block a user