"""Garden system crop definitions and configuration models.""" from __future__ import annotations from typing import Dict, List, Tuple from PWF.Convention.Runtime.Config import * from PWF.Convention.Runtime.Architecture import Architecture from PWF.Convention.Runtime.GlobalConfig import ConsoleFrontColor, ProjectConfig from PWF.CoreModules.plugin_interface import DatabaseModel from pydantic import BaseModel, Field # 注意：延迟导入 GardenConfig 避免循环依赖 # Shared logger/config _config: ProjectConfig = Architecture.Get(ProjectConfig) _config.SaveProperties() class GardenExtraReward(BaseModel): kind: str = Field(..., description="points 或 item") payload: Dict[str, int] = Field(default_factory=dict) base_rate: float = Field(..., ge=0.0, le=1.0) class Config: allow_mutation = False class GardenCropDefinition(BaseModel): seed_id: str fruit_id: str display_name: str tier: str # common / rare growth_minutes: int seed_price: int base_yield: int extra_reward: GardenExtraReward extra_item_id: str | None = None wine_item_id: str | None = None wine_tier: str | None = None class Config: allow_mutation = False GARDEN_CONFIG_DEFAULTS: Dict[str, int | float | str] = { "garden_max_plots_per_user": 4, "garden_sale_multiplier": 10, "garden_fortune_coeff": 0.03, "garden_theft_threshold_ratio": 0.5, "garden_seed_store_limit": 5, "garden_crops_config_path": "Plugins/garden_crops.json", } for key, value in GARDEN_CONFIG_DEFAULTS.items(): _config.FindItem(key, value) _config.SaveProperties() COMMON_HERB_CROPS: Tuple[GardenCropDefinition, ...] = ( GardenCropDefinition( seed_id="garden_seed_mint", fruit_id="garden_fruit_mint", display_name="薄荷", tier="common", growth_minutes=30, seed_price=30, base_yield=4, extra_reward=GardenExtraReward(kind="points", payload={"min": 10, "max": 120}, base_rate=0.6), wine_item_id="garden_wine_mint", wine_tier="rare", ), GardenCropDefinition( seed_id="garden_seed_basil", fruit_id="garden_fruit_basil", display_name="罗勒", tier="common", growth_minutes=40, seed_price=36, base_yield=5, extra_reward=GardenExtraReward(kind="points", payload={"min": 15, "max": 150}, base_rate=0.55), wine_item_id="garden_wine_basil", wine_tier="rare", ), GardenCropDefinition( seed_id="garden_seed_sage", fruit_id="garden_fruit_sage", display_name="鼠尾草", tier="common", growth_minutes=50, seed_price=42, base_yield=5, extra_reward=GardenExtraReward(kind="points", payload={"min": 20, "max": 180}, base_rate=0.5), wine_item_id="garden_wine_sage", wine_tier="rare", ), GardenCropDefinition( seed_id="garden_seed_rosemary", fruit_id="garden_fruit_rosemary", display_name="迷迭香", tier="common", growth_minutes=60, seed_price=50, base_yield=6, extra_reward=GardenExtraReward(kind="points", payload={"min": 30, "max": 220}, base_rate=0.45), wine_item_id="garden_wine_rosemary", wine_tier="rare", ), ) RARE_TREE_CROPS: Tuple[GardenCropDefinition, ...] = ( GardenCropDefinition( seed_id="garden_seed_ginkgo", fruit_id="garden_fruit_ginkgo", display_name="银杏", tier="rare", growth_minutes=120, seed_price=120, base_yield=3, extra_reward=GardenExtraReward(kind="item", payload={"min": 2, "max": 6}, base_rate=0.5), extra_item_id="garden_wood_ginkgo", wine_item_id="garden_wine_ginkgo", wine_tier="epic", ), GardenCropDefinition( seed_id="garden_seed_sakura", fruit_id="garden_fruit_sakura", display_name="樱花", tier="rare", growth_minutes=160, seed_price=150, base_yield=3, extra_reward=GardenExtraReward(kind="item", payload={"min": 3, "max": 8}, base_rate=0.45), extra_item_id="garden_wood_sakura", wine_item_id="garden_wine_sakura", wine_tier="epic", ), GardenCropDefinition( seed_id="garden_seed_maple", fruit_id="garden_fruit_maple", display_name="红枫", tier="rare", growth_minutes=180, seed_price=180, base_yield=4, extra_reward=GardenExtraReward(kind="item", payload={"min": 4, "max": 10}, base_rate=0.4), extra_item_id="garden_wood_maple", wine_item_id="garden_wine_maple", wine_tier="epic", ), ) def load_crops_from_config(config_path: str) -> Dict[str, GardenCropDefinition]: """从配置文件加载作物定义 Args: config_path: 配置文件路径 Returns: 作物字典，seed_id -> GardenCropDefinition """ import json from PWF.Convention.Runtime.File import ToolFile try: config_file = ToolFile(config_path) if not config_file.Exists(): _config.Log("Warning", f"作物配置文件不存在: {config_path}") return {} # 读取配置文件 with open(config_file.GetFullPath(), 'r', encoding='utf-8') as f: data = json.load(f) # 验证版本 version = data.get("version", "1.0") if version != "1.0": _config.Log("Warning", f"不支持的配置文件版本: {version}，期望 1.0") return {} crops_data = data.get("crops", {}) crops_dict = {} # 处理所有作物类别 for category, crop_list in crops_data.items(): for crop_data in crop_list: try: # 构建 GardenCropDefinition 对象 extra_reward_data = crop_data.get("extra_reward", {}) extra_reward = GardenExtraReward( kind=extra_reward_data.get("kind", "points"), payload=extra_reward_data.get("payload", {}), base_rate=extra_reward_data.get("base_rate", 0.0) ) crop = GardenCropDefinition( seed_id=crop_data["seed_id"], fruit_id=crop_data["fruit_id"], display_name=crop_data["display_name"], tier=crop_data["tier"], growth_minutes=crop_data["growth_minutes"], seed_price=crop_data["seed_price"], base_yield=crop_data["base_yield"], extra_reward=extra_reward, extra_item_id=crop_data.get("extra_item_id"), wine_item_id=crop_data.get("wine_item_id"), wine_tier=crop_data.get("wine_tier") ) crops_dict[crop.seed_id] = crop except KeyError as e: _config.Log("Warning", f"作物配置缺少必要字段: {e}, 跳过此作物") continue except Exception as e: _config.Log("Warning", f"解析作物配置失败: {e}, 跳过此作物") continue _config.Log("Info", f"从配置文件加载了 {len(crops_dict)} 种作物") return crops_dict except json.JSONDecodeError as e: _config.Log("Warning", f"配置文件JSON格式错误: {e}") return {} except Exception as e: _config.Log("Warning", f"加载作物配置文件失败: {e}") return {} # 延迟初始化作物字典，避免循环导入 _GARDEN_CROPS_CACHE: Optional[Dict[str, GardenCropDefinition]] = None _GARDEN_FRUITS_CACHE: Optional[Dict[str, GardenCropDefinition]] = None def _initialize_crops() -> Dict[str, GardenCropDefinition]: """延迟初始化作物字典""" global _GARDEN_CROPS_CACHE if _GARDEN_CROPS_CACHE is None: # 延迟导入避免循环依赖 from .garden_service import GardenConfig config = GardenConfig.load() config_crops = load_crops_from_config(config.crops_config_path) if config_crops: _GARDEN_CROPS_CACHE = config_crops else: # fallback到硬编码定义 _config.Log("Info", "使用硬编码作物定义作为fallback") _GARDEN_CROPS_CACHE = {crop.seed_id: crop for crop in (*COMMON_HERB_CROPS, *RARE_TREE_CROPS)} return _GARDEN_CROPS_CACHE def _initialize_fruits() -> Dict[str, GardenCropDefinition]: """延迟初始化果实字典""" global _GARDEN_FRUITS_CACHE if _GARDEN_FRUITS_CACHE is None: crops = _initialize_crops() _GARDEN_FRUITS_CACHE = {crop.fruit_id: crop for crop in crops.values()} return _GARDEN_FRUITS_CACHE def register_crop(crop: GardenCropDefinition, *, overwrite: bool = False) -> None: """运行时注册作物定义到菜园系统 允许外部系统（如冒险系统）动态注册种子，使其可以在菜园中种植。 Args: crop: 作物定义对象 overwrite: 如果为True，允许覆盖已存在的作物；如果为False，重复注册会记录警告但不覆盖 Raises: ValueError: 如果种子ID或果实ID已存在且overwrite=False """ # 初始化缓存（如果还未初始化） crops_dict = _initialize_crops() fruits_dict = _initialize_fruits() if crop.seed_id in crops_dict: if overwrite: old_crop = crops_dict[crop.seed_id] _config.Log( "Info", f"{ConsoleFrontColor.YELLOW}覆盖已存在的种子定义: {crop.seed_id} ({old_crop.display_name} -> {crop.display_name}){ConsoleFrontColor.RESET}" ) else: _config.Log( "Warning", f"{ConsoleFrontColor.YELLOW}种子 {crop.seed_id} ({crop.display_name}) 已存在，跳过注册。如需覆盖，请设置 overwrite=True{ConsoleFrontColor.RESET}" ) return if crop.fruit_id in fruits_dict: if overwrite: _config.Log( "Info", f"{ConsoleFrontColor.YELLOW}覆盖已存在的果实定义: {crop.fruit_id}{ConsoleFrontColor.RESET}" ) else: _config.Log( "Warning", f"{ConsoleFrontColor.YELLOW}果实 {crop.fruit_id} 已存在，跳过注册。如需覆盖，请设置 overwrite=True{ConsoleFrontColor.RESET}" ) return # 更新缓存 global _GARDEN_CROPS_CACHE, _GARDEN_FRUITS_CACHE if _GARDEN_CROPS_CACHE is None: _GARDEN_CROPS_CACHE = {} if _GARDEN_FRUITS_CACHE is None: _GARDEN_FRUITS_CACHE = {} _GARDEN_CROPS_CACHE[crop.seed_id] = crop _GARDEN_FRUITS_CACHE[crop.fruit_id] = crop _config.Log( "Info", f"{ConsoleFrontColor.GREEN}成功注册作物: {crop.display_name} (种子ID: {crop.seed_id}, 果实ID: {crop.fruit_id}){ConsoleFrontColor.RESET}" ) # 创建只读代理类，提供延迟初始化的字典访问 class _GardenCropsProxy: """代理类，提供只读的作物字典访问""" def __getitem__(self, key): return _initialize_crops()[key] def __iter__(self): return iter(_initialize_crops()) def __len__(self): return len(_initialize_crops()) def __contains__(self, key): return key in _initialize_crops() def keys(self): return _initialize_crops().keys() def values(self): return _initialize_crops().values() def items(self): return _initialize_crops().items() def get(self, key, default=None): return _initialize_crops().get(key, default) class _GardenFruitsProxy: """代理类，提供只读的果实字典访问""" def __getitem__(self, key): return _initialize_fruits()[key] def __iter__(self): return iter(_initialize_fruits()) def __len__(self): return len(_initialize_fruits()) def __contains__(self, key): return key in _initialize_fruits() def keys(self): return _initialize_fruits().keys() def values(self): return _initialize_fruits().values() def items(self): return _initialize_fruits().items() def get(self, key, default=None): return _initialize_fruits().get(key, default) # 导出只读代理对象 GARDEN_CROPS = _GardenCropsProxy() GARDEN_FRUITS = _GardenFruitsProxy() GARDEN_MISC_ITEMS = { "garden_item_rot_fruit": { "name": "腐败的果实", "tier": "common", "description": "放置过久的果实，只能作为炼金失败的副产物。", } } def get_garden_db_models() -> List[DatabaseModel]: return [ DatabaseModel( table_name="garden_plots", column_defs={ "user_id": "INTEGER NOT NULL", "chat_id": "INTEGER NOT NULL", "plot_index": "INTEGER NOT NULL", "seed_id": "TEXT NOT NULL", "seed_quality": "TEXT NOT NULL DEFAULT 'common'", "planted_at": "TEXT NOT NULL", "mature_at": "TEXT NOT NULL", "is_mature": "INTEGER NOT NULL DEFAULT 0", "base_yield": "INTEGER NOT NULL", "extra_type": "TEXT", "extra_payload": "TEXT", "remaining_fruit": "INTEGER NOT NULL", "theft_users": "TEXT DEFAULT '[]'", "scheduled_task_id": "INTEGER", "PRIMARY KEY (user_id, plot_index)": "", }, ), ] __all__ = [ "GardenCropDefinition", "GardenExtraReward", "GARDEN_CROPS", "GARDEN_FRUITS", "GARDEN_MISC_ITEMS", "get_garden_db_models", "load_crops_from_config", "register_crop", ]