564 lines
21 KiB
Python
564 lines
21 KiB
Python
"""Shared base class for garden plugins."""
|
||
|
||
from __future__ import annotations
|
||
|
||
import json
|
||
from typing import Any, Dict, List, Optional, Sequence, Tuple, Type, Union, override
|
||
|
||
from PWF.Convention.Runtime.Architecture import Architecture
|
||
from PWF.Convention.Runtime.GlobalConfig import ConsoleFrontColor, ProjectConfig
|
||
from PWF.CoreModules.plugin_interface import DatabaseModel
|
||
|
||
from Plugins.WPSAPI import GuideEntry, GuideSection, WPSAPI
|
||
from Plugins.WPSBackpackSystem import (
|
||
BackpackItemTier,
|
||
WPSBackpackSystem,
|
||
)
|
||
from Plugins.WPSStoreSystem import WPSStoreSystem
|
||
from Plugins.WPSConfigSystem import WPSConfigAPI
|
||
from Plugins.WPSFortuneSystem import WPSFortuneSystem
|
||
from Plugins.WPSAlchemyGame import WPSAlchemyGame
|
||
|
||
from .garden_models import (
|
||
GARDEN_CROPS,
|
||
GARDEN_FRUITS,
|
||
GARDEN_MISC_ITEMS,
|
||
GARDEN_TRAPS,
|
||
GardenCropDefinition,
|
||
get_garden_db_models,
|
||
)
|
||
from .garden_service import GardenService
|
||
|
||
|
||
class WPSGardenBase(WPSAPI):
|
||
_service: GardenService | None = None
|
||
_initialized: bool = False
|
||
|
||
def get_guide_subtitle(self) -> str:
|
||
return "菜园作物种植与联动系统的核心服务"
|
||
|
||
def get_guide_metadata(self) -> Dict[str, str]:
|
||
service = self.service()
|
||
config = service.config
|
||
return {
|
||
"作物数量": str(len(GARDEN_CROPS)),
|
||
"最大地块": str(config.max_plots),
|
||
"出售倍率": str(config.sale_multiplier),
|
||
}
|
||
|
||
def collect_item_entries(self) -> Sequence[GuideEntry]:
|
||
tier_counter: Dict[str, int] = {}
|
||
wine_counter: int = 0
|
||
for crop in GARDEN_CROPS.values():
|
||
tier_counter[crop.tier] = tier_counter.get(crop.tier, 0) + 1
|
||
if crop.wine_item_id:
|
||
wine_counter += 1
|
||
entries: List[GuideEntry] = []
|
||
for tier, count in sorted(tier_counter.items()):
|
||
entries.append(
|
||
{
|
||
"title": f"{tier.title()} 作物",
|
||
"description": f"{count} 种作物，可收获果实与额外奖励。",
|
||
}
|
||
)
|
||
entries.append(
|
||
{
|
||
"title": "果酒配方",
|
||
"description": f"{wine_counter} 种作物支持果酒配方，并与战斗系统的果酒增益联动。",
|
||
}
|
||
)
|
||
if GARDEN_MISC_ITEMS:
|
||
entries.append(
|
||
{
|
||
"title": "杂项素材",
|
||
"description": f"{len(GARDEN_MISC_ITEMS)} 种额外素材，可用于任务或商店出售。",
|
||
}
|
||
)
|
||
return tuple(entries)
|
||
|
||
def collect_guide_entries(self) -> Sequence[GuideEntry]:
|
||
service = self.service()
|
||
return (
|
||
{
|
||
"title": "成长流程",
|
||
"description": (
|
||
"种植后根据作物 `growth_minutes` 决定成熟时间，系统会在成熟时通过时钟任务提醒。"
|
||
),
|
||
},
|
||
{
|
||
"title": "收获收益",
|
||
"description": (
|
||
"收获基础产量由 `base_yield` 决定，额外奖励受运势与 `extra_reward` 配置影响。"
|
||
),
|
||
},
|
||
{
|
||
"title": "果实售出",
|
||
"description": (
|
||
f"通过 `菜园 售出` 指令以 {service.config.sale_multiplier} 倍种子价格出售果实并获取积分。"
|
||
),
|
||
},
|
||
)
|
||
|
||
def collect_additional_sections(self) -> Sequence[GuideSection]:
|
||
sections = list(super().collect_additional_sections())
|
||
|
||
crop_entries: List[GuideEntry] = []
|
||
tier_icon = {
|
||
"common": "🌱",
|
||
"rare": "🌳",
|
||
"epic": "🍷",
|
||
"legendary": "🏵️",
|
||
}
|
||
for crop in GARDEN_CROPS.values():
|
||
reward_desc = ""
|
||
if crop.extra_reward.kind == "points":
|
||
payload = crop.extra_reward.payload
|
||
reward_desc = (
|
||
f"额外积分 {payload.get('min', 0)}~{payload.get('max', 0)}，"
|
||
f"触发率 {crop.extra_reward.base_rate*100:.0f}%"
|
||
)
|
||
elif crop.extra_reward.kind == "item":
|
||
payload = crop.extra_reward.payload
|
||
reward_desc = (
|
||
f"额外物品 `{crop.extra_item_id}` 数量 {payload.get('min', 0)}~{payload.get('max', 0)}，"
|
||
f"触发率 {crop.extra_reward.base_rate*100:.0f}%"
|
||
)
|
||
details: List[Union[str, Dict[str, Any]]] = [
|
||
{
|
||
"type": "list",
|
||
"items": [
|
||
f"生长时间：{crop.growth_minutes} 分钟",
|
||
f"基础产量：{crop.base_yield} 个果实",
|
||
reward_desc or "无额外奖励",
|
||
f"种子售价：{crop.seed_price} 分",
|
||
],
|
||
}
|
||
]
|
||
if crop.wine_item_id:
|
||
details.append(
|
||
{
|
||
"type": "list",
|
||
"items": [
|
||
f"果酒：{crop.wine_item_id}（稀有度 {crop.wine_tier or 'rare'}）",
|
||
"炼金配方：三份果实 + 炼金坩埚 → 果酒。",
|
||
],
|
||
}
|
||
)
|
||
crop_entries.append(
|
||
GuideEntry(
|
||
title=crop.display_name,
|
||
identifier=crop.seed_id,
|
||
description=f"果实 ID：{crop.fruit_id}",
|
||
category="作物",
|
||
metadata={
|
||
"稀有度": crop.tier,
|
||
"果实ID": crop.fruit_id,
|
||
},
|
||
icon=tier_icon.get(crop.tier.lower(), "🌿"),
|
||
tags=(crop.tier.title(),),
|
||
details=details,
|
||
)
|
||
)
|
||
if crop_entries:
|
||
sections.append(
|
||
GuideSection(
|
||
title="作物图鉴",
|
||
entries=crop_entries,
|
||
layout="grid",
|
||
section_id="garden-crops",
|
||
description="每种作物的成长周期、产出与额外奖励说明。",
|
||
)
|
||
)
|
||
|
||
if GARDEN_MISC_ITEMS:
|
||
misc_entries: List[GuideEntry] = []
|
||
for item_id, meta in GARDEN_MISC_ITEMS.items():
|
||
misc_entries.append(
|
||
GuideEntry(
|
||
title=meta.get("name", item_id),
|
||
identifier=item_id,
|
||
description=meta.get("description", ""),
|
||
category="杂项素材",
|
||
icon="🧺",
|
||
)
|
||
)
|
||
sections.append(
|
||
GuideSection(
|
||
title="杂项素材",
|
||
entries=misc_entries,
|
||
layout="grid",
|
||
section_id="garden-misc",
|
||
description="园艺相关的任务或合成所需的特殊素材。",
|
||
)
|
||
)
|
||
|
||
return tuple(sections)
|
||
|
||
@classmethod
|
||
def service(cls) -> GardenService:
|
||
if cls._service is None:
|
||
cls._service = GardenService()
|
||
cls._service.recover_overdue_plots()
|
||
return cls._service
|
||
|
||
def dependencies(self) -> List[Type]:
|
||
return [
|
||
WPSConfigAPI,
|
||
WPSBackpackSystem,
|
||
WPSStoreSystem,
|
||
WPSFortuneSystem,
|
||
WPSAlchemyGame,
|
||
]
|
||
|
||
def register_db_model(self) -> List[DatabaseModel]:
|
||
return get_garden_db_models()
|
||
|
||
def wake_up(self) -> None:
|
||
if WPSGardenBase._initialized:
|
||
return
|
||
WPSGardenBase._initialized = True
|
||
logger: ProjectConfig = Architecture.Get(ProjectConfig)
|
||
backpack: WPSBackpackSystem = Architecture.Get(WPSBackpackSystem)
|
||
store: WPSStoreSystem = Architecture.Get(WPSStoreSystem)
|
||
alchemy: WPSAlchemyGame = Architecture.Get(WPSAlchemyGame)
|
||
service = self.service()
|
||
|
||
for crop in GARDEN_CROPS.values():
|
||
seed_name = f"{crop.display_name}的种子"
|
||
fruit_name = f"{crop.display_name}的果实"
|
||
# 支持所有tier等级：common, rare, epic, legendary
|
||
tier_map = {
|
||
"common": BackpackItemTier.COMMON,
|
||
"rare": BackpackItemTier.RARE,
|
||
"epic": BackpackItemTier.EPIC,
|
||
"legendary": BackpackItemTier.LEGENDARY,
|
||
}
|
||
tier = tier_map.get(crop.tier.lower(), BackpackItemTier.RARE)
|
||
seed_desc = f"{crop.display_name}的种子，可在菜园种植获取相应作物。"
|
||
fruit_desc = f"{crop.display_name}成熟后的果实，可食用或售出换取积分。"
|
||
self._safe_register_item(backpack, crop.seed_id, seed_name, tier, seed_desc)
|
||
self._safe_register_item(backpack, crop.fruit_id, fruit_name, tier, fruit_desc)
|
||
if crop.extra_reward and crop.extra_reward.kind == "item" and crop.extra_item_id:
|
||
wood_name = f"{crop.display_name}的木材"
|
||
wood_desc = f"{crop.display_name}加工所得的木材，可用于特定任务或制作。"
|
||
self._safe_register_item(
|
||
backpack,
|
||
crop.extra_item_id,
|
||
wood_name,
|
||
BackpackItemTier.RARE,
|
||
wood_desc,
|
||
)
|
||
if crop.wine_item_id and crop.wine_tier:
|
||
wine_tier = getattr(BackpackItemTier, crop.wine_tier.upper(), BackpackItemTier.RARE)
|
||
wine_name = f"{crop.display_name}的果酒"
|
||
wine_desc = self._generate_wine_description(crop.display_name, crop.wine_item_id)
|
||
self._safe_register_item(
|
||
backpack,
|
||
crop.wine_item_id,
|
||
wine_name,
|
||
wine_tier,
|
||
wine_desc,
|
||
)
|
||
|
||
self._safe_register_mode(
|
||
store,
|
||
crop,
|
||
limit_amount=service.config.seed_store_limit,
|
||
)
|
||
if crop.wine_item_id and crop.wine_tier:
|
||
wine_price = crop.seed_price * service.config.sale_multiplier * 5
|
||
self._safe_register_wine_mode(
|
||
store,
|
||
crop.wine_item_id,
|
||
wine_price,
|
||
limit_amount=service.config.seed_store_limit,
|
||
)
|
||
self._safe_register_recipe(alchemy, crop)
|
||
|
||
for item_id, meta in GARDEN_MISC_ITEMS.items():
|
||
self._safe_register_item(
|
||
backpack,
|
||
item_id,
|
||
meta["name"],
|
||
BackpackItemTier.COMMON,
|
||
meta.get("description", ""),
|
||
)
|
||
|
||
# 注册陷阱物品
|
||
for trap in GARDEN_TRAPS:
|
||
trap_tier = tier_map.get(trap.tier.lower(), BackpackItemTier.RARE)
|
||
self._safe_register_item(
|
||
backpack,
|
||
trap.item_id,
|
||
trap.display_name,
|
||
trap_tier,
|
||
trap.description,
|
||
)
|
||
|
||
# 注册陷阱的炼金合成配方
|
||
# 获取稀有树木的木材ID
|
||
ginkgo_crop = GARDEN_CROPS.get("garden_seed_ginkgo")
|
||
sakura_crop = GARDEN_CROPS.get("garden_seed_sakura")
|
||
maple_crop = GARDEN_CROPS.get("garden_seed_maple")
|
||
|
||
# 防盗网：矿石 + 银杏木材 + 樱花木材
|
||
if ginkgo_crop and sakura_crop and ginkgo_crop.extra_item_id and sakura_crop.extra_item_id:
|
||
self._safe_register_trap_recipe(
|
||
alchemy,
|
||
("combat_material_ore", ginkgo_crop.extra_item_id, sakura_crop.extra_item_id),
|
||
"garden_trap_net",
|
||
0.75, # 75%成功率
|
||
)
|
||
|
||
# 荆棘陷阱：宝石 + 红枫木材 + 银杏木材
|
||
if maple_crop and ginkgo_crop and maple_crop.extra_item_id and ginkgo_crop.extra_item_id:
|
||
self._safe_register_trap_recipe(
|
||
alchemy,
|
||
("combat_material_gem", maple_crop.extra_item_id, ginkgo_crop.extra_item_id),
|
||
"garden_trap_thorn",
|
||
0.70, # 70%成功率
|
||
)
|
||
|
||
# 魔法结界：水晶 + 两种木材
|
||
if ginkgo_crop and sakura_crop and ginkgo_crop.extra_item_id and sakura_crop.extra_item_id:
|
||
self._safe_register_trap_recipe(
|
||
alchemy,
|
||
("combat_material_crystal", ginkgo_crop.extra_item_id, sakura_crop.extra_item_id),
|
||
"garden_trap_magic",
|
||
0.65, # 65%成功率
|
||
)
|
||
|
||
# 传奇守护：精华 + 水晶 + 红枫木材
|
||
if maple_crop and maple_crop.extra_item_id:
|
||
self._safe_register_trap_recipe(
|
||
alchemy,
|
||
("combat_material_essence", "combat_material_crystal", maple_crop.extra_item_id),
|
||
"garden_trap_legend",
|
||
0.60, # 60%成功率
|
||
)
|
||
|
||
logger.Log(
|
||
"Info",
|
||
f"{ConsoleFrontColor.GREEN}WPSGarden 系统完成物品与商店初始化{ConsoleFrontColor.RESET}",
|
||
)
|
||
|
||
# region Helpers
|
||
def _safe_register_item(
|
||
self,
|
||
backpack: WPSBackpackSystem,
|
||
item_id: str,
|
||
name: str,
|
||
tier: BackpackItemTier,
|
||
description: str,
|
||
) -> None:
|
||
try:
|
||
backpack.register_item(item_id, name, tier, description)
|
||
except Exception:
|
||
pass
|
||
|
||
def _safe_register_mode(
|
||
self,
|
||
store: WPSStoreSystem,
|
||
crop: GardenCropDefinition,
|
||
*,
|
||
limit_amount: int,
|
||
) -> None:
|
||
try:
|
||
store.register_mode(
|
||
item_id=crop.seed_id,
|
||
price=crop.seed_price,
|
||
limit_amount=limit_amount,
|
||
)
|
||
except Exception:
|
||
pass
|
||
|
||
def _safe_register_wine_mode(
|
||
self,
|
||
store: WPSStoreSystem,
|
||
item_id: str,
|
||
price: int,
|
||
*,
|
||
limit_amount: int,
|
||
) -> None:
|
||
try:
|
||
store.register_mode(
|
||
item_id=item_id,
|
||
price=price,
|
||
limit_amount=limit_amount,
|
||
)
|
||
except Exception:
|
||
pass
|
||
|
||
def _generate_wine_description(self, crop_name: str, wine_item_id: str) -> str:
|
||
"""生成包含buff加成信息的果酒描述"""
|
||
# 尝试导入战斗系统的WINE_BUFFS（可选依赖）
|
||
try:
|
||
from Plugins.WPSCombatSystem.combat_models import WINE_BUFFS
|
||
buffs = WINE_BUFFS.get(wine_item_id, {})
|
||
except ImportError:
|
||
buffs = {}
|
||
|
||
parts = [f"{crop_name}酿制的果酒，饮用后可触发战斗增益。"]
|
||
|
||
if buffs:
|
||
buff_parts = []
|
||
buff_names = {
|
||
"time_reduction": "冒险时间",
|
||
"reward_boost": "冒险收益",
|
||
"success_rate": "冒险成功率",
|
||
"atk_boost": "攻击力",
|
||
"def_boost": "防御力",
|
||
"crit_boost": "暴击率",
|
||
}
|
||
for buff_key, buff_value in sorted(buffs.items()):
|
||
buff_name = buff_names.get(buff_key, buff_key)
|
||
if buff_key == "time_reduction":
|
||
buff_parts.append(f"{buff_name}-{buff_value*100:.0f}%")
|
||
else:
|
||
buff_parts.append(f"{buff_name}+{buff_value*100:.0f}%")
|
||
if buff_parts:
|
||
parts.append(f"效果：{', '.join(buff_parts)}")
|
||
|
||
return " | ".join(parts)
|
||
|
||
def _safe_register_recipe(
|
||
self,
|
||
alchemy: WPSAlchemyGame,
|
||
crop: GardenCropDefinition,
|
||
) -> None:
|
||
try:
|
||
if not crop.wine_item_id:
|
||
return
|
||
success_rate = 0.75
|
||
alchemy.register_recipe(
|
||
(crop.fruit_id, crop.fruit_id, crop.fruit_id),
|
||
crop.wine_item_id,
|
||
"garden_item_rot_fruit",
|
||
success_rate,
|
||
)
|
||
except Exception:
|
||
pass
|
||
|
||
def _safe_register_trap_recipe(
|
||
self,
|
||
alchemy: WPSAlchemyGame,
|
||
materials: Tuple[str, str, str],
|
||
result_item_id: str,
|
||
success_rate: float,
|
||
) -> None:
|
||
"""注册陷阱的炼金合成配方"""
|
||
try:
|
||
alchemy.register_recipe(
|
||
materials,
|
||
result_item_id,
|
||
"alchemy_ash", # 失败产物是炉灰
|
||
success_rate,
|
||
)
|
||
except Exception:
|
||
pass
|
||
|
||
async def _clock_mark_mature(self, user_id: int, chat_id: int, plot_index: int) -> None:
|
||
service = self.service()
|
||
plot = service.get_plot(user_id, plot_index)
|
||
if not plot:
|
||
return
|
||
if int(plot["is_mature"]) == 1:
|
||
return
|
||
service.mark_mature(user_id, plot_index)
|
||
crop = GARDEN_CROPS.get(plot["seed_id"])
|
||
if crop is None:
|
||
return
|
||
message = (
|
||
"# 🌾 作物成熟提醒\n"
|
||
f"- 地块 {plot_index}：{crop.display_name} 已成熟，记得收获！"
|
||
)
|
||
await self.send_markdown_message(message, chat_id, user_id)
|
||
|
||
def _format_timestamp(self, ts: str) -> str:
|
||
return self.service().format_display_time(ts)
|
||
|
||
def resolve_seed_id(self, keyword: str) -> Optional[GardenCropDefinition]:
|
||
key = keyword.strip().lower()
|
||
for crop in GARDEN_CROPS.values():
|
||
if crop.seed_id.lower() == key:
|
||
return crop
|
||
if crop.display_name.lower() == key:
|
||
return crop
|
||
if f"{crop.display_name}的种子".lower() == key:
|
||
return crop
|
||
return None
|
||
|
||
def resolve_fruit_id(self, keyword: str) -> Optional[GardenCropDefinition]:
|
||
key = keyword.strip().lower()
|
||
for crop in GARDEN_FRUITS.values():
|
||
if crop.fruit_id.lower() == key:
|
||
return crop
|
||
if crop.display_name.lower() == key:
|
||
return crop
|
||
if f"{crop.display_name}的果实".lower() == key:
|
||
return crop
|
||
return None
|
||
|
||
def format_garden_overview(self, user_id: int, show_trap: bool = True) -> str:
|
||
service = self.service()
|
||
plots = service.list_plots(user_id)
|
||
config = service.config
|
||
lines = ["# 🌱 菜园概览"]
|
||
if not plots:
|
||
lines.append("> 尚未种植任何作物，使用 `种植 <种子>` 开始耕种。")
|
||
else:
|
||
for plot in plots:
|
||
crop = GARDEN_CROPS.get(plot["seed_id"], None)
|
||
name = crop.display_name if crop else plot["seed_id"]
|
||
idx = plot["plot_index"]
|
||
is_mature = bool(plot["is_mature"])
|
||
mature_at = plot["mature_at"]
|
||
formatted_time = self._format_timestamp(mature_at)
|
||
if is_mature:
|
||
remaining = plot["remaining_fruit"]
|
||
try:
|
||
theft_users_str = plot["theft_users"]
|
||
theft_users = len(json.loads(theft_users_str)) if theft_users_str else 0
|
||
except (KeyError, TypeError, json.JSONDecodeError):
|
||
theft_users = 0
|
||
|
||
trap_info = ""
|
||
if show_trap:
|
||
try:
|
||
trap_item_id = plot["trap_item_id"]
|
||
if trap_item_id and isinstance(trap_item_id, str) and trap_item_id.strip():
|
||
try:
|
||
trap_durability = int(plot["trap_durability"]) if plot["trap_durability"] is not None else 0
|
||
except (KeyError, TypeError, ValueError):
|
||
trap_durability = 0
|
||
from .garden_models import GARDEN_TRAPS_DICT
|
||
trap = GARDEN_TRAPS_DICT.get(trap_item_id)
|
||
if trap:
|
||
trap_info = f"｜陷阱：{trap.display_name}({trap_durability}次)"
|
||
except (KeyError, TypeError):
|
||
# 字段不存在或值为None，不显示陷阱信息
|
||
pass
|
||
status = f"✅ 已成熟（成熟于 {formatted_time}）"
|
||
lines.append(
|
||
f"- 地块 {idx}｜{name}｜{status}｜剩余果实 {remaining}｜被偷次数 {theft_users}{trap_info}"
|
||
)
|
||
else:
|
||
status = f"⌛ 生长中，预计成熟 {formatted_time}"
|
||
lines.append(f"- 地块 {idx}｜{name}｜{status}")
|
||
available = config.max_plots - len(plots)
|
||
if available > 0:
|
||
lines.append(f"\n> 尚有 {available} 块空地可用。")
|
||
lines.append(
|
||
"\n---\n- `种植 <种子>`：消耗种子种下作物\n"
|
||
"- `收获 <地块序号>`：收成成熟作物\n"
|
||
"- `偷取 <用户> <地块序号>`：从他人成熟作物中偷取果实\n"
|
||
"- `铲除 <地块序号>`：立即清空指定地块\n"
|
||
"- `菜园 售出 <果实> <数量>`：出售果实换取积分"
|
||
)
|
||
return "\n".join(lines)
|
||
|
||
# endregion
|
||
|
||
|
||
__all__ = ["WPSGardenBase"]
|