diff --git a/src/classes/action/assassinate.py b/src/classes/action/assassinate.py index 24d59a0..9ef7c09 100644 --- a/src/classes/action/assassinate.py +++ b/src/classes/action/assassinate.py @@ -127,34 +127,17 @@ class Assassinate(InstantAction): if not (isinstance(res, tuple) and len(res) == 4): return [] - winner, loser, loser_damage, winner_damage = res + start_text = getattr(self, '_start_event_content', "") - is_fatal = loser.hp <= 0 - - prefix = f"暗杀失败！双方爆发激战。" - - if is_fatal: - result_text = f"{prefix} {winner.name} 最终战胜并斩杀了 {loser.name} (伤害 {loser_damage})。" - loot_text = await kill_and_grab(winner, loser) - result_text += loot_text - else: - result_text = f"{prefix} {winner.name} 战胜了 {loser.name}，造成 {loser_damage} 点伤害，自身受损 {winner_damage} 点。" - - result_event = Event(self.world.month_stamp, result_text, related_avatars=rel_ids, is_major=True) - - # 生成故事 - story = await StoryTeller.tell_story( - self._start_event_content, - result_event.content, + from src.classes.battle import handle_battle_finish + return await handle_battle_finish( + self.world, self.avatar, target, - prompt=self.STORY_PROMPT_FAIL, - allow_relation_changes=True + res, + start_text, + self.STORY_PROMPT_FAIL, + prefix="暗杀失败！双方爆发激战。", + check_loot=True ) - story_event = Event(self.world.month_stamp, story, related_avatars=rel_ids, is_story=True) - - if is_fatal: - handle_death(self.world, loser, DeathReason.BATTLE) - - return [result_event, story_event] diff --git a/src/classes/action/attack.py b/src/classes/action/attack.py index 26c8029..a143b8f 100644 --- a/src/classes/action/attack.py +++ b/src/classes/action/attack.py @@ -78,38 +78,17 @@ class Attack(InstantAction): res = self._last_result if not (isinstance(res, tuple) and len(res) == 4): return [] - winner, loser = res[0], res[1] - loser_damage, winner_damage = res[2], res[3] - - # 判定是否致死 - is_fatal = loser.hp <= 0 - if is_fatal: - result_text = f"{winner.name} 战胜了 {loser.name}，造成{loser_damage}点伤害。{loser.name} 遭受重创，当场陨落。" - - # 杀人夺宝 - loot_text = await kill_and_grab(winner, loser) - result_text += loot_text - - else: - result_text = f"{winner.name} 战胜了 {loser.name}，{loser.name} 受伤{loser_damage}点，{winner.name} 也受伤{winner_damage}点" - - rel_ids = [self.avatar.id] + target = self._get_target(avatar_name) - try: - if target is not None: - rel_ids.append(target.id) - except Exception: - pass - result_event = Event(self.world.month_stamp, result_text, related_avatars=rel_ids, is_major=True) - - # 生成战斗小故事 - start_text = self._start_event_content if hasattr(self, '_start_event_content') else result_event.content - # 战斗强制双人模式，允许改变关系 - story = await StoryTeller.tell_story(start_text, result_event.content, self.avatar, target, prompt=self.STORY_PROMPT, allow_relation_changes=True) - story_event = Event(self.world.month_stamp, story, related_avatars=rel_ids, is_story=True) - - # 如果死亡，执行死亡清理（在故事生成后，保证关系数据可用） - if is_fatal: - handle_death(self.world, loser, DeathReason.BATTLE) - - return [result_event, story_event] + start_text = getattr(self, '_start_event_content', "") + + from src.classes.battle import handle_battle_finish + return await handle_battle_finish( + self.world, + self.avatar, + target, + res, + start_text, + self.STORY_PROMPT, + check_loot=True + ) diff --git a/src/classes/avatar.py b/src/classes/avatar.py deleted file mode 100644 index d3b6b57..0000000 --- a/src/classes/avatar.py +++ /dev/null @@ -1,1026 +0,0 @@ -import random -from dataclasses import dataclass, field -from enum import Enum -from typing import Optional, List, TYPE_CHECKING -from collections import defaultdict -import json - -if TYPE_CHECKING: - from src.classes.sect_ranks import SectRank - -from src.classes.calendar import MonthStamp -from src.classes.action import Action -from src.classes.action_runtime import ActionStatus, ActionResult -from src.classes.action.registry import ActionRegistry -from src.classes.world import World -from src.sim.save.avatar_save_mixin import AvatarSaveMixin -from src.sim.load.avatar_load_mixin import AvatarLoadMixin -from src.classes.tile import Tile -from src.classes.region import Region -from src.classes.cultivation import CultivationProgress -from src.classes.root import Root -from src.classes.technique import Technique, get_random_technique_for_avatar, get_technique_by_sect -from src.classes.age import Age -from src.classes.event import NULL_EVENT, Event -from src.classes.typings import ACTION_NAME, ACTION_PARAMS, ACTION_NAME_PARAMS_PAIRS, ACTION_NAME_PARAMS_PAIR -from src.classes.action_runtime import ActionPlan, ActionInstance -from src.classes.effect import _merge_effects, _evaluate_conditional_effect -from src.classes.alignment import Alignment -from src.classes.persona import Persona, personas_by_id, get_random_compatible_personas -from src.classes.item import Item -from src.classes.weapon import Weapon, get_common_weapon -from src.classes.auxiliary import Auxiliary -from src.classes.weapon_type import WeaponType -from src.classes.equipment_grade import EquipmentGrade -from src.classes.magic_stone import MagicStone -from src.classes.hp_and_mp import HP, HP_MAX_BY_REALM -from src.utils.id_generator import get_avatar_id -from src.utils.config import CONFIG -from src.classes.relation import Relation, get_reciprocal, get_relation_label -from src.run.log import get_logger -from src.classes.alignment import Alignment -from src.utils.params import filter_kwargs_for_callable -from src.classes.sect import Sect -from src.classes.appearance import Appearance, get_random_appearance -from src.classes.battle import get_base_strength -from src.classes.spirit_animal import SpiritAnimal -from src.classes.long_term_objective import LongTermObjective -from src.classes.nickname_data import Nickname - -persona_num = CONFIG.avatar.persona_num - -class Gender(Enum): - MALE = "male" - FEMALE = "female" - - def __str__(self) -> str: - return gender_strs.get(self, self.value) - -gender_strs = { - Gender.MALE: "男", - Gender.FEMALE: "女", -} - -# 历史事件的最大数量 -MAX_HISTORY_EVENTS = 10 - -@dataclass -class Avatar(AvatarSaveMixin, AvatarLoadMixin): - """ - NPC的类。 - 包含了这个角色的一切信息。 - """ - world: World - name: str - id: str - birth_month_stamp: MonthStamp - age: Age - gender: Gender - cultivation_progress: CultivationProgress = field(default_factory=lambda: CultivationProgress(0)) - pos_x: int = 0 - pos_y: int = 0 - tile: Optional[Tile] = None - - root: Root = field(default_factory=lambda: random.choice(list(Root))) - personas: List[Persona] = field(default_factory=list) - technique: Technique | None = None - history_events: List[Event] = field(default_factory=list) - _pending_events: List[Event] = field(default_factory=list) - current_action: Optional[ActionInstance] = None - planned_actions: List[ActionPlan] = field(default_factory=list) - thinking: str = "" - short_term_objective: str = "" - long_term_objective: Optional[LongTermObjective] = None - magic_stone: MagicStone = field(default_factory=lambda: MagicStone(0)) # 灵石，即货币 - items: dict[Item, int] = field(default_factory=dict) - hp: HP = field(default_factory=lambda: HP(0, 0)) # 将在__post_init__中初始化 - relations: dict["Avatar", Relation] = field(default_factory=dict) - alignment: Alignment | None = None - # 所属宗门（可为空，表示散修/无门无派） - sect: Sect | None = None - # 宗门职位（仅当有宗门时有效） - sect_rank: "SectRank | None" = None - # 外貌（1~10级），创建时随机生成 - appearance: Appearance = field(default_factory=get_random_appearance) - # 兵器（必有，无则分配普通兵器） - weapon: Optional[Weapon] = None - # 兵器熟练度（0-100），更换兵器归零 - weapon_proficiency: float = 0.0 - # 辅助装备（可选） - auxiliary: Optional[Auxiliary] = None - # 灵兽：最多一个；若再次捕捉则覆盖 - spirit_animal: Optional[SpiritAnimal] = None - # 绰号：江湖中对该角色的称谓，满足条件后生成，永久不变 - nickname: Optional[Nickname] = None - # 自定义头像ID：如果设置，优先使用此ID显示头像 - custom_pic_id: Optional[int] = None - - # 死亡状态 - is_dead: bool = False - # 死亡信息：{ "time": MonthStamp, "reason": str, "location": (x, y) } - death_info: Optional[dict] = None - - # 当月/当步新设动作标记：在 commit_next_plan 设为 True，首次 tick_action 后清为 False - _new_action_set_this_step: bool = False - # 动作冷却：记录动作类名 -> 上次完成月戳 - _action_cd_last_months: dict[str, int] = field(default_factory=dict) - # 不缓存 effects；实时从宗门与功法合并 - - # 知道的区域 ID 集合 - known_regions: set[int] = field(default_factory=set) - - def join_sect(self, sect: Sect, rank: "SectRank") -> None: - """加入宗门""" - if self.is_dead: - return - if self.sect: - self.leave_sect() - self.sect = sect - self.sect_rank = rank - sect.add_member(self) - # 更新阵营倾向为宗门阵营（如果之前是中立或不同）- 可选，视设计而定 - # 这里暂不强制修改个人阵营，除非完全不兼容 - - def leave_sect(self) -> None: - """退出宗门""" - if self.sect: - self.sect.remove_member(self) - self.sect = None - self.sect_rank = None - - def set_dead(self, reason: str, time: MonthStamp) -> None: - """ - 设置角色死亡状态。 - - Args: - reason: 死亡原因 - time: 死亡时间 - """ - if self.is_dead: - return - - self.is_dead = True - self.death_info = { - "time": int(time), - "reason": reason, - "location": (self.pos_x, self.pos_y) - } - - # 清空所有计划和当前动作 - self.planned_actions.clear() - self.current_action = None - self._pending_events.clear() - self.thinking = "" - self.short_term_objective = "" - - # 退出宗门（保留职位记录还是清除？通常死人不再担任职位） - # 但为了历史记录，也许可以保留 sect 字段，但从宗门成员列表中移除 - if self.sect: - self.sect.remove_member(self) - # 不清除 self.sect 和 self.sect_rank，作为生平记录保留 - - - def __post_init__(self): - """ - 在Avatar创建后自动初始化tile和HP - """ - self.tile = self.world.map.get_tile(self.pos_x, self.pos_y) - - # 根据当前境界初始化HP - max_hp = HP_MAX_BY_REALM.get(self.cultivation_progress.realm, 100) - self.hp = HP(max_hp, max_hp) - - # 最大寿元已在 Age 构造时基于境界初始化 - - # 如果personas列表为空，则随机分配符合条件且不互斥的persona - if not self.personas: - self.personas = get_random_compatible_personas(persona_num, avatar=self) - - # 出生即按宗门分配功法： - # - 散修：仅从无宗门功法抽样 - # - 有宗门：从“无宗门 + 本宗门”集合抽样 - if self.technique is None: - self.technique = get_technique_by_sect(self.sect) - - # 确保宗门引用同步 - if self.sect: - self.sect.add_member(self) - - # 若未设定阵营，则依据宗门/无门无派规则设置，避免后续为 None - if self.alignment is None: - if self.sect is not None: - self.alignment = self.sect.alignment - else: - from src.classes.alignment import Alignment as _Alignment - self.alignment = random.choice(list(_Alignment)) - - - - # 初始化时计算所有长期效果（HP等） - self.recalc_effects() - - # 初始化已知区域 - self._init_known_regions() - - def _init_known_regions(self): - """初始化已知区域：当前位置 + 宗门驻地""" - # 1. 当前位置 - if self.tile and self.tile.region: - self.known_regions.add(self.tile.region.id) - - # 2. 宗门驻地 - if self.sect: - # 遍历地图寻找该宗门的驻地 - # map.sect_regions 是 {region_id: SectRegion} - for r in self.world.map.sect_regions.values(): - if r.sect_id == self.sect.id: - self.known_regions.add(r.id) - break - - @property - def effects(self) -> dict[str, object]: - merged: dict[str, object] = defaultdict(str) - # 来自宗门 - if self.sect is not None: - evaluated = _evaluate_conditional_effect(self.sect.effects, self) - merged = _merge_effects(merged, evaluated) - # 来自功法 - evaluated = _evaluate_conditional_effect(self.technique.effects, self) - merged = _merge_effects(merged, evaluated) - # 来自灵根 - evaluated = _evaluate_conditional_effect(self.root.effects, self) - merged = _merge_effects(merged, evaluated) - # 来自特质（persona） - for persona in self.personas: - evaluated = _evaluate_conditional_effect(persona.effects, self) - merged = _merge_effects(merged, evaluated) - # 来自兵器 - if self.weapon is not None: - evaluated = _evaluate_conditional_effect(self.weapon.effects, self) - merged = _merge_effects(merged, evaluated) - # 来自辅助装备 - if self.auxiliary is not None: - evaluated = _evaluate_conditional_effect(self.auxiliary.effects, self) - merged = _merge_effects(merged, evaluated) - # 来自灵兽 - if self.spirit_animal is not None: - evaluated = _evaluate_conditional_effect(self.spirit_animal.effects, self) - merged = _merge_effects(merged, evaluated) - # 来自天地灵机（世界级buff/debuff） - if self.world.current_phenomenon is not None: - evaluated = _evaluate_conditional_effect(self.world.current_phenomenon.effects, self) - merged = _merge_effects(merged, evaluated) - # 评估动态效果表达式：值以 "eval(...)" 形式给出 - final: dict[str, object] = {} - for k, v in merged.items(): - if isinstance(v, str): - s = v.strip() - if s.startswith("eval(") and s.endswith(")"): - expr = s[5:-1] - final[k] = eval(expr, {"__builtins__": {}}, {"avatar": self}) - continue - final[k] = v - return final - - - def __hash__(self) -> int: - return hash(self.id) - - def get_info(self, detailed: bool = False) -> dict: - """ - 获取 avatar 的信息，返回 dict；根据 detailed 控制信息粒度。 - """ - region = self.tile.region if self.tile is not None else None - from src.classes.relation import get_relations_strs - relation_lines = get_relations_strs(self, max_lines=8) - relations_info = "；".join(relation_lines) if relation_lines else "无" - magic_stone_info = str(self.magic_stone) - - from src.classes.sect import get_sect_info_with_rank - - if detailed: - weapon_info = f"{self.weapon.get_detailed_info()}，熟练度：{self.weapon_proficiency:.1f}%" - auxiliary_info = self.auxiliary.get_detailed_info() if self.auxiliary is not None else "无" - sect_info = get_sect_info_with_rank(self, detailed=True) - alignment_info = self.alignment.get_detailed_info() if self.alignment is not None else "未知" - region_info = region.get_detailed_info() if region is not None else "无" - root_info = self.root.get_detailed_info() - technique_info = self.technique.get_detailed_info() if self.technique is not None else "无" - cultivation_info = self.cultivation_progress.get_detailed_info() - personas_info = ", ".join([p.get_detailed_info() for p in self.personas]) if self.personas else "无" - items_info = "，".join([f"{item.get_detailed_info()}x{quantity}" for item, quantity in self.items.items()]) if self.items else "无" - appearance_info = self.appearance.get_detailed_info(self.gender) - spirit_animal_info = self.spirit_animal.get_info() if self.spirit_animal is not None else "无" - else: - weapon_info = self.weapon.get_info() if self.weapon is not None else "无" - auxiliary_info = self.auxiliary.get_info() if self.auxiliary is not None else "无" - # 宗门信息：非详细模式下只显示"宗门名+职位" - sect_info = get_sect_info_with_rank(self, detailed=False) - region_info = region.get_info() if region is not None else "无" - alignment_info = self.alignment.get_info() if self.alignment is not None else "未知" - root_info = self.root.get_info() - technique_info = self.technique.get_info() if self.technique is not None else "无" - cultivation_info = self.cultivation_progress.get_info() - personas_info = ", ".join([p.get_detailed_info() for p in self.personas]) if self.personas else "无" - items_info = "，".join([f"{item.get_info()}x{quantity}" for item, quantity in self.items.items()]) if self.items else "无" - appearance_info = self.appearance.get_info() - spirit_animal_info = self.spirit_animal.get_info() if self.spirit_animal is not None else "无" - - info_dict = { - "名字": self.name, - "性别": str(self.gender), - "年龄": str(self.age), - "hp": str(self.hp), - "灵石": magic_stone_info, - "关系": relations_info, - "宗门": sect_info, - "阵营": alignment_info, - "地区": region_info, - "灵根": root_info, - "功法": technique_info, - "境界": cultivation_info, - "特质": personas_info, - "物品": items_info, - "外貌": appearance_info, - "兵器": weapon_info, - "辅助装备": auxiliary_info, - } - # 绰号：仅在存在时显示 - if self.nickname is not None: - info_dict["绰号"] = self.nickname.value - # 灵兽：仅在存在时显示 - if self.spirit_animal is not None: - info_dict["灵兽"] = spirit_animal_info - # 长期目标：仅在存在时显示 - if self.long_term_objective is not None: - info_dict["长期目标"] = self.long_term_objective.content - # 短期目标：仅在存在时显示 - if self.short_term_objective: - info_dict["短期目标"] = self.short_term_objective - return info_dict - - def get_structured_info(self) -> dict: - """ - 获取结构化的角色信息，用于前端展示和交互。 - """ - # 基础信息 - info = { - "id": self.id, - "name": self.name, - "gender": str(self.gender), - "age": self.age.age, - "lifespan": self.age.max_lifespan, - "realm": self.cultivation_progress.realm.value, - "level": self.cultivation_progress.level, - "hp": {"cur": self.hp.cur, "max": self.hp.max}, - "alignment": str(self.alignment) if self.alignment else "未知", - "magic_stone": self.magic_stone.value, - "thinking": self.thinking, - "short_term_objective": self.short_term_objective, - "long_term_objective": self.long_term_objective.content if self.long_term_objective else "", - "nickname": self.nickname.value if self.nickname else None, - "nickname_reason": self.nickname.reason if self.nickname else None, - "is_dead": self.is_dead, - "death_info": self.death_info, - } - - # 复杂对象结构化 - - # 1. 特质 (Personas) - info["personas"] = [p.get_structured_info() for p in self.personas] - - # 2. 功法 (Technique) - if self.technique: - info["technique"] = self.technique.get_structured_info() - else: - info["technique"] = None - - # 3. 宗门 (Sect) - if self.sect: - sect_info = self.sect.get_structured_info() - # 补充职位信息 - if self.sect_rank: - from src.classes.sect_ranks import get_rank_display_name - sect_info["rank"] = get_rank_display_name(self.sect_rank, self.sect) - else: - sect_info["rank"] = "弟子" - info["sect"] = sect_info - else: - info["sect"] = None - - # 补充：阵营详情 - from src.classes.alignment import alignment_infos, alignment_strs - # 保持 alignment 字段为 string (value) 兼容现有逻辑 - info["alignment"] = str(self.alignment) if self.alignment else "未知" - if self.alignment: - cn_name = alignment_strs.get(self.alignment, self.alignment.value) - desc = alignment_infos.get(self.alignment, "") - info["alignment_detail"] = { - "name": cn_name, - "desc": desc, - } - - # 4. 装备 (Weapon & Auxiliary) - if self.weapon: - w_info = self.weapon.get_structured_info() - w_info["proficiency"] = f"{self.weapon_proficiency:.1f}%" - info["weapon"] = w_info - else: - info["weapon"] = None - - if self.auxiliary: - info["auxiliary"] = self.auxiliary.get_structured_info() - else: - info["auxiliary"] = None - - # 5. 物品 (Items) - items_list = [] - for item, count in self.items.items(): - i_info = item.get_structured_info() - i_info["count"] = count - items_list.append(i_info) - info["items"] = items_list - - # 6. 关系 (Relations) - relations_list = [] - for other, relation in self.relations.items(): - relations_list.append({ - "target_id": other.id, - "name": other.name, - "relation": get_relation_label(relation, self, other), - # 可以加更多 info，比如境界，用于列表中展示 - "realm": other.cultivation_progress.realm.value, - "sect": other.sect.name if other.sect else "散修" - }) - info["relations"] = relations_list - - # 7. 外貌 - info["appearance"] = self.appearance.get_info() - - # 8. 灵根 - from src.classes.root import format_root_cn - root_str = format_root_cn(self.root) - info["root"] = root_str - info["root_detail"] = { - "name": root_str, - "desc": f"包含元素：{'、'.join(str(e) for e in self.root.elements)}", - "effect_desc": self.root.effect_desc - } - - # 9. 灵兽 - if self.spirit_animal: - info["spirit_animal"] = self.spirit_animal.get_structured_info() - - return info - - def __str__(self) -> str: - return str(self.get_info(detailed=False)) - - def create_action(self, action_name: ACTION_NAME) -> Action: - """ - 根据动作名称创建新的action实例 - - Args: - action_name: 动作类的名称（如 'Cultivate', 'Breakthrough' 等） - - Returns: - 新创建的Action实例 - - Raises: - ValueError: 如果找不到对应的动作类 - """ - action_cls = ActionRegistry.get(action_name) - return action_cls(self, self.world) - - def load_decide_result_chain(self, action_name_params_pairs: ACTION_NAME_PARAMS_PAIRS, avatar_thinking: str, short_term_objective: str, prepend: bool = False): - """ - 加载AI的决策结果（动作链），立即设置第一个为当前动作，其余进入队列。 - - Args: - action_name_params_pairs: 动作名和参数对列表 - avatar_thinking: 思考内容 - short_term_objective: 短期目标 - prepend: 是否插队到最前面（默认False，即追加到末尾） - """ - if not action_name_params_pairs: - return - self.thinking = avatar_thinking - self.short_term_objective = short_term_objective - # 转为计划并入队（不立即提交，交由提交阶段统一触发开始事件） - plans: List[ActionPlan] = [ActionPlan(name, params) for name, params in action_name_params_pairs] - if prepend: - self.planned_actions[0:0] = plans - else: - self.planned_actions.extend(plans) - - def clear_plans(self) -> None: - self.planned_actions.clear() - - def has_plans(self) -> bool: - return len(self.planned_actions) > 0 - - def commit_next_plan(self) -> Optional[Event]: - """ - 提交下一个可启动的计划为当前动作；返回开始事件（若有）。 - """ - if self.current_action is not None: - return None - while self.planned_actions: - plan = self.planned_actions.pop(0) - try: - action = self.create_action(plan.action_name) - except Exception as e: - logger = get_logger().logger - logger.warning("非法动作: Avatar(name=%s,id=%s) 的动作 %s 参数=%s 无法启动，原因=%s", self.name, self.id, plan.action_name, plan.params, e) - continue - # 再验证 - params_for_can_start = filter_kwargs_for_callable(action.can_start, plan.params) - can_start, reason = action.can_start(**params_for_can_start) - if not can_start: - # 记录不合法动作 - logger = get_logger().logger - logger.warning("非法动作: Avatar(name=%s,id=%s) 的动作 %s 参数=%s 无法启动，原因=%s", self.name, self.id, plan.action_name, plan.params, reason) - continue - # 启动 - params_for_start = filter_kwargs_for_callable(action.start, plan.params) - start_event = action.start(**params_for_start) - self.current_action = ActionInstance(action=action, params=plan.params, status="running") - # 标记为“本轮新设动作”，用于本月补充执行 - self._new_action_set_this_step = True - return start_event - return None - - def peek_next_plan(self) -> Optional[ActionPlan]: - if not self.planned_actions: - return None - return self.planned_actions[0] - - async def tick_action(self) -> List[Event]: - """ - 推进当前动作一步；返回过程中由动作内部产生的事件（通过 add_event 收集）。 - """ - if self.current_action is None: - return [] - # 记录当前动作实例引用，用于检测执行过程中是否发生了"抢占/切换" - action_instance_before = self.current_action - action = action_instance_before.action - params = action_instance_before.params - params_for_step = filter_kwargs_for_callable(action.step, params) - result: ActionResult = action.step(**params_for_step) - if result.status == ActionStatus.COMPLETED: - params_for_finish = filter_kwargs_for_callable(action.finish, params) - finish_events = await action.finish(**params_for_finish) - # 仅当当前动作仍然是刚才执行的那个实例时才清空 - # 若在 step() 内部通过"抢占"机制切换了动作（如 Escape 失败立即切到 Attack），不要清空新动作 - if self.current_action is action_instance_before: - self.current_action = None - # 动作完成后，如果有待执行计划，立即提交下一个（支持同月链式执行） - if self.has_plans(): - start_event = self.commit_next_plan() - if start_event is not None: - self._pending_events.append(start_event) - if finish_events: - # 允许 finish 直接返回事件（极少用），统一并入 pending - for e in finish_events: - self._pending_events.append(e) - # 合并动作返回的事件（通常为空） - if result.events: - for e in result.events: - self._pending_events.append(e) - events, self._pending_events = self._pending_events, [] - # 本轮已执行过，清除"新设动作"标记（但如果刚刚提交了新动作，commit_next_plan会重新设置为True） - if self.current_action is None: - # 当前无动作时才清除标记，避免清除新提交动作的标记 - self._new_action_set_this_step = False - - return events - - def update_cultivation(self, new_level: int): - """ - 更新修仙进度，并在境界提升时更新寿命和宗门职位 - """ - old_realm = self.cultivation_progress.realm - self.cultivation_progress.level = new_level - self.cultivation_progress.realm = self.cultivation_progress.get_realm(new_level) - - # 如果境界提升了，更新寿命期望和长期效果 - if self.cultivation_progress.realm != old_realm: - self.age.update_realm(self.cultivation_progress.realm) - # 境界变化会影响 HP/MP 基础值，需要重新计算 - self.recalc_effects() - # 如果有宗门，检查是否需要晋升职位 - from src.classes.sect_ranks import check_and_promote_sect_rank - check_and_promote_sect_rank(self, old_realm, self.cultivation_progress.realm) - - def death_by_old_age(self) -> bool: - """ - 检查是否老死 - - 返回: - 如果老死返回True，否则返回False - """ - return self.age.death_by_old_age(self.cultivation_progress.realm) - - def update_age(self, current_month_stamp: MonthStamp): - """ - 更新年龄 - """ - self.age.update_age(current_month_stamp, self.birth_month_stamp) - - - def is_in_region(self, region: Region|None) -> bool: - current_region = self.tile.region - if current_region is None: - tile = self.world.map.get_tile(self.pos_x, self.pos_y) - current_region = tile.region - return current_region == region - - def add_item(self, item: Item, quantity: int = 1) -> None: - """ - 添加物品到背包 - - Args: - item: 要添加的物品 - quantity: 添加数量，默认为1 - """ - if quantity <= 0: - return - - if item in self.items: - self.items[item] += quantity - else: - self.items[item] = quantity - - def remove_item(self, item: Item, quantity: int = 1) -> bool: - """ - 从背包移除物品 - - Args: - item: 要移除的物品 - quantity: 移除数量，默认为1 - - Returns: - bool: 是否成功移除（如果物品不足则返回False） - """ - if quantity <= 0: - return True - - if item not in self.items: - return False - - if self.items[item] < quantity: - return False - - self.items[item] -= quantity - - # 如果数量为0，从字典中移除该物品 - if self.items[item] == 0: - del self.items[item] - - return True - - def has_item(self, item: Item, quantity: int = 1) -> bool: - """ - 检查是否拥有足够数量的物品 - - Args: - item: 要检查的物品 - quantity: 需要的数量，默认为1 - - Returns: - bool: 是否拥有足够数量的物品 - """ - return item in self.items and self.items[item] >= quantity - - def get_item_quantity(self, item: Item) -> int: - """ - 获取指定物品的数量 - - Args: - item: 要查询的物品 - - Returns: - int: 物品数量，如果没有该物品则返回0 - """ - return self.items.get(item, 0) - - def add_event(self, event: Event, *, to_sidebar: bool = True) -> None: - """ - 添加事件： - - to_sidebar: 是否进入全局侧边栏（通过 Avatar._pending_events 暂存） - - 注意：事件会先存入_pending_events，统一由Simulator写入event_manager，避免重复 - """ - if to_sidebar: - self._pending_events.append(event) - - - def get_expanded_info( - self, - co_region_avatars: Optional[List["Avatar"]] = None, - other_avatar: Optional["Avatar"] = None, - detailed: bool = False - ) -> dict: - """ - 获取角色的扩展信息，包含基础信息、观察到的角色和事件历史。 - - Args: - co_region_avatars: 同区域的其他角色列表，用于"观察到的角色"字段 - other_avatar: 另一个角色，如果提供则返回两人共同经历的事件，否则返回单人事件 - detailed: 是否返回详细信息 - """ - info = self.get_info(detailed=detailed) - - observed: list[str] = [] - if co_region_avatars: - for other in co_region_avatars[:8]: - observed.append(f"{other.name}，境界：{other.cultivation_progress.get_info()}") - - # 历史事件改为从全局事件管理器分类查询 - em = self.world.event_manager - major_limit = CONFIG.social.major_event_context_num - minor_limit = CONFIG.social.minor_event_context_num - - # 根据是否提供 other_avatar 决定获取单人事件还是双人共同事件 - if other_avatar is not None: - major_events = em.get_major_events_between(self.id, other_avatar.id, limit=major_limit) - minor_events = em.get_minor_events_between(self.id, other_avatar.id, limit=minor_limit) - else: - major_events = em.get_major_events_by_avatar(self.id, limit=major_limit) - minor_events = em.get_minor_events_by_avatar(self.id, limit=minor_limit) - - major_list = [str(e) for e in major_events] - minor_list = [str(e) for e in minor_events] - - info["周围角色"] = observed - info["重大事件"] = major_list - info["短期事件"] = minor_list - info["长期目标"] = self.long_term_objective.content if self.long_term_objective else "无" - return info - - def get_hover_info(self) -> list[str]: - """ - 返回用于前端悬浮提示的多行信息。 - """ - def add_kv(lines: list[str], key: str, value: object) -> None: - lines.append(f"{key}: {value}") - - def add_section(lines: list[str], title: str, body: list[str]) -> None: - lines.append("") - lines.append(f"{title}:") - lines.extend(body) - - lines: list[str] = [] - # 基础信息 - if self.nickname: - add_kv(lines, "绰号", f"「{self.nickname.value}」") - - add_kv(lines, "性别", self.gender) - add_kv(lines, "年龄", self.age) - add_kv(lines, "外貌", self.appearance.get_info()) - add_kv(lines, "阵营", self.alignment) - add_kv(lines, "境界", str(self.cultivation_progress)) - add_kv(lines, "HP", self.hp) - add_kv(lines, "战斗力", int(get_base_strength(self))) - add_kv(lines, "宗门", self.get_sect_str()) - - from src.classes.root import format_root_cn - add_kv(lines, "灵根", format_root_cn(self.root)) - - tech_str = self.technique.get_colored_info() if self.technique is not None else "无" - add_kv(lines, "功法", tech_str) - - if self.personas: - persona_parts = [p.get_colored_info() for p in self.personas] - add_kv(lines, "特质", ", ".join(persona_parts)) - - add_kv(lines, "灵石", str(self.magic_stone)) - - # 物品 - if self.items: - items_lines = [f" {item.name} x{quantity}" for item, quantity in self.items.items()] - add_section(lines, "物品", items_lines) - else: - add_kv(lines, "物品", "无") - - # 思考与目标 - if self.thinking: - add_section(lines, "思考", [self.thinking]) - if self.long_term_objective: - add_section(lines, "长期目标", [self.long_term_objective.content]) - if self.short_term_objective: - add_section(lines, "短期目标", [self.short_term_objective]) - - # 兵器（必有，使用颜色标记等级） - if self.weapon is not None: - weapon_text = self.weapon.get_colored_info() - if self.weapon.desc: - weapon_text += f"（{self.weapon.desc}）" - add_kv(lines, "兵器", weapon_text) - - # 辅助装备（可选，使用颜色标记等级） - if self.auxiliary is not None: - auxiliary_text = self.auxiliary.get_colored_info() - if self.auxiliary.desc: - auxiliary_text += f"（{self.auxiliary.desc}）" - add_kv(lines, "辅助装备", auxiliary_text) - else: - add_kv(lines, "辅助装备", "无") - - # 灵兽：仅在存在时显示 - if self.spirit_animal is not None: - add_kv(lines, "灵兽", self.spirit_animal.get_info()) - - # 关系（从自身视角分组展示） - from src.classes.relation import get_relations_strs - relation_lines = get_relations_strs(self, max_lines=15) - if relation_lines: - add_section(lines, "关系", [f" {s}" for s in relation_lines]) - else: - add_kv(lines, "关系", "无") - - return lines - - def get_sect_str(self) -> str: - """ - 获取宗门显示名：有宗门则返回"宗门名+职位"，否则返回"散修"。 - 例如："合欢宗长老"、"散修" - """ - if self.sect is None: - return "散修" - - # 有宗门但无职位（理论上不应该出现，兜底处理） - if self.sect_rank is None: - return self.sect.name - - from src.classes.sect_ranks import get_rank_display_name - rank_name = get_rank_display_name(self.sect_rank, self.sect) - return f"{self.sect.name}{rank_name}" - - def get_sect_rank_name(self) -> str: - """ - 获取宗门职位的显示名称 - """ - if self.sect is None or self.sect_rank is None: - return "散修" - - from src.classes.sect_ranks import get_rank_display_name - return get_rank_display_name(self.sect_rank, self.sect) - - def set_relation(self, other: "Avatar", relation: Relation) -> None: - """ - 设置与另一个角色的关系。 - 委托给 relations.py 中的函数。 - """ - from src.classes.relations import set_relation - set_relation(self, other, relation) - - def get_relation(self, other: "Avatar") -> Optional[Relation]: - """ - 获取与另一个角色的关系。 - 委托给 relations.py 中的函数。 - """ - from src.classes.relations import get_relation - return get_relation(self, other) - - def clear_relation(self, other: "Avatar") -> None: - """ - 清除与另一个角色的关系。 - 委托给 relations.py 中的函数。 - """ - from src.classes.relations import clear_relation - clear_relation(self, other) - - - def get_co_region_avatars(self, avatars: List["Avatar"]) -> List["Avatar"]: - """ - 返回与自己处于同一区域的角色列表（不含自己）。 - """ - if self.tile is None: - return [] - same_region: list[Avatar] = [] - for other in avatars: - if other is self or other.tile is None: - continue - if other.tile.region == self.tile.region: - same_region.append(other) - return same_region - - def get_other_avatar_info(self, other_avatar: "Avatar") -> str: - """ - 仅显示几个字段：名字、绰号、境界、关系、宗门、阵营、外貌、功法、武器、辅助装备、HP - """ - nickname = other_avatar.nickname.value if other_avatar.nickname else "无" - sect = other_avatar.sect.name if other_avatar.sect else "散修" - tech = other_avatar.technique.get_info() if other_avatar.technique else "无" - weapon = other_avatar.weapon.get_info() if other_avatar.weapon else "无" - aux = other_avatar.auxiliary.get_info() if other_avatar.auxiliary else "无" - alignment = other_avatar.alignment - - # 关系可能为空 - relation = self.get_relation(other_avatar) or "无" - - return ( - f"{other_avatar.name}，绰号：{nickname}，境界：{other_avatar.cultivation_progress.get_info()}，" - f"关系：{relation}，宗门：{sect}，阵营：{alignment}，" - f"外貌：{other_avatar.appearance.get_info()}，功法：{tech}，兵器：{weapon}，辅助：{aux}，HP：{other_avatar.hp}" - ) - - def update_time_effect(self) -> None: - """ - 随时间更新的被动效果。 - 当前实现：当 HP 未满时，回复最大生命值的 1%（受HP恢复速率加成影响）。 - """ - if self.hp.cur < self.hp.max: - base_recover = self.hp.max * 0.01 - - # 应用HP恢复速率加成 - recovery_rate_raw = self.effects.get("extra_hp_recovery_rate", 0.0) - recovery_rate_multiplier = 1.0 + float(recovery_rate_raw or 0.0) - - recover_amount = int(base_recover * recovery_rate_multiplier) - self.hp.recover(recover_amount) - - @property - def move_step_length(self) -> int: - """ - 获取角色的移动步长 - """ - return self.cultivation_progress.get_move_step() - - def recalc_effects(self) -> None: - """ - 重新计算所有长期效果 - 在装备更换、突破境界等情况下调用 - - 说明： - - self.effects 是 @property，每次访问都会重新 merge 所有来源的 effects - - 包括：宗门、功法、灵根、特质、兵器、辅助装备、灵兽 - - 也会重新计算动态表达式（如 eval(...)） - - 当前包括： - - HP 最大值 - - 寿命最大值 - """ - # 计算基础最大值（基于境界） - base_max_hp = HP_MAX_BY_REALM.get(self.cultivation_progress.realm, 100) - - # 访问 self.effects 会触发 @property，重新 merge 所有 effects - effects = self.effects - extra_max_hp = int(effects.get("extra_max_hp", 0)) - extra_max_lifespan = int(effects.get("extra_max_lifespan", 0)) - - # 计算新的最大值 - new_max_hp = base_max_hp + extra_max_hp - - # 更新最大值 - self.hp.max = new_max_hp - - # 更新寿命 - # 如果 effects 中有额外寿命加成，需要加到 base_max_lifespan 上吗？ - # 不，base_max_lifespan 是基于境界和年龄计算的基础值（裸值）。 - # max_lifespan 是最终值，应该是 base + extra。 - # 但是 Age 类内部逻辑是：set_base -> update max (max = base)。 - # 所以我们需要显式设置 max_lifespan = base + extra - - if self.age: - self.age.max_lifespan = self.age.base_max_lifespan + extra_max_lifespan - - # 调整当前值（不超过新的最大值） - if self.hp.cur > new_max_hp: - self.hp.cur = new_max_hp - - def change_weapon(self, new_weapon: Weapon) -> None: - """ - 更换兵器，熟练度归零，并重新计算长期效果 - - Args: - new_weapon: 新的兵器 - """ - self.weapon = new_weapon - self.weapon_proficiency = 0.0 - self.recalc_effects() - - def change_auxiliary(self, new_auxiliary: Optional[Auxiliary]) -> None: - """ - 更换辅助装备，并重新计算长期效果 - - Args: - new_auxiliary: 新的辅助装备（可为 None 表示卸下） - """ - self.auxiliary = new_auxiliary - self.recalc_effects() - - def increase_weapon_proficiency(self, amount: float) -> None: - """ - 增加兵器熟练度，上限100 - - Args: - amount: 增加的熟练度值 - """ - # 应用extra_weapon_proficiency_gain效果（倍率加成） - gain_multiplier = 1.0 + self.effects.get("extra_weapon_proficiency_gain", 0.0) - actual_amount = amount * gain_multiplier - self.weapon_proficiency = min(100.0, self.weapon_proficiency + actual_amount) - - diff --git a/src/classes/avatar/__init__.py b/src/classes/avatar/__init__.py new file mode 100644 index 0000000..9392bde --- /dev/null +++ b/src/classes/avatar/__init__.py @@ -0,0 +1,34 @@ +""" +Avatar 模块 + +将原 avatar.py 拆分为多个子模块，通过此 __init__.py 导出以保持向后兼容。 +""" +from src.classes.avatar.core import ( + Avatar, + Gender, + gender_strs, + MAX_HISTORY_EVENTS, +) + +from src.classes.avatar.info_presenter import ( + get_avatar_info, + get_avatar_structured_info, + get_avatar_hover_info, + get_avatar_expanded_info, + get_other_avatar_info, +) + +__all__ = [ + # 核心类 + "Avatar", + "Gender", + "gender_strs", + "MAX_HISTORY_EVENTS", + # 信息展示函数 + "get_avatar_info", + "get_avatar_structured_info", + "get_avatar_hover_info", + "get_avatar_expanded_info", + "get_other_avatar_info", +] + diff --git a/src/classes/avatar/action_mixin.py b/src/classes/avatar/action_mixin.py new file mode 100644 index 0000000..8779004 --- /dev/null +++ b/src/classes/avatar/action_mixin.py @@ -0,0 +1,158 @@ +""" +Avatar 动作管理 Mixin +""" +from __future__ import annotations + +from typing import TYPE_CHECKING, Optional, List + +if TYPE_CHECKING: + from src.classes.avatar.core import Avatar + +from src.classes.action import Action +from src.classes.action_runtime import ActionStatus, ActionResult, ActionPlan, ActionInstance +from src.classes.action.registry import ActionRegistry +from src.classes.event import Event +from src.classes.typings import ACTION_NAME, ACTION_NAME_PARAMS_PAIRS +from src.utils.params import filter_kwargs_for_callable +from src.run.log import get_logger + + +class ActionMixin: + """动作管理相关方法""" + + def create_action(self: "Avatar", action_name: ACTION_NAME) -> Action: + """ + 根据动作名称创建新的action实例 + + Args: + action_name: 动作类的名称（如 'Cultivate', 'Breakthrough' 等） + + Returns: + 新创建的Action实例 + + Raises: + ValueError: 如果找不到对应的动作类 + """ + action_cls = ActionRegistry.get(action_name) + return action_cls(self, self.world) + + def load_decide_result_chain( + self: "Avatar", + action_name_params_pairs: ACTION_NAME_PARAMS_PAIRS, + avatar_thinking: str, + short_term_objective: str, + prepend: bool = False + ): + """ + 加载AI的决策结果（动作链），立即设置第一个为当前动作，其余进入队列。 + + Args: + action_name_params_pairs: 动作名和参数对列表 + avatar_thinking: 思考内容 + short_term_objective: 短期目标 + prepend: 是否插队到最前面（默认False，即追加到末尾） + """ + if not action_name_params_pairs: + return + self.thinking = avatar_thinking + self.short_term_objective = short_term_objective + # 转为计划并入队（不立即提交，交由提交阶段统一触发开始事件） + plans: List[ActionPlan] = [ActionPlan(name, params) for name, params in action_name_params_pairs] + if prepend: + self.planned_actions[0:0] = plans + else: + self.planned_actions.extend(plans) + + def clear_plans(self: "Avatar") -> None: + self.planned_actions.clear() + + def has_plans(self: "Avatar") -> bool: + return len(self.planned_actions) > 0 + + def commit_next_plan(self: "Avatar") -> Optional[Event]: + """ + 提交下一个可启动的计划为当前动作；返回开始事件（若有）。 + """ + if self.current_action is not None: + return None + while self.planned_actions: + plan = self.planned_actions.pop(0) + try: + action = self.create_action(plan.action_name) + except Exception as e: + logger = get_logger().logger + logger.warning( + "非法动作: Avatar(name=%s,id=%s) 的动作 %s 参数=%s 无法启动，原因=%s", + self.name, self.id, plan.action_name, plan.params, e + ) + continue + # 再验证 + params_for_can_start = filter_kwargs_for_callable(action.can_start, plan.params) + can_start, reason = action.can_start(**params_for_can_start) + if not can_start: + # 记录不合法动作 + logger = get_logger().logger + logger.warning( + "非法动作: Avatar(name=%s,id=%s) 的动作 %s 参数=%s 无法启动，原因=%s", + self.name, self.id, plan.action_name, plan.params, reason + ) + continue + # 启动 + params_for_start = filter_kwargs_for_callable(action.start, plan.params) + start_event = action.start(**params_for_start) + self.current_action = ActionInstance(action=action, params=plan.params, status="running") + # 标记为"本轮新设动作"，用于本月补充执行 + self._new_action_set_this_step = True + return start_event + return None + + def peek_next_plan(self: "Avatar") -> Optional[ActionPlan]: + if not self.planned_actions: + return None + return self.planned_actions[0] + + async def tick_action(self: "Avatar") -> List[Event]: + """ + 推进当前动作一步；返回过程中由动作内部产生的事件（通过 add_event 收集）。 + """ + if self.current_action is None: + return [] + # 记录当前动作实例引用，用于检测执行过程中是否发生了"抢占/切换" + action_instance_before = self.current_action + action = action_instance_before.action + params = action_instance_before.params + params_for_step = filter_kwargs_for_callable(action.step, params) + result: ActionResult = action.step(**params_for_step) + if result.status == ActionStatus.COMPLETED: + params_for_finish = filter_kwargs_for_callable(action.finish, params) + finish_events = await action.finish(**params_for_finish) + # 仅当当前动作仍然是刚才执行的那个实例时才清空 + # 若在 step() 内部通过"抢占"机制切换了动作（如 Escape 失败立即切到 Attack），不要清空新动作 + if self.current_action is action_instance_before: + self.current_action = None + if finish_events: + # 允许 finish 直接返回事件（极少用），统一并入 pending + for e in finish_events: + self._pending_events.append(e) + # 合并动作返回的事件（通常为空） + if result.events: + for e in result.events: + self._pending_events.append(e) + events, self._pending_events = self._pending_events, [] + # 本轮已执行过，清除"新设动作"标记（但如果刚刚提交了新动作，commit_next_plan会重新设置为True） + if self.current_action is None: + # 当前无动作时才清除标记，避免清除新提交动作的标记 + self._new_action_set_this_step = False + + return events + + def add_event(self: "Avatar", event: Event, *, to_sidebar: bool = True) -> None: + """ + 添加事件： + - to_sidebar: 是否进入全局侧边栏（通过 Avatar._pending_events 暂存） + + 注意：事件会先存入_pending_events，统一由Simulator写入event_manager，避免重复 + """ + if to_sidebar: + self._pending_events.append(event) + diff --git a/src/classes/avatar/core.py b/src/classes/avatar/core.py new file mode 100644 index 0000000..039cdad --- /dev/null +++ b/src/classes/avatar/core.py @@ -0,0 +1,311 @@ +""" +Avatar 核心类 + +精简后的 Avatar 类，通过 Mixin 组合完整功能。 +""" +import random +from dataclasses import dataclass, field +from enum import Enum +from typing import Optional, List, TYPE_CHECKING + +if TYPE_CHECKING: + from src.classes.sect_ranks import SectRank + +from src.classes.calendar import MonthStamp +from src.classes.world import World +from src.sim.save.avatar_save_mixin import AvatarSaveMixin +from src.sim.load.avatar_load_mixin import AvatarLoadMixin +from src.classes.tile import Tile +from src.classes.region import Region +from src.classes.cultivation import CultivationProgress +from src.classes.root import Root +from src.classes.technique import Technique, get_technique_by_sect +from src.classes.age import Age +from src.classes.event import Event +from src.classes.action_runtime import ActionPlan, ActionInstance +from src.classes.alignment import Alignment +from src.classes.persona import Persona, get_random_compatible_personas +from src.classes.item import Item +from src.classes.weapon import Weapon +from src.classes.auxiliary import Auxiliary +from src.classes.magic_stone import MagicStone +from src.classes.hp_and_mp import HP, HP_MAX_BY_REALM +from src.classes.relation import Relation +from src.classes.sect import Sect +from src.classes.appearance import Appearance, get_random_appearance +from src.classes.spirit_animal import SpiritAnimal +from src.classes.long_term_objective import LongTermObjective +from src.classes.nickname_data import Nickname +from src.utils.config import CONFIG + +# Mixin 导入 +from src.classes.avatar.effects_mixin import EffectsMixin +from src.classes.avatar.inventory_mixin import InventoryMixin +from src.classes.avatar.action_mixin import ActionMixin + +persona_num = CONFIG.avatar.persona_num + + +class Gender(Enum): + MALE = "male" + FEMALE = "female" + + def __str__(self) -> str: + return gender_strs.get(self, self.value) + + +gender_strs = { + Gender.MALE: "男", + Gender.FEMALE: "女", +} + +# 历史事件的最大数量 +MAX_HISTORY_EVENTS = 10 + + +@dataclass +class Avatar( + AvatarSaveMixin, + AvatarLoadMixin, + EffectsMixin, + InventoryMixin, + ActionMixin, +): + """ + NPC的类。 + 包含了这个角色的一切信息。 + """ + world: World + name: str + id: str + birth_month_stamp: MonthStamp + age: Age + gender: Gender + cultivation_progress: CultivationProgress = field(default_factory=lambda: CultivationProgress(0)) + pos_x: int = 0 + pos_y: int = 0 + tile: Optional[Tile] = None + + root: Root = field(default_factory=lambda: random.choice(list(Root))) + personas: List[Persona] = field(default_factory=list) + technique: Technique | None = None + history_events: List[Event] = field(default_factory=list) + _pending_events: List[Event] = field(default_factory=list) + current_action: Optional[ActionInstance] = None + planned_actions: List[ActionPlan] = field(default_factory=list) + thinking: str = "" + short_term_objective: str = "" + long_term_objective: Optional[LongTermObjective] = None + magic_stone: MagicStone = field(default_factory=lambda: MagicStone(0)) + items: dict[Item, int] = field(default_factory=dict) + hp: HP = field(default_factory=lambda: HP(0, 0)) + relations: dict["Avatar", Relation] = field(default_factory=dict) + alignment: Alignment | None = None + sect: Sect | None = None + sect_rank: "SectRank | None" = None + appearance: Appearance = field(default_factory=get_random_appearance) + weapon: Optional[Weapon] = None + weapon_proficiency: float = 0.0 + auxiliary: Optional[Auxiliary] = None + spirit_animal: Optional[SpiritAnimal] = None + nickname: Optional[Nickname] = None + custom_pic_id: Optional[int] = None + + is_dead: bool = False + death_info: Optional[dict] = None + + _new_action_set_this_step: bool = False + _action_cd_last_months: dict[str, int] = field(default_factory=dict) + + known_regions: set[int] = field(default_factory=set) + + # ========== 宗门相关 ========== + + def join_sect(self, sect: Sect, rank: "SectRank") -> None: + """加入宗门""" + if self.is_dead: + return + if self.sect: + self.leave_sect() + self.sect = sect + self.sect_rank = rank + sect.add_member(self) + + def leave_sect(self) -> None: + """退出宗门""" + if self.sect: + self.sect.remove_member(self) + self.sect = None + self.sect_rank = None + + def get_sect_str(self) -> str: + """获取宗门显示名：有宗门则返回"宗门名+职位"，否则返回"散修"。""" + if self.sect is None: + return "散修" + if self.sect_rank is None: + return self.sect.name + from src.classes.sect_ranks import get_rank_display_name + rank_name = get_rank_display_name(self.sect_rank, self.sect) + return f"{self.sect.name}{rank_name}" + + def get_sect_rank_name(self) -> str: + """获取宗门职位的显示名称""" + if self.sect is None or self.sect_rank is None: + return "散修" + from src.classes.sect_ranks import get_rank_display_name + return get_rank_display_name(self.sect_rank, self.sect) + + # ========== 死亡相关 ========== + + def set_dead(self, reason: str, time: MonthStamp) -> None: + """设置角色死亡状态。""" + if self.is_dead: + return + + self.is_dead = True + self.death_info = { + "time": int(time), + "reason": reason, + "location": (self.pos_x, self.pos_y) + } + + self.planned_actions.clear() + self.current_action = None + self._pending_events.clear() + self.thinking = "" + self.short_term_objective = "" + + if self.sect: + self.sect.remove_member(self) + + def death_by_old_age(self) -> bool: + """检查是否老死""" + return self.age.death_by_old_age(self.cultivation_progress.realm) + + # ========== 年龄与修为 ========== + + def update_age(self, current_month_stamp: MonthStamp): + """更新年龄""" + self.age.update_age(current_month_stamp, self.birth_month_stamp) + + def update_cultivation(self, new_level: int): + """更新修仙进度，并在境界提升时更新寿命和宗门职位""" + old_realm = self.cultivation_progress.realm + self.cultivation_progress.level = new_level + self.cultivation_progress.realm = self.cultivation_progress.get_realm(new_level) + + if self.cultivation_progress.realm != old_realm: + self.age.update_realm(self.cultivation_progress.realm) + self.recalc_effects() + from src.classes.sect_ranks import check_and_promote_sect_rank + check_and_promote_sect_rank(self, old_realm, self.cultivation_progress.realm) + + # ========== 区域与位置 ========== + + def is_in_region(self, region: Region | None) -> bool: + current_region = self.tile.region + if current_region is None: + tile = self.world.map.get_tile(self.pos_x, self.pos_y) + current_region = tile.region + return current_region == region + + def get_co_region_avatars(self, avatars: List["Avatar"]) -> List["Avatar"]: + """返回与自己处于同一区域的角色列表（不含自己）。""" + if self.tile is None: + return [] + same_region: list[Avatar] = [] + for other in avatars: + if other is self or other.tile is None: + continue + if other.tile.region == self.tile.region: + same_region.append(other) + return same_region + + def _init_known_regions(self): + """初始化已知区域：当前位置 + 宗门驻地""" + if self.tile and self.tile.region: + self.known_regions.add(self.tile.region.id) + + if self.sect: + for r in self.world.map.sect_regions.values(): + if r.sect_id == self.sect.id: + self.known_regions.add(r.id) + break + + # ========== 关系相关 ========== + + def set_relation(self, other: "Avatar", relation: Relation) -> None: + """设置与另一个角色的关系。""" + from src.classes.relations import set_relation + set_relation(self, other, relation) + + def get_relation(self, other: "Avatar") -> Optional[Relation]: + """获取与另一个角色的关系。""" + from src.classes.relations import get_relation + return get_relation(self, other) + + def clear_relation(self, other: "Avatar") -> None: + """清除与另一个角色的关系。""" + from src.classes.relations import clear_relation + clear_relation(self, other) + + # ========== 信息展示（委托） ========== + + def get_info(self, detailed: bool = False) -> dict: + from src.classes.avatar.info_presenter import get_avatar_info + return get_avatar_info(self, detailed) + + def get_structured_info(self) -> dict: + from src.classes.avatar.info_presenter import get_avatar_structured_info + return get_avatar_structured_info(self) + + def get_hover_info(self) -> list[str]: + from src.classes.avatar.info_presenter import get_avatar_hover_info + return get_avatar_hover_info(self) + + def get_expanded_info( + self, + co_region_avatars: Optional[List["Avatar"]] = None, + other_avatar: Optional["Avatar"] = None, + detailed: bool = False + ) -> dict: + from src.classes.avatar.info_presenter import get_avatar_expanded_info + return get_avatar_expanded_info(self, co_region_avatars, other_avatar, detailed) + + def get_other_avatar_info(self, other_avatar: "Avatar") -> str: + from src.classes.avatar.info_presenter import get_other_avatar_info + return get_other_avatar_info(self, other_avatar) + + # ========== 魔法方法 ========== + + def __post_init__(self): + """在Avatar创建后自动初始化tile和HP""" + self.tile = self.world.map.get_tile(self.pos_x, self.pos_y) + + max_hp = HP_MAX_BY_REALM.get(self.cultivation_progress.realm, 100) + self.hp = HP(max_hp, max_hp) + + if not self.personas: + self.personas = get_random_compatible_personas(persona_num, avatar=self) + + if self.technique is None: + self.technique = get_technique_by_sect(self.sect) + + if self.sect: + self.sect.add_member(self) + + if self.alignment is None: + if self.sect is not None: + self.alignment = self.sect.alignment + else: + self.alignment = random.choice(list(Alignment)) + + self.recalc_effects() + self._init_known_regions() + + def __hash__(self) -> int: + return hash(self.id) + + def __str__(self) -> str: + return str(self.get_info(detailed=False)) + diff --git a/src/classes/avatar/effects_mixin.py b/src/classes/avatar/effects_mixin.py new file mode 100644 index 0000000..b290a31 --- /dev/null +++ b/src/classes/avatar/effects_mixin.py @@ -0,0 +1,119 @@ +""" +Avatar 效果计算 Mixin + +负责角色效果的计算和应用。 +""" +from __future__ import annotations + +from collections import defaultdict +from typing import TYPE_CHECKING + +if TYPE_CHECKING: + from src.classes.avatar.core import Avatar + +from src.classes.effect import _merge_effects, _evaluate_conditional_effect +from src.classes.hp_and_mp import HP_MAX_BY_REALM + + +class EffectsMixin: + """效果计算相关方法""" + + @property + def effects(self: "Avatar") -> dict[str, object]: + """ + 合并所有来源的效果：宗门、功法、灵根、特质、兵器、辅助装备、灵兽、天地灵机 + """ + merged: dict[str, object] = defaultdict(str) + # 来自宗门 + if self.sect is not None: + evaluated = _evaluate_conditional_effect(self.sect.effects, self) + merged = _merge_effects(merged, evaluated) + # 来自功法 + evaluated = _evaluate_conditional_effect(self.technique.effects, self) + merged = _merge_effects(merged, evaluated) + # 来自灵根 + evaluated = _evaluate_conditional_effect(self.root.effects, self) + merged = _merge_effects(merged, evaluated) + # 来自特质（persona） + for persona in self.personas: + evaluated = _evaluate_conditional_effect(persona.effects, self) + merged = _merge_effects(merged, evaluated) + # 来自兵器 + if self.weapon is not None: + evaluated = _evaluate_conditional_effect(self.weapon.effects, self) + merged = _merge_effects(merged, evaluated) + # 来自辅助装备 + if self.auxiliary is not None: + evaluated = _evaluate_conditional_effect(self.auxiliary.effects, self) + merged = _merge_effects(merged, evaluated) + # 来自灵兽 + if self.spirit_animal is not None: + evaluated = _evaluate_conditional_effect(self.spirit_animal.effects, self) + merged = _merge_effects(merged, evaluated) + # 来自天地灵机（世界级buff/debuff） + if self.world.current_phenomenon is not None: + evaluated = _evaluate_conditional_effect(self.world.current_phenomenon.effects, self) + merged = _merge_effects(merged, evaluated) + # 评估动态效果表达式：值以 "eval(...)" 形式给出 + final: dict[str, object] = {} + for k, v in merged.items(): + if isinstance(v, str): + s = v.strip() + if s.startswith("eval(") and s.endswith(")"): + expr = s[5:-1] + final[k] = eval(expr, {"__builtins__": {}}, {"avatar": self}) + continue + final[k] = v + return final + + def recalc_effects(self: "Avatar") -> None: + """ + 重新计算所有长期效果 + 在装备更换、突破境界等情况下调用 + + 当前包括： + - HP 最大值 + - 寿命最大值 + """ + # 计算基础最大值（基于境界） + base_max_hp = HP_MAX_BY_REALM.get(self.cultivation_progress.realm, 100) + + # 访问 self.effects 会触发 @property，重新 merge 所有 effects + effects = self.effects + extra_max_hp = int(effects.get("extra_max_hp", 0)) + extra_max_lifespan = int(effects.get("extra_max_lifespan", 0)) + + # 计算新的最大值 + new_max_hp = base_max_hp + extra_max_hp + + # 更新最大值 + self.hp.max = new_max_hp + + # 更新寿命 + if self.age: + self.age.max_lifespan = self.age.base_max_lifespan + extra_max_lifespan + + # 调整当前值（不超过新的最大值） + if self.hp.cur > new_max_hp: + self.hp.cur = new_max_hp + + def update_time_effect(self: "Avatar") -> None: + """ + 随时间更新的被动效果。 + 当前实现：当 HP 未满时，回复最大生命值的 1%（受HP恢复速率加成影响）。 + """ + if self.hp.cur < self.hp.max: + base_recover = self.hp.max * 0.01 + + # 应用HP恢复速率加成 + recovery_rate_raw = self.effects.get("extra_hp_recovery_rate", 0.0) + recovery_rate_multiplier = 1.0 + float(recovery_rate_raw or 0.0) + + recover_amount = int(base_recover * recovery_rate_multiplier) + self.hp.recover(recover_amount) + + @property + def move_step_length(self: "Avatar") -> int: + """获取角色的移动步长""" + return self.cultivation_progress.get_move_step() + diff --git a/src/classes/avatar/info_presenter.py b/src/classes/avatar/info_presenter.py new file mode 100644 index 0000000..aae12cc --- /dev/null +++ b/src/classes/avatar/info_presenter.py @@ -0,0 +1,349 @@ +""" +Avatar 信息展示模块 + +将信息格式化逻辑从 Avatar 类中分离，作为独立函数提供。 +""" +from __future__ import annotations + +from typing import TYPE_CHECKING, Optional, List + +if TYPE_CHECKING: + from src.classes.avatar.core import Avatar + +from src.classes.battle import get_base_strength +from src.classes.relation import get_relation_label +from src.utils.config import CONFIG + + +def get_avatar_info(avatar: "Avatar", detailed: bool = False) -> dict: + """ + 获取 avatar 的信息，返回 dict；根据 detailed 控制信息粒度。 + """ + region = avatar.tile.region if avatar.tile is not None else None + from src.classes.relation import get_relations_strs + relation_lines = get_relations_strs(avatar, max_lines=8) + relations_info = "；".join(relation_lines) if relation_lines else "无" + magic_stone_info = str(avatar.magic_stone) + + from src.classes.sect import get_sect_info_with_rank + + if detailed: + weapon_info = f"{avatar.weapon.get_detailed_info()}，熟练度：{avatar.weapon_proficiency:.1f}%" + auxiliary_info = avatar.auxiliary.get_detailed_info() if avatar.auxiliary is not None else "无" + sect_info = get_sect_info_with_rank(avatar, detailed=True) + alignment_info = avatar.alignment.get_detailed_info() if avatar.alignment is not None else "未知" + region_info = region.get_detailed_info() if region is not None else "无" + root_info = avatar.root.get_detailed_info() + technique_info = avatar.technique.get_detailed_info() if avatar.technique is not None else "无" + cultivation_info = avatar.cultivation_progress.get_detailed_info() + personas_info = ", ".join([p.get_detailed_info() for p in avatar.personas]) if avatar.personas else "无" + items_info = "，".join([f"{item.get_detailed_info()}x{quantity}" for item, quantity in avatar.items.items()]) if avatar.items else "无" + appearance_info = avatar.appearance.get_detailed_info(avatar.gender) + spirit_animal_info = avatar.spirit_animal.get_info() if avatar.spirit_animal is not None else "无" + else: + weapon_info = avatar.weapon.get_info() if avatar.weapon is not None else "无" + auxiliary_info = avatar.auxiliary.get_info() if avatar.auxiliary is not None else "无" + sect_info = get_sect_info_with_rank(avatar, detailed=False) + region_info = region.get_info() if region is not None else "无" + alignment_info = avatar.alignment.get_info() if avatar.alignment is not None else "未知" + root_info = avatar.root.get_info() + technique_info = avatar.technique.get_info() if avatar.technique is not None else "无" + cultivation_info = avatar.cultivation_progress.get_info() + personas_info = ", ".join([p.get_detailed_info() for p in avatar.personas]) if avatar.personas else "无" + items_info = "，".join([f"{item.get_info()}x{quantity}" for item, quantity in avatar.items.items()]) if avatar.items else "无" + appearance_info = avatar.appearance.get_info() + spirit_animal_info = avatar.spirit_animal.get_info() if avatar.spirit_animal is not None else "无" + + info_dict = { + "名字": avatar.name, + "性别": str(avatar.gender), + "年龄": str(avatar.age), + "hp": str(avatar.hp), + "灵石": magic_stone_info, + "关系": relations_info, + "宗门": sect_info, + "阵营": alignment_info, + "地区": region_info, + "灵根": root_info, + "功法": technique_info, + "境界": cultivation_info, + "特质": personas_info, + "物品": items_info, + "外貌": appearance_info, + "兵器": weapon_info, + "辅助装备": auxiliary_info, + } + # 绰号：仅在存在时显示 + if avatar.nickname is not None: + info_dict["绰号"] = avatar.nickname.value + # 灵兽：仅在存在时显示 + if avatar.spirit_animal is not None: + info_dict["灵兽"] = spirit_animal_info + # 长期目标：仅在存在时显示 + if avatar.long_term_objective is not None: + info_dict["长期目标"] = avatar.long_term_objective.content + # 短期目标：仅在存在时显示 + if avatar.short_term_objective: + info_dict["短期目标"] = avatar.short_term_objective + return info_dict + + +def get_avatar_structured_info(avatar: "Avatar") -> dict: + """ + 获取结构化的角色信息，用于前端展示和交互。 + """ + # 基础信息 + info = { + "id": avatar.id, + "name": avatar.name, + "gender": str(avatar.gender), + "age": avatar.age.age, + "lifespan": avatar.age.max_lifespan, + "realm": avatar.cultivation_progress.realm.value, + "level": avatar.cultivation_progress.level, + "hp": {"cur": avatar.hp.cur, "max": avatar.hp.max}, + "alignment": str(avatar.alignment) if avatar.alignment else "未知", + "magic_stone": avatar.magic_stone.value, + "thinking": avatar.thinking, + "short_term_objective": avatar.short_term_objective, + "long_term_objective": avatar.long_term_objective.content if avatar.long_term_objective else "", + "nickname": avatar.nickname.value if avatar.nickname else None, + "nickname_reason": avatar.nickname.reason if avatar.nickname else None, + "is_dead": avatar.is_dead, + "death_info": avatar.death_info, + } + + # 1. 特质 (Personas) + info["personas"] = [p.get_structured_info() for p in avatar.personas] + + # 2. 功法 (Technique) + if avatar.technique: + info["technique"] = avatar.technique.get_structured_info() + else: + info["technique"] = None + + # 3. 宗门 (Sect) + if avatar.sect: + sect_info = avatar.sect.get_structured_info() + if avatar.sect_rank: + from src.classes.sect_ranks import get_rank_display_name + sect_info["rank"] = get_rank_display_name(avatar.sect_rank, avatar.sect) + else: + sect_info["rank"] = "弟子" + info["sect"] = sect_info + else: + info["sect"] = None + + # 补充：阵营详情 + from src.classes.alignment import alignment_infos, alignment_strs + info["alignment"] = str(avatar.alignment) if avatar.alignment else "未知" + if avatar.alignment: + cn_name = alignment_strs.get(avatar.alignment, avatar.alignment.value) + desc = alignment_infos.get(avatar.alignment, "") + info["alignment_detail"] = { + "name": cn_name, + "desc": desc, + } + + # 4. 装备 (Weapon & Auxiliary) + if avatar.weapon: + w_info = avatar.weapon.get_structured_info() + w_info["proficiency"] = f"{avatar.weapon_proficiency:.1f}%" + info["weapon"] = w_info + else: + info["weapon"] = None + + if avatar.auxiliary: + info["auxiliary"] = avatar.auxiliary.get_structured_info() + else: + info["auxiliary"] = None + + # 5. 物品 (Items) + items_list = [] + for item, count in avatar.items.items(): + i_info = item.get_structured_info() + i_info["count"] = count + items_list.append(i_info) + info["items"] = items_list + + # 6. 关系 (Relations) + relations_list = [] + for other, relation in avatar.relations.items(): + relations_list.append({ + "target_id": other.id, + "name": other.name, + "relation": get_relation_label(relation, avatar, other), + "realm": other.cultivation_progress.realm.value, + "sect": other.sect.name if other.sect else "散修" + }) + info["relations"] = relations_list + + # 7. 外貌 + info["appearance"] = avatar.appearance.get_info() + + # 8. 灵根 + from src.classes.root import format_root_cn + root_str = format_root_cn(avatar.root) + info["root"] = root_str + info["root_detail"] = { + "name": root_str, + "desc": f"包含元素：{'、'.join(str(e) for e in avatar.root.elements)}", + "effect_desc": avatar.root.effect_desc + } + + # 9. 灵兽 + if avatar.spirit_animal: + info["spirit_animal"] = avatar.spirit_animal.get_structured_info() + + return info + + +def get_avatar_hover_info(avatar: "Avatar") -> list[str]: + """ + 返回用于前端悬浮提示的多行信息。 + """ + def add_kv(lines: list[str], key: str, value: object) -> None: + lines.append(f"{key}: {value}") + + def add_section(lines: list[str], title: str, body: list[str]) -> None: + lines.append("") + lines.append(f"{title}:") + lines.extend(body) + + lines: list[str] = [] + # 基础信息 + if avatar.nickname: + add_kv(lines, "绰号", f"「{avatar.nickname.value}」") + + add_kv(lines, "性别", avatar.gender) + add_kv(lines, "年龄", avatar.age) + add_kv(lines, "外貌", avatar.appearance.get_info()) + add_kv(lines, "阵营", avatar.alignment) + add_kv(lines, "境界", str(avatar.cultivation_progress)) + add_kv(lines, "HP", avatar.hp) + add_kv(lines, "战斗力", int(get_base_strength(avatar))) + add_kv(lines, "宗门", avatar.get_sect_str()) + + from src.classes.root import format_root_cn + add_kv(lines, "灵根", format_root_cn(avatar.root)) + + tech_str = avatar.technique.get_colored_info() if avatar.technique is not None else "无" + add_kv(lines, "功法", tech_str) + + if avatar.personas: + persona_parts = [p.get_colored_info() for p in avatar.personas] + add_kv(lines, "特质", ", ".join(persona_parts)) + + add_kv(lines, "灵石", str(avatar.magic_stone)) + + # 物品 + if avatar.items: + items_lines = [f" {item.name} x{quantity}" for item, quantity in avatar.items.items()] + add_section(lines, "物品", items_lines) + else: + add_kv(lines, "物品", "无") + + # 思考与目标 + if avatar.thinking: + add_section(lines, "思考", [avatar.thinking]) + if avatar.long_term_objective: + add_section(lines, "长期目标", [avatar.long_term_objective.content]) + if avatar.short_term_objective: + add_section(lines, "短期目标", [avatar.short_term_objective]) + + # 兵器（必有，使用颜色标记等级） + if avatar.weapon is not None: + weapon_text = avatar.weapon.get_colored_info() + if avatar.weapon.desc: + weapon_text += f"（{avatar.weapon.desc}）" + add_kv(lines, "兵器", weapon_text) + + # 辅助装备（可选，使用颜色标记等级） + if avatar.auxiliary is not None: + auxiliary_text = avatar.auxiliary.get_colored_info() + if avatar.auxiliary.desc: + auxiliary_text += f"（{avatar.auxiliary.desc}）" + add_kv(lines, "辅助装备", auxiliary_text) + else: + add_kv(lines, "辅助装备", "无") + + # 灵兽：仅在存在时显示 + if avatar.spirit_animal is not None: + add_kv(lines, "灵兽", avatar.spirit_animal.get_info()) + + # 关系（从自身视角分组展示） + from src.classes.relation import get_relations_strs + relation_lines = get_relations_strs(avatar, max_lines=15) + if relation_lines: + add_section(lines, "关系", [f" {s}" for s in relation_lines]) + else: + add_kv(lines, "关系", "无") + + return lines + + +def get_avatar_expanded_info( + avatar: "Avatar", + co_region_avatars: Optional[List["Avatar"]] = None, + other_avatar: Optional["Avatar"] = None, + detailed: bool = False +) -> dict: + """ + 获取角色的扩展信息，包含基础信息、观察到的角色和事件历史。 + + Args: + avatar: 目标角色 + co_region_avatars: 同区域的其他角色列表，用于"观察到的角色"字段 + other_avatar: 另一个角色，如果提供则返回两人共同经历的事件，否则返回单人事件 + detailed: 是否返回详细信息 + """ + info = get_avatar_info(avatar, detailed=detailed) + + observed: list[str] = [] + if co_region_avatars: + for other in co_region_avatars[:8]: + observed.append(f"{other.name}，境界：{other.cultivation_progress.get_info()}") + + # 历史事件改为从全局事件管理器分类查询 + em = avatar.world.event_manager + major_limit = CONFIG.social.major_event_context_num + minor_limit = CONFIG.social.minor_event_context_num + + # 根据是否提供 other_avatar 决定获取单人事件还是双人共同事件 + if other_avatar is not None: + major_events = em.get_major_events_between(avatar.id, other_avatar.id, limit=major_limit) + minor_events = em.get_minor_events_between(avatar.id, other_avatar.id, limit=minor_limit) + else: + major_events = em.get_major_events_by_avatar(avatar.id, limit=major_limit) + minor_events = em.get_minor_events_by_avatar(avatar.id, limit=minor_limit) + + major_list = [str(e) for e in major_events] + minor_list = [str(e) for e in minor_events] + + info["周围角色"] = observed + info["重大事件"] = major_list + info["短期事件"] = minor_list + info["长期目标"] = avatar.long_term_objective.content if avatar.long_term_objective else "无" + return info + + +def get_other_avatar_info(from_avatar: "Avatar", to_avatar: "Avatar") -> str: + """ + 仅显示几个字段：名字、绰号、境界、关系、宗门、阵营、外貌、功法、武器、辅助装备、HP + """ + nickname = to_avatar.nickname.value if to_avatar.nickname else "无" + sect = to_avatar.sect.name if to_avatar.sect else "散修" + tech = to_avatar.technique.get_info() if to_avatar.technique else "无" + weapon = to_avatar.weapon.get_info() if to_avatar.weapon else "无" + aux = to_avatar.auxiliary.get_info() if to_avatar.auxiliary else "无" + alignment = to_avatar.alignment + + # 关系可能为空 + relation = from_avatar.get_relation(to_avatar) or "无" + + return ( + f"{to_avatar.name}，绰号：{nickname}，境界：{to_avatar.cultivation_progress.get_info()}，" + f"关系：{relation}，宗门：{sect}，阵营：{alignment}，" + f"外貌：{to_avatar.appearance.get_info()}，功法：{tech}，兵器：{weapon}，辅助：{aux}，HP：{to_avatar.hp}" + ) + diff --git a/src/classes/avatar/inventory_mixin.py b/src/classes/avatar/inventory_mixin.py new file mode 100644 index 0000000..ef1abd8 --- /dev/null +++ b/src/classes/avatar/inventory_mixin.py @@ -0,0 +1,119 @@ +""" +Avatar 物品与装备管理 Mixin +""" +from __future__ import annotations + +from typing import TYPE_CHECKING, Optional + +if TYPE_CHECKING: + from src.classes.avatar.core import Avatar + from src.classes.item import Item + from src.classes.weapon import Weapon + from src.classes.auxiliary import Auxiliary + + +class InventoryMixin: + """物品与装备管理相关方法""" + + def add_item(self: "Avatar", item: "Item", quantity: int = 1) -> None: + """ + 添加物品到背包 + + Args: + item: 要添加的物品 + quantity: 添加数量，默认为1 + """ + if quantity <= 0: + return + + if item in self.items: + self.items[item] += quantity + else: + self.items[item] = quantity + + def remove_item(self: "Avatar", item: "Item", quantity: int = 1) -> bool: + """ + 从背包移除物品 + + Args: + item: 要移除的物品 + quantity: 移除数量，默认为1 + + Returns: + bool: 是否成功移除（如果物品不足则返回False） + """ + if quantity <= 0: + return True + + if item not in self.items: + return False + + if self.items[item] < quantity: + return False + + self.items[item] -= quantity + + # 如果数量为0，从字典中移除该物品 + if self.items[item] == 0: + del self.items[item] + + return True + + def has_item(self: "Avatar", item: "Item", quantity: int = 1) -> bool: + """ + 检查是否拥有足够数量的物品 + + Args: + item: 要检查的物品 + quantity: 需要的数量，默认为1 + + Returns: + bool: 是否拥有足够数量的物品 + """ + return item in self.items and self.items[item] >= quantity + + def get_item_quantity(self: "Avatar", item: "Item") -> int: + """ + 获取指定物品的数量 + + Args: + item: 要查询的物品 + + Returns: + int: 物品数量，如果没有该物品则返回0 + """ + return self.items.get(item, 0) + + def change_weapon(self: "Avatar", new_weapon: "Weapon") -> None: + """ + 更换兵器，熟练度归零，并重新计算长期效果 + + Args: + new_weapon: 新的兵器 + """ + self.weapon = new_weapon + self.weapon_proficiency = 0.0 + self.recalc_effects() + + def change_auxiliary(self: "Avatar", new_auxiliary: Optional["Auxiliary"]) -> None: + """ + 更换辅助装备，并重新计算长期效果 + + Args: + new_auxiliary: 新的辅助装备（可为 None 表示卸下） + """ + self.auxiliary = new_auxiliary + self.recalc_effects() + + def increase_weapon_proficiency(self: "Avatar", amount: float) -> None: + """ + 增加兵器熟练度，上限100 + + Args: + amount: 增加的熟练度值 + """ + # 应用extra_weapon_proficiency_gain效果（倍率加成） + gain_multiplier = 1.0 + self.effects.get("extra_weapon_proficiency_gain", 0.0) + actual_amount = amount * gain_multiplier + self.weapon_proficiency = min(100.0, self.weapon_proficiency + actual_amount) + diff --git a/src/classes/battle.py b/src/classes/battle.py index c604511..3c33c00 100644 --- a/src/classes/battle.py +++ b/src/classes/battle.py @@ -2,12 +2,13 @@ from __future__ import annotations import math import random -from typing import Tuple, TYPE_CHECKING +from typing import Tuple, TYPE_CHECKING, Callable, Awaitable, Optional from src.classes.technique import TechniqueGrade, get_suppression_bonus if TYPE_CHECKING: from src.classes.avatar import Avatar + from src.classes.event import Event # 战斗力参数（参考文明6思想，但适配本项目数值体系） @@ -114,19 +115,6 @@ def _base_damage_scale(defender: "Avatar") -> float: max_hp = defender.hp.max return max(1.0, max_hp / 100.0) - -def _damage_from_to(attacker: "Avatar", defender: "Avatar") -> int: - """ - 使用 Civ6 风格伤害：damage = U(24,36)×scale × e^(K×差值) - - scale = defender.maxHP / 100，使不同境界下伤害相对一致 - - 差值 = strength(att) - strength(def) - """ - diff = _strength_diff(attacker, defender) - base = random.randint(_BASE_DAMAGE_LOW, _BASE_DAMAGE_HIGH) * _base_damage_scale(defender) - dmg = base * math.exp(_CIV6_K * diff) - return max(1, int(dmg)) - - def _damage_pair(winner: "Avatar", loser: "Avatar") -> tuple[int, int]: """ 成对伤害：使用同一基础与对称比值，保证赢家伤害严格小于败者伤害。 @@ -217,3 +205,112 @@ def get_assassination_success_rate(attacker: "Avatar", defender: "Avatar") -> fl rate += extra return max(0.01, min(1.0, rate)) + + +async def gen_battle_result_text( + winner: "Avatar", + loser: "Avatar", + l_dmg: int, + w_dmg: int, + is_fatal: bool, + prefix: str = "", + action_desc: str = "战胜了", + postfix: str = "", + check_loot: bool = False +) -> str: + """ + 生成标准战斗结果文本。 + """ + text_prefix = f"{prefix} " if prefix else "" + if is_fatal: + text = f"{text_prefix}{winner.name} {action_desc} {loser.name}{postfix}，造成 {l_dmg} 点伤害。{loser.name} 遭受重创，当场陨落。" + if check_loot: + from src.classes.kill_and_grab import kill_and_grab + text += await kill_and_grab(winner, loser) + return text + else: + return f"{text_prefix}{winner.name} {action_desc} {loser.name}{postfix}，{loser.name} 受伤 {l_dmg} 点，{winner.name} 也受伤 {w_dmg} 点。" + + +async def handle_battle_finish( + world, + attacker: "Avatar", + target: "Avatar", + res: Tuple["Avatar", "Avatar", int, int], + start_event_content: str, + story_prompt: str, + outcome_text_func: Optional[Callable[["Avatar", "Avatar", int, int, bool], Awaitable[str]]] = None, + check_loot: bool = False, + prefix: str = "", + action_desc: str = "战胜了", + postfix: str = "" +) -> list["Event"]: + """ + 处理战斗结果的通用逻辑（生成事件、故事、处理死亡）。 + + Args: + world: 世界对象 + attacker: 发起者 + target: 目标 + res: decide_battle 的结果 (winner, loser, loser_damage, winner_damage) + start_event_content: 开始事件的内容（用于故事生成） + story_prompt: 故事生成的提示词 + outcome_text_func: (可选) 异步回调函数，用于生成结果文本。 + 如果为 None，则使用标准生成逻辑。 + 参数: (winner, loser, loser_damage, winner_damage, is_fatal) + 返回: 结果文本字符串 + check_loot: 是否检查杀人夺宝（仅当 is_fatal 为 True 且使用默认文本生成时生效，或者 outcome_text_func 自己处理） + prefix: 默认文本生成的前缀（仅当 outcome_text_func 为 None 时生效） + action_desc: 默认文本生成的动作描述（仅当 outcome_text_func 为 None 时生效） + postfix: 默认文本生成的后缀（仅当 outcome_text_func 为 None 时生效） + """ + from src.classes.event import Event + from src.classes.story_teller import StoryTeller + from src.classes.death import handle_death + from src.classes.death_reason import DeathReason + + winner, loser, loser_damage, winner_damage = res + is_fatal = loser.hp <= 0 + + # 生成结果文本 + if outcome_text_func: + result_text = await outcome_text_func(winner, loser, loser_damage, winner_damage, is_fatal) + else: + result_text = await gen_battle_result_text( + winner, loser, loser_damage, winner_damage, is_fatal, + prefix=prefix, action_desc=action_desc, postfix=postfix, check_loot=check_loot + ) + + # 构造事件 + rel_ids = [attacker.id] + if target: + rel_ids.append(target.id) + + result_event = Event(world.month_stamp, result_text, related_avatars=rel_ids, is_major=True) + + # 确定故事生成的起始文本（如果为空则使用结果文本作为兜底） + start_content = start_event_content + if not start_content: + start_content = result_text + + # 生成故事 + story = await StoryTeller.tell_story( + start_content, + result_event.content, + attacker, + target, + prompt=story_prompt, + allow_relation_changes=True + ) + story_event = Event(world.month_stamp, story, related_avatars=rel_ids, is_story=True) + + # 处理死亡 + if is_fatal: + handle_death(world, loser, DeathReason.BATTLE) + + # 将事件分发给目标（如果目标不是发起者），发起者由 ActionMixin 处理 + if target and target.id != attacker.id: + target.add_event(result_event) + target.add_event(story_event) + + return [result_event, story_event] diff --git a/src/classes/mutual_action/mutual_action.py b/src/classes/mutual_action/mutual_action.py index ce272e3..a6d6863 100644 --- a/src/classes/mutual_action/mutual_action.py +++ b/src/classes/mutual_action/mutual_action.py @@ -20,7 +20,7 @@ if TYPE_CHECKING: from src.classes.world import World -class MutualAction(DefineAction, LLMAction, TargetingMixin): +class MutualAction(DefineAction, LLMAction, ActualActionMixin, TargetingMixin): """ 互动动作：A 对 B 发起动作，B 可以给出反馈（由 LLM 决策）。 子类需要定义： diff --git a/src/classes/mutual_action/occupy.py b/src/classes/mutual_action/occupy.py index 946218f..5716cce 100644 --- a/src/classes/mutual_action/occupy.py +++ b/src/classes/mutual_action/occupy.py @@ -1,62 +1,56 @@ from __future__ import annotations +import random from typing import TYPE_CHECKING + from src.classes.mutual_action.mutual_action import MutualAction from src.classes.event import Event from src.classes.action.registry import register_action +from src.classes.action.cooldown import cooldown_action from src.classes.region import resolve_region, CultivateRegion from src.classes.action_runtime import ActionResult, ActionStatus +from src.classes.battle import decide_battle +from src.classes.story_teller import StoryTeller +from src.classes.death import handle_death +from src.classes.death_reason import DeathReason if TYPE_CHECKING: from src.classes.avatar import Avatar - from src.classes.world import World + +@cooldown_action @register_action(actual=True) class Occupy(MutualAction): """ 占据动作（互动版）： 占据指定的洞府。如果是无主洞府直接占据；如果是有主洞府，则发起抢夺。 + 对方拒绝则进入战斗，进攻方胜利则洞府易主。 """ ACTION_NAME = "Occupy" COMMENT = "占据或抢夺洞府" - - # 参数：洞府名称 PARAMS = {"region_name": "str"} - - # 对方的反馈选项（仅在抢夺时有效） FEEDBACK_ACTIONS = ["Yield", "Reject"] - - # 反馈对应的中文描述 - FEEDBACK_LABELS = { - "Yield": "让步", - "Reject": "拒绝", - } - - # 是大事 + FEEDBACK_LABELS = {"Yield": "让步", "Reject": "拒绝"} IS_MAJOR = True + ACTION_CD_MONTHS = 6 + + STORY_PROMPT = "这是一场争夺洞府的战斗。不要出现具体血量或伤害数值。" - def _get_region_and_host(self, region_name: str) -> tuple[CultivateRegion | None, Avatar | None, str]: - """ - 解析区域并获取主人 - """ - try: - region = resolve_region(self.world, region_name) - except Exception as e: + def _get_region_and_host(self, region_name: str) -> tuple[CultivateRegion | None, "Avatar | None", str]: + """解析区域并获取主人""" + region = resolve_region(self.world, region_name) + if region is None: return None, None, f"无法找到区域：{region_name}" - if not isinstance(region, CultivateRegion): return None, None, f"{region.name} 不是修炼区域，无法占据" - return region, region.host_avatar, "" def can_start(self, region_name: str) -> tuple[bool, str]: region, host, err = self._get_region_and_host(region_name) if err: return False, err - if region.host_avatar == self.avatar: return False, "已经是该洞府的主人了" - return super().can_start(target_avatar=host) def start(self, region_name: str) -> Event: @@ -68,19 +62,63 @@ class Occupy(MutualAction): return super().step(target_avatar=host) def _settle_feedback(self, target_avatar: "Avatar", feedback_name: str) -> None: - """ - 处理反馈结果 - """ + """处理反馈结果""" region = self.avatar.tile.region + if feedback_name == "Yield": - # 对方让步：转移所有权 + # 对方让步：直接转移所有权 region.host_avatar = self.avatar - # 记录事件 - self.avatar.add_event(self.create_event(f"成功从 {target_avatar.name} 手中夺取了 {region.name}", related_avatars=[target_avatar.id])) - target_avatar.add_event(Event(self.world.month_stamp, f"面对 {self.avatar.name} 的逼迫，不得不让出了 {region.name}", related_avatars=[self.avatar.id], is_major=True)) + # 共用一个事件 + event_text = f"{self.avatar.name} 逼迫 {target_avatar.name} 让出了 {region.name}。" + event = Event( + self.world.month_stamp, + event_text, + related_avatars=[self.avatar.id, target_avatar.id], + is_major=True + ) + self.avatar.add_event(event) + target_avatar.add_event(event) + + self._last_result = None elif feedback_name == "Reject": - # 对方拒绝：所有权不变 - self.avatar.add_event(self.create_event(f"试图抢夺 {region.name}，但被 {target_avatar.name} 拒绝", related_avatars=[target_avatar.id])) - target_avatar.add_event(Event(self.world.month_stamp, f"拒绝了 {self.avatar.name} 对 {region.name} 的抢夺要求", related_avatars=[self.avatar.id], is_major=True)) + # 对方拒绝：进入战斗 + winner, loser, loser_dmg, winner_dmg = decide_battle(self.avatar, target_avatar) + loser.hp.reduce(loser_dmg) + winner.hp.reduce(winner_dmg) + + # 进攻方胜利则洞府易主 + attacker_won = winner == self.avatar + if attacker_won: + region.host_avatar = self.avatar + + self._last_result = (winner, loser, loser_dmg, winner_dmg, region.name, attacker_won) + + async def finish(self, region_name: str) -> list[Event]: + """完成动作，生成战斗故事并处理死亡""" + res = self._last_result if hasattr(self, '_last_result') else None + if res is None: + return [] + + # res format from occupy: (winner, loser, l_dmg, w_dmg, r_name, attacker_won) + winner, loser, l_dmg, w_dmg, r_name, attacker_won = res + battle_res = (winner, loser, l_dmg, w_dmg) + + target = loser if winner == self.avatar else winner + + start_text = f"{self.avatar.name} 试图抢夺 {target.name} 的洞府 {r_name}，{target.name} 拒绝并应战" + + postfix = f"，成功夺取了 {r_name}" if attacker_won else f"，守住了 {r_name}" + + from src.classes.battle import handle_battle_finish + return await handle_battle_finish( + self.world, + self.avatar, + target, + battle_res, + start_text, + self.STORY_PROMPT, + action_desc="击败了", + postfix=postfix + ) diff --git a/src/classes/mutual_action/talk.py b/src/classes/mutual_action/talk.py index 019fd66..ebce573 100644 --- a/src/classes/mutual_action/talk.py +++ b/src/classes/mutual_action/talk.py @@ -56,13 +56,17 @@ class Talk(MutualAction): ) EventHelper.push_pair(accept_event, initiator=self.avatar, target=target, to_sidebar_once=True) - # 将 Conversation 加入计划队列，在Talk完成后立即执行（插队到最前） + # 将 Conversation 加入计划队列并立即提交 self.avatar.load_decide_result_chain( [("Conversation", {"target_avatar": target.name})], self.avatar.thinking, self.avatar.short_term_objective, prepend=True ) + # 立即提交为当前动作 + start_event = self.avatar.commit_next_plan() + if start_event is not None: + EventHelper.push_pair(start_event, initiator=self.avatar, target=target, to_sidebar_once=True) else: # 拒绝攀谈 reject_event = Event( diff --git a/tests/test_simulator.py b/tests/test_simulator.py index eac69ae..2f4f5be 100644 --- a/tests/test_simulator.py +++ b/tests/test_simulator.py @@ -8,6 +8,8 @@ from src.classes.map import Map from src.classes.tile import TileType from src.classes.action import Move from src.classes.name import get_random_name +from src.classes.age import Age +from src.classes.cultivation import Realm def test_simulator_step_moves_avatar_and_sets_tile(): @@ -28,7 +30,7 @@ def test_simulator_step_moves_avatar_and_sets_tile(): name=get_random_name(Gender.MALE), id="1", birth_month_stamp=create_month_stamp(Year(2000), Month.JANUARY), - age=20, + age=Age(20, Realm.Qi_Refinement), gender=Gender.MALE, pos_x=1, pos_y=1, @@ -36,7 +38,7 @@ def test_simulator_step_moves_avatar_and_sets_tile(): sim = Simulator(world) - sim.avatars["1"] = avatar + world.avatar_manager.avatars["1"] = avatar # 执行一步模拟 sim.step()