refactor all gather logic
This commit is contained in:
bridge
@@ -1,9 +1,8 @@
|
||||
from __future__ import annotations
|
||||
|
||||
import random
|
||||
from src.classes.action import TimedAction
|
||||
from src.classes.event import Event
|
||||
from src.classes.region import NormalRegion
|
||||
from src.utils.gather import execute_gather, check_can_start_gather
|
||||
|
||||
|
||||
class Harvest(TimedAction):
|
||||
@@ -20,54 +19,31 @@ class Harvest(TimedAction):
|
||||
|
||||
duration_months = 6
|
||||
|
||||
def __init__(self, avatar, world):
|
||||
super().__init__(avatar, world)
|
||||
self.gained_items: dict[str, int] = {}
|
||||
|
||||
def _execute(self) -> None:
|
||||
"""
|
||||
执行采集动作
|
||||
"""
|
||||
region = self.avatar.tile.region
|
||||
plants = getattr(region, "plants", [])
|
||||
if len(plants) == 0:
|
||||
return
|
||||
available_plants = [
|
||||
plant for plant in plants
|
||||
if self.avatar.cultivation_progress.realm >= plant.realm
|
||||
]
|
||||
if len(available_plants) == 0:
|
||||
return
|
||||
|
||||
# 目前固定100%成功率
|
||||
if random.random() < 1.0:
|
||||
target_plant = random.choice(available_plants)
|
||||
# 随机选择该植物的一种物品
|
||||
item = random.choice(target_plant.items)
|
||||
# 基础获得1个，额外物品来自effects
|
||||
base_quantity = 1
|
||||
extra_items = int(self.avatar.effects.get("extra_harvest_items", 0) or 0)
|
||||
total_quantity = base_quantity + extra_items
|
||||
self.avatar.add_item(item, total_quantity)
|
||||
gained = execute_gather(self.avatar, "plants", "extra_harvest_items")
|
||||
for name, count in gained.items():
|
||||
self.gained_items[name] = self.gained_items.get(name, 0) + count
|
||||
|
||||
def can_start(self) -> tuple[bool, str]:
|
||||
region = self.avatar.tile.region
|
||||
if not isinstance(region, NormalRegion):
|
||||
return False, "当前不在普通区域"
|
||||
plants = getattr(region, "plants", [])
|
||||
if len(plants) == 0:
|
||||
return False, "当前区域没有植物"
|
||||
available_plants = [
|
||||
plant for plant in plants
|
||||
if self.avatar.cultivation_progress.realm >= plant.realm
|
||||
]
|
||||
if len(available_plants) == 0:
|
||||
return False, "当前区域的植物境界过高"
|
||||
return True, ""
|
||||
return check_can_start_gather(self.avatar, "plants", "植物")
|
||||
|
||||
def start(self) -> Event:
|
||||
region = self.avatar.tile.region
|
||||
return Event(self.world.month_stamp, f"{self.avatar.name} 在 {self.avatar.tile.location_name} 开始采集", related_avatars=[self.avatar.id])
|
||||
|
||||
# TimedAction 已统一 step 逻辑
|
||||
|
||||
async def finish(self) -> list[Event]:
|
||||
return []
|
||||
|
||||
|
||||
# 必定有产出
|
||||
items_desc = "、".join([f"{k}x{v}" for k, v in self.gained_items.items()])
|
||||
return [Event(
|
||||
self.world.month_stamp,
|
||||
f"{self.avatar.name} 结束了采集，获得了：{items_desc}",
|
||||
related_avatars=[self.avatar.id]
|
||||
)]
|
||||
|
||||
@@ -1,9 +1,8 @@
|
||||
from __future__ import annotations
|
||||
|
||||
import random
|
||||
from src.classes.action import TimedAction
|
||||
from src.classes.event import Event
|
||||
from src.classes.region import NormalRegion
|
||||
from src.utils.gather import execute_gather, check_can_start_gather
|
||||
|
||||
|
||||
class Hunt(TimedAction):
|
||||
@@ -20,54 +19,31 @@ class Hunt(TimedAction):
|
||||
|
||||
duration_months = 6
|
||||
|
||||
def __init__(self, avatar, world):
|
||||
super().__init__(avatar, world)
|
||||
self.gained_items: dict[str, int] = {}
|
||||
|
||||
def _execute(self) -> None:
|
||||
"""
|
||||
执行狩猎动作
|
||||
"""
|
||||
region = self.avatar.tile.region
|
||||
animals = getattr(region, "animals", [])
|
||||
if len(animals) == 0:
|
||||
return
|
||||
available_animals = [
|
||||
animal for animal in animals
|
||||
if self.avatar.cultivation_progress.realm >= animal.realm
|
||||
]
|
||||
if len(available_animals) == 0:
|
||||
return
|
||||
|
||||
# 目前固定100%成功率
|
||||
if random.random() < 1.0:
|
||||
target_animal = random.choice(available_animals)
|
||||
# 随机选择该动物的一种物品
|
||||
item = random.choice(target_animal.items)
|
||||
# 基础获得1个，额外物品来自effects
|
||||
base_quantity = 1
|
||||
extra_items = int(self.avatar.effects.get("extra_hunt_items", 0) or 0)
|
||||
total_quantity = base_quantity + extra_items
|
||||
self.avatar.add_item(item, total_quantity)
|
||||
gained = execute_gather(self.avatar, "animals", "extra_hunt_items")
|
||||
for name, count in gained.items():
|
||||
self.gained_items[name] = self.gained_items.get(name, 0) + count
|
||||
|
||||
def can_start(self) -> tuple[bool, str]:
|
||||
region = self.avatar.tile.region
|
||||
if not isinstance(region, NormalRegion):
|
||||
return False, "当前不在普通区域"
|
||||
animals = getattr(region, "animals", [])
|
||||
if len(animals) == 0:
|
||||
return False, f"当前区域{region.name}没有动物"
|
||||
available_animals = [
|
||||
animal for animal in animals
|
||||
if self.avatar.cultivation_progress.realm >= animal.realm
|
||||
]
|
||||
if len(available_animals) == 0:
|
||||
return False, "当前区域的动物境界过高"
|
||||
return True, ""
|
||||
return check_can_start_gather(self.avatar, "animals", "动物")
|
||||
|
||||
def start(self) -> Event:
|
||||
region = self.avatar.tile.region
|
||||
return Event(self.world.month_stamp, f"{self.avatar.name} 在 {self.avatar.tile.location_name} 开始狩猎", related_avatars=[self.avatar.id])
|
||||
|
||||
# TimedAction 已统一 step 逻辑
|
||||
|
||||
async def finish(self) -> list[Event]:
|
||||
return []
|
||||
|
||||
|
||||
# 必定有产出
|
||||
items_desc = "、".join([f"{k}x{v}" for k, v in self.gained_items.items()])
|
||||
return [Event(
|
||||
self.world.month_stamp,
|
||||
f"{self.avatar.name} 结束了狩猎，获得了：{items_desc}",
|
||||
related_avatars=[self.avatar.id]
|
||||
)]
|
||||
|
||||
@@ -1,9 +1,8 @@
|
||||
from __future__ import annotations
|
||||
|
||||
import random
|
||||
from src.classes.action import TimedAction
|
||||
from src.classes.event import Event
|
||||
from src.classes.region import NormalRegion
|
||||
from src.utils.gather import execute_gather, check_can_start_gather
|
||||
|
||||
|
||||
class Mine(TimedAction):
|
||||
@@ -20,53 +19,30 @@ class Mine(TimedAction):
|
||||
|
||||
duration_months = 6
|
||||
|
||||
def __init__(self, avatar, world):
|
||||
super().__init__(avatar, world)
|
||||
self.gained_items: dict[str, int] = {}
|
||||
|
||||
def _execute(self) -> None:
|
||||
"""
|
||||
执行挖矿动作
|
||||
"""
|
||||
region = self.avatar.tile.region
|
||||
lodes = getattr(region, "lodes", [])
|
||||
if len(lodes) == 0:
|
||||
return
|
||||
available_lodes = [
|
||||
lode for lode in lodes
|
||||
if self.avatar.cultivation_progress.realm >= lode.realm
|
||||
]
|
||||
if len(available_lodes) == 0:
|
||||
return
|
||||
|
||||
# 目前固定100%成功率
|
||||
if random.random() < 1.0:
|
||||
target_lode = random.choice(available_lodes)
|
||||
# 随机选择该矿脉的一种物品
|
||||
item = random.choice(target_lode.items)
|
||||
# 基础获得1个，额外物品来自effects
|
||||
base_quantity = 1
|
||||
extra_items = int(self.avatar.effects.get("extra_mine_items", 0) or 0)
|
||||
total_quantity = base_quantity + extra_items
|
||||
self.avatar.add_item(item, total_quantity)
|
||||
gained = execute_gather(self.avatar, "lodes", "extra_mine_items")
|
||||
for name, count in gained.items():
|
||||
self.gained_items[name] = self.gained_items.get(name, 0) + count
|
||||
|
||||
def can_start(self) -> tuple[bool, str]:
|
||||
region = self.avatar.tile.region
|
||||
if not isinstance(region, NormalRegion):
|
||||
return False, "当前不在普通区域"
|
||||
lodes = getattr(region, "lodes", [])
|
||||
if len(lodes) == 0:
|
||||
return False, "当前区域没有矿脉"
|
||||
available_lodes = [
|
||||
lode for lode in lodes
|
||||
if self.avatar.cultivation_progress.realm >= lode.realm
|
||||
]
|
||||
if len(available_lodes) == 0:
|
||||
return False, "当前区域的矿脉境界过高"
|
||||
return True, ""
|
||||
return check_can_start_gather(self.avatar, "lodes", "矿脉")
|
||||
|
||||
def start(self) -> Event:
|
||||
region = self.avatar.tile.region
|
||||
return Event(self.world.month_stamp, f"{self.avatar.name} 在 {self.avatar.tile.location_name} 开始挖矿", related_avatars=[self.avatar.id])
|
||||
|
||||
# TimedAction 已统一 step 逻辑
|
||||
|
||||
async def finish(self) -> list[Event]:
|
||||
return []
|
||||
|
||||
items_desc = "、".join([f"{k}x{v}" for k, v in self.gained_items.items()])
|
||||
return [Event(
|
||||
self.world.month_stamp,
|
||||
f"{self.avatar.name} 结束了挖矿，获得了：{items_desc}",
|
||||
related_avatars=[self.avatar.id]
|
||||
)]
|
||||
|
||||
76
src/utils/gather.py
Normal file
76
src/utils/gather.py
Normal file
@@ -0,0 +1,76 @@
|
||||
from __future__ import annotations
|
||||
import random
|
||||
from typing import TYPE_CHECKING, Any
|
||||
|
||||
if TYPE_CHECKING:
|
||||
from src.classes.avatar import Avatar
|
||||
|
||||
def check_can_start_gather(
|
||||
avatar: Avatar,
|
||||
resource_attr: str, # "lodes", "animals", "plants"
|
||||
resource_name_cn: str # "矿脉", "动物", "植物"
|
||||
) -> tuple[bool, str]:
|
||||
from src.classes.region import NormalRegion
|
||||
|
||||
region = avatar.tile.region
|
||||
if not isinstance(region, NormalRegion):
|
||||
return False, "当前不在普通区域"
|
||||
|
||||
resources = getattr(region, resource_attr, [])
|
||||
if not resources:
|
||||
return False, f"当前区域没有{resource_name_cn}"
|
||||
|
||||
# 筛选境界符合的资源
|
||||
available = [
|
||||
r for r in resources
|
||||
if avatar.cultivation_progress.realm >= r.realm
|
||||
]
|
||||
if not available:
|
||||
return False, f"当前区域的{resource_name_cn}境界过高"
|
||||
|
||||
return True, ""
|
||||
|
||||
def execute_gather(
|
||||
avatar: Avatar,
|
||||
resource_attr: str,
|
||||
extra_effect_key: str
|
||||
) -> dict[str, int]:
|
||||
"""
|
||||
执行采集逻辑。
|
||||
返回: {item_name: count}
|
||||
"""
|
||||
from src.classes.region import NormalRegion
|
||||
region = avatar.tile.region
|
||||
|
||||
# 再次校验类型，防止运行时环境变化
|
||||
if not isinstance(region, NormalRegion):
|
||||
return {}
|
||||
|
||||
resources = getattr(region, resource_attr, [])
|
||||
|
||||
# 筛选
|
||||
available = [
|
||||
r for r in resources
|
||||
if avatar.cultivation_progress.realm >= r.realm
|
||||
]
|
||||
|
||||
if not available:
|
||||
return {}
|
||||
|
||||
# 1. 随机选择资源点 (均匀分布)
|
||||
target = random.choice(available)
|
||||
|
||||
# 2. 随机选择产出物
|
||||
if not hasattr(target, "items") or not target.items:
|
||||
return {}
|
||||
|
||||
item = random.choice(target.items)
|
||||
|
||||
base_quantity = 1
|
||||
extra_items = int(avatar.effects.get(extra_effect_key, 0) or 0)
|
||||
total_quantity = max(1, base_quantity + extra_items)
|
||||
|
||||
avatar.add_item(item, total_quantity)
|
||||
|
||||
return {item.name: total_quantity}
|
||||
|
||||
117
tests/test_gather.py
Normal file
117
tests/test_gather.py
Normal file
@@ -0,0 +1,117 @@
|
||||
import pytest
|
||||
from unittest.mock import MagicMock, patch
|
||||
from src.utils.gather import execute_gather, check_can_start_gather
|
||||
from src.classes.cultivation import Realm
|
||||
from src.classes.region import NormalRegion
|
||||
|
||||
@pytest.fixture
|
||||
def mock_region(dummy_avatar):
|
||||
"""设置一个 Mock 的 NormalRegion 到 avatar 所在 tile"""
|
||||
real_region = NormalRegion(id=999, name="TestRegion", desc="Testing")
|
||||
# Bypass post_init loading from global dicts by manually setting fields
|
||||
real_region.lodes = []
|
||||
real_region.animals = []
|
||||
real_region.plants = []
|
||||
|
||||
dummy_avatar.tile.region = real_region
|
||||
return real_region
|
||||
|
||||
@pytest.fixture
|
||||
def mock_resource_item():
|
||||
item = MagicMock()
|
||||
item.name = "TestItem"
|
||||
item.realm = Realm.Qi_Refinement
|
||||
return item
|
||||
|
||||
@pytest.fixture
|
||||
def mock_resource(mock_resource_item):
|
||||
"""创建一个通用的资源对象 (Lode/Animal/Plant)"""
|
||||
res = MagicMock()
|
||||
res.realm = Realm.Qi_Refinement
|
||||
res.items = [mock_resource_item]
|
||||
return res
|
||||
|
||||
def test_check_can_start_gather_success(dummy_avatar, mock_region, mock_resource):
|
||||
"""测试采集检查通过的情况"""
|
||||
mock_region.lodes = [mock_resource]
|
||||
|
||||
can, msg = check_can_start_gather(dummy_avatar, "lodes", "矿脉")
|
||||
assert can is True
|
||||
assert msg == ""
|
||||
|
||||
def test_check_can_start_gather_not_normal_region(dummy_avatar):
|
||||
"""测试不在普通区域的情况"""
|
||||
dummy_avatar.tile.region = "NotARegion"
|
||||
|
||||
can, msg = check_can_start_gather(dummy_avatar, "lodes", "矿脉")
|
||||
assert can is False
|
||||
assert "当前不在普通区域" in msg
|
||||
|
||||
def test_check_can_start_gather_no_resources(dummy_avatar, mock_region):
|
||||
"""测试区域没有资源的情况"""
|
||||
mock_region.lodes = []
|
||||
|
||||
can, msg = check_can_start_gather(dummy_avatar, "lodes", "矿脉")
|
||||
assert can is False
|
||||
assert "当前区域没有矿脉" in msg
|
||||
|
||||
def test_check_can_start_gather_realm_too_low(dummy_avatar, mock_region, mock_resource):
|
||||
"""测试境界不足的情况"""
|
||||
# 提升资源境界到筑基
|
||||
mock_resource.realm = Realm.Foundation_Establishment
|
||||
mock_region.lodes = [mock_resource]
|
||||
# avatar 默认为练气
|
||||
|
||||
can, msg = check_can_start_gather(dummy_avatar, "lodes", "矿脉")
|
||||
assert can is False
|
||||
assert "当前区域的矿脉境界过高" in msg
|
||||
|
||||
def test_execute_gather_success(dummy_avatar, mock_region, mock_resource, mock_resource_item):
|
||||
"""测试执行采集逻辑成功"""
|
||||
mock_region.lodes = [mock_resource]
|
||||
|
||||
# 模拟 add_item
|
||||
dummy_avatar.add_item = MagicMock()
|
||||
|
||||
result = execute_gather(dummy_avatar, "lodes", "extra_mine_items")
|
||||
|
||||
assert "TestItem" in result
|
||||
assert result["TestItem"] >= 1
|
||||
dummy_avatar.add_item.assert_called_once()
|
||||
|
||||
# 验证获得的物品是正确的
|
||||
args, _ = dummy_avatar.add_item.call_args
|
||||
assert args[0] == mock_resource_item
|
||||
assert args[1] >= 1
|
||||
|
||||
def test_execute_gather_with_extra_effect(dummy_avatar, mock_region, mock_resource):
|
||||
"""测试带有加成效果的采集"""
|
||||
mock_region.lodes = [mock_resource]
|
||||
|
||||
# effects 是只读属性，它通过合并各个组件的 effects 来计算。
|
||||
# 为了测试，我们 Mock 掉 effects 属性。
|
||||
with patch.object(type(dummy_avatar), 'effects', new_callable=lambda: {"extra_mine_items": 2}):
|
||||
dummy_avatar.add_item = MagicMock()
|
||||
|
||||
result = execute_gather(dummy_avatar, "lodes", "extra_mine_items")
|
||||
|
||||
# 基础1 + 加成2 = 3
|
||||
assert result["TestItem"] == 3
|
||||
|
||||
def test_execute_gather_random_selection(dummy_avatar, mock_region):
|
||||
"""测试从多个资源中随机选择"""
|
||||
res1 = MagicMock()
|
||||
res1.realm = Realm.Qi_Refinement
|
||||
res1.items = [MagicMock(name="Item1")]
|
||||
res1.items[0].name = "Item1"
|
||||
|
||||
res2 = MagicMock()
|
||||
res2.realm = Realm.Qi_Refinement
|
||||
res2.items = [MagicMock(name="Item2")]
|
||||
res2.items[0].name = "Item2"
|
||||
|
||||
mock_region.lodes = [res1, res2]
|
||||
dummy_avatar.add_item = MagicMock()
|
||||
|
||||
execute_gather(dummy_avatar, "lodes", "extra_mine_items")
|
||||
dummy_avatar.add_item.assert_called_once()
|
||||
Reference in New Issue
Block a user