Files
NewWPSBot/Plugins/WPSGardenSystem/garden_models.py
2025-11-15 17:06:12 +08:00

502 lines
17 KiB
Python
Raw Blame History

This file contains ambiguous Unicode characters

This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.

"""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 GardenTrapDefinition(BaseModel):
"""陷阱物品定义"""
item_id: str
display_name: str
tier: str # common / rare / epic / legendary
description: str
trigger_rate: float = Field(..., ge=0.0, le=1.0, description="触发概率")
fine_points: int = Field(..., ge=0, description="罚金积分")
ban_hours: int = Field(..., ge=0, description="禁止偷盗时长(小时)")
durability: int = Field(..., ge=1, description="耐久度(可触发次数)")
trigger_message: str = Field(..., description="触发时发送给偷盗者的消息")
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": "放置过久的果实,只能作为炼金失败的副产物。",
}
}
# 陷阱物品定义
GARDEN_TRAPS: Tuple[GardenTrapDefinition, ...] = (
GardenTrapDefinition(
item_id="garden_trap_net",
display_name="防盗网",
tier="common",
description="基础防护陷阱50%概率触发,对偷盗者造成小额罚金并短时禁止偷盗。",
trigger_rate=0.5,
fine_points=1000,
ban_hours=12,
durability=1, # 普通陷阱1次使用
trigger_message="🕸️ 你触发了防盗网!被罚款 {fine} 分,并且在 {hours} 小时内无法继续偷盗。",
),
GardenTrapDefinition(
item_id="garden_trap_thorn",
display_name="荆棘陷阱",
tier="rare",
description="带刺的防护陷阱60%概率触发,造成中等罚金并禁止偷盗更长时间。",
trigger_rate=0.6,
fine_points=2500,
ban_hours=24,
durability=2, # 稀有陷阱2次使用
trigger_message="🌵 你踩到了荆棘陷阱!被罚款 {fine} 分,并且在 {hours} 小时内无法继续偷盗。",
),
GardenTrapDefinition(
item_id="garden_trap_magic",
display_name="魔法结界",
tier="epic",
description="强大的魔法防护70%概率触发,造成高额罚金并长时间禁止偷盗。",
trigger_rate=0.7,
fine_points=5000,
ban_hours=48,
durability=3, # 史诗陷阱4次使用
trigger_message="✨ 你触碰到了魔法结界!被罚款 {fine} 分,并且在 {hours} 小时内无法继续偷盗。",
),
GardenTrapDefinition(
item_id="garden_trap_legend",
display_name="传奇守护",
tier="legendary",
description="传说级的防护装置80%概率触发,造成巨额罚金并长期禁止偷盗。",
trigger_rate=0.8,
fine_points=10000,
ban_hours=72,
durability=4, # 传说陷阱4次使用
trigger_message="⚡ 你惊醒了传奇守护!被罚款 {fine} 分,并且在 {hours} 小时内无法继续偷盗。",
),
)
# 陷阱物品字典item_id -> GardenTrapDefinition
GARDEN_TRAPS_DICT: Dict[str, GardenTrapDefinition] = {trap.item_id: trap for trap in GARDEN_TRAPS}
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",
"trap_item_id": "TEXT",
"trap_config": "TEXT",
"trap_durability": "INTEGER DEFAULT 0",
"PRIMARY KEY (user_id, plot_index)": "",
},
),
DatabaseModel(
table_name="garden_theft_ban",
column_defs={
"user_id": "INTEGER NOT NULL",
"banned_until": "TEXT NOT NULL",
"PRIMARY KEY (user_id)": "",
},
),
]
__all__ = [
"GardenCropDefinition",
"GardenExtraReward",
"GardenTrapDefinition",
"GARDEN_CROPS",
"GARDEN_FRUITS",
"GARDEN_MISC_ITEMS",
"GARDEN_TRAPS",
"GARDEN_TRAPS_DICT",
"get_garden_db_models",
"load_crops_from_config",
"register_crop",
]