"""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 @override def get_webhook_url(self, message: str, chat_id: int, user_id: int) -> str: config : WPSConfigAPI = Architecture.Get(WPSConfigAPI) url = config.get_user_url(user_id) if url: return url else: return super().get_webhook_url(message, chat_id, user_id) 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"] theft_users = len(json.loads(plot["theft_users"])) if plot.get("theft_users") else 0 trap_info = "" if show_trap: trap_item_id = plot.get("trap_item_id") if trap_item_id and isinstance(trap_item_id, str) and trap_item_id.strip(): trap_durability = int(plot.get("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}次)" 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"]