import logging import json import random import math import time from copy import deepcopy from combat import generate_uuid from notification import send_public from static_db import db logger = logging.getLogger(__name__) class CraftSystem: def __init__(self, config, new_equip_templates=None): self.config = config self.new_equip_templates = new_equip_templates or {} def _get_base_equipment(self, name, item_type): """从静态缓存获取装备模板""" equipment_data = db.get_equipment() for category, items in equipment_data.items(): for item in items: if item.get('name') == name and item.get('type') == item_type: return deepcopy(item) for key, template in self.new_equip_templates.items(): if template.get('name') == name and template.get('type') == item_type: return deepcopy(template) return None def craft_equipment(self, state, uuids, usePoints=False): # 获取合成配置（统一从 craft 节读取） craft_cfg = self.config.get('craft', {}) required_items = craft_cfg.get('required_items_count', 3) cost_by_quality = craft_cfg.get('cost_by_quality', {}) default_success_rate = craft_cfg.get('default_success_rate', 0.5) special_effect_chance = craft_cfg.get('special_effect_chance', 1.0) special_effects_list = craft_cfg.get('weapon_special_effects', ['攻击吸血', '攻击回蓝', '双倍伤害', '无视防御']) point_cost_by_quality = craft_cfg.get('point_cost_by_quality', [10, 50, 100]) # 检查物品数量 if len(uuids) != required_items: return state, f'需要选择 {required_items} 件装备进行合成' warehouse = state.get('warehouse', []) # 查找物品 indices, items, error = self._find_items_by_uuids(warehouse, uuids) if error: return state, error for item in items: if 'quality' not in item or item['quality'] is None: item['quality'] = '普通' # 检查物品是否可合成 base_item = items[0] from_quality = base_item.get('quality', '普通') for i, item in enumerate(items): if item['type'] != base_item['type']: return state, f'物品类型不一致：{item["name"]} 不是 {base_item["type"]} 类型' if item['name'] != base_item['name']: return state, f'物品名称不一致：{item["name"]} 与 {base_item["name"]} 不同' if item.get('quality', '普通') != from_quality: return state, f'物品品质不一致，必须 {required_items} 件相同品质' if item['type'] == 'potion': return state, '药水不能合成' if item.get('bindOnEquip') or item.get('bindOnPickup'): return state, '绑定的装备不能合成' # 排除特戒 special_rings = self.config.get('equipmentData', {}).get('specialRings', []) special_ring_ids = {ring.get('id') for ring in special_rings} if base_item.get('id') in special_ring_ids: return state, '特戒不能参与合成' # 品质映射（普通->极品，极品->神品，神品->超神品） quality_map = { '普通': '极品', '极品': '神品', '神品': '超神品' } if from_quality not in quality_map: return state, f'当前品质"{from_quality}"无法继续合成（仅支持普通、极品、神品）' target_quality = quality_map[from_quality] cost_gold = cost_by_quality.get(from_quality, 100000) # 积分消耗（根据品质等级从 point_cost_by_quality 列表中取） quality_level = {'普通': 0, '极品': 1, '神品': 2}.get(from_quality, 0) if isinstance(point_cost_by_quality, list) and quality_level < len(point_cost_by_quality): required_points = point_cost_by_quality[quality_level] else: required_points = point_cost_by_quality if isinstance(point_cost_by_quality, int) else 10 # 根据 usePoints 决定积分处理及成功率 if usePoints: if 'task' not in state: state['task'] = {'points': 0} task_points = state['task'].get('points', 0) if task_points < required_points: return state, f'积分不足，需要 {required_points} 积分（当前 {task_points}）' state['task']['points'] = task_points - required_points success_rate = 1.0 else: success_rate = default_success_rate # 检查金币 if state['player']['gold'] < cost_gold: return state, f'金币不足，需要 {cost_gold} 金币' # 扣除金币 state['player']['gold'] -= cost_gold # 随机判定 success = random.random() < success_rate if success: base_config = self._get_base_equipment(base_item['name'], base_item['type']) if not base_config: return state, f'找不到装备 {base_item["name"]} 的配置' new_item = deepcopy(base_config) new_item['uuid'] = generate_uuid() new_item['enhanceLevel'] = 0 new_item['quality'] = target_quality new_item['droppedBy'] = f"{state['player']['name']}合成" new_item['acquiredTime'] = int(time.time()) if 'type' not in new_item or not new_item['type']: new_item['type'] = base_item.get('type') # 应用品质加成 if target_quality != '普通': self.apply_quality_bonus(new_item, target_quality) # 武器特效（仅超神品，且根据配置概率） if target_quality == '超神品' and new_item.get('type') == 'weapon': if random.random() < special_effect_chance: new_item['special'] = random.choice(special_effects_list) # 过滤零属性 if new_item.get('attackMin', 0) == 0 and new_item.get('attackMax', 0) == 0: new_item.pop('attackMin', None) new_item.pop('attackMax', None) if new_item.get('defenseMin', 0) == 0 and new_item.get('defenseMax', 0) == 0: new_item.pop('defenseMin', None) new_item.pop('defenseMax', None) # 移除旧装备并添加新装备 for idx in sorted(indices, reverse=True): warehouse.pop(idx) if len(warehouse) < state.get('maxWarehouseSlots', 500): warehouse.append(new_item) message = f'✨ 合成成功！获得 {new_item["name"]}（{target_quality}）' if target_quality == '超神品' and new_item.get('type') == 'weapon' and new_item.get('special'): from notification import send_public send_public(f"{state['player']['name']} 通过合成打造出超神武器【{new_item['name']}】并获得特效『{new_item['special']}』！") else: message = f'⚠️ 合成成功但仓库已满，{new_item["name"]} 消失' else: message = f'❌ 合成失败，消耗 {cost_gold} 金币，再接再厉！' state['warehouse'] = warehouse state = self._update_final_stats(state) return state, message def apply_quality_bonus(self, item, quality): drop_config = self.config.get('dropConfig', {}).get('equipmentDrop', {}) quality_percent_bonus = drop_config.get('qualityPercentBonus', {}) quality_min_bonus = drop_config.get('qualityBonus', {}) bonus_percent = quality_percent_bonus.get(quality, 0) min_bonus = quality_min_bonus.get(quality, 0) if bonus_percent <= 0: return item # 攻击属性加成 if item.get('attackMin', 0) > 0: att_bonus = max(min_bonus, int(item['attackMin'] * bonus_percent)) item['attackMin'] += att_bonus item['attackMax'] += att_bonus # 防御属性加成 if item.get('defenseMin', 0) > 0: def_bonus = max(min_bonus, int(item['defenseMin'] * bonus_percent)) item['defenseMin'] += def_bonus item['defenseMax'] += def_bonus # 生命加成（斗笠、披风可用） if item.get('hpBonus', 0) > 0: hp_bonus = max(min_bonus * 10, int(item['hpBonus'] * bonus_percent)) item['hpBonus'] += hp_bonus # 魔法加成（马牌可用） if item.get('mpBonus', 0) > 0: mp_bonus = max(min_bonus * 10, int(item['mpBonus'] * bonus_percent)) item['mpBonus'] += mp_bonus return item def _update_final_stats(self, state): """更新最终属性（从combat.py复制简化版）""" player = state['player'] hero = state['hero'] equip = state['equipment'] player['finalAttackMin'] = player.get('attackMin', 0) + player.get('attackEnhance', 0) player['finalAttackMax'] = player.get('attackMax', 0) + player.get('attackEnhance', 0) player['finalDefenseMin'] = player.get('defenseMin', 0) + player.get('defenseEnhance', 0) player['finalDefenseMax'] = player.get('defenseMax', 0) + player.get('defenseEnhance', 0) for slot, item in equip.get('player', {}).items(): if item: player['finalAttackMin'] += item.get('attackMin', 0) + item.get('enhanceLevel', 0) player['finalAttackMax'] += item.get('attackMax', 0) + item.get('enhanceLevel', 0) player['finalDefenseMin'] += item.get('defenseMin', 0) + item.get('enhanceLevel', 0) player['finalDefenseMax'] += item.get('defenseMax', 0) + item.get('enhanceLevel', 0) if hero.get('summoned'): hero['finalAttackMin'] = hero.get('attackMin', 0) + hero.get('attackEnhance', 0) hero['finalAttackMax'] = hero.get('attackMax', 0) + hero.get('attackEnhance', 0) hero['finalDefenseMin'] = hero.get('defenseMin', 0) + hero.get('defenseEnhance', 0) hero['finalDefenseMax'] = hero.get('defenseMax', 0) + hero.get('defenseEnhance', 0) for slot, item in equip.get('hero', {}).items(): if item: hero['finalAttackMin'] += item.get('attackMin', 0) + item.get('enhanceLevel', 0) hero['finalAttackMax'] += item.get('attackMax', 0) + item.get('enhanceLevel', 0) hero['finalDefenseMin'] += item.get('defenseMin', 0) + item.get('enhanceLevel', 0) hero['finalDefenseMax'] += item.get('defenseMax', 0) + item.get('enhanceLevel', 0) return state def _find_items_by_uuids(self, inventory, uuids): """通过UUID列表查找物品，返回索引列表和物品列表""" indices = [] items = [] for uuid in uuids: found = False for i, item in enumerate(inventory): if item.get('uuid') == uuid: indices.append(i) items.append(item) found = True break if not found: return None, None, f'物品UUID {uuid} 不存在' return indices, items, None def get_craftable_items(self, state): warehouse = state.get('warehouse', []) craftable = {} for item in warehouse: # 跳过药水 if item.get('type') == 'potion': continue # 跳过绑定装备 if item.get('bindOnEquip') or item.get('bindOnPickup'): continue # ---------- 关键：强制标准化字段 ---------- name = str(item.get('name', '')).strip() if not name: continue # 处理 type：优先取顶层，若为空或非字符串则尝试从 data 中读取 item_type = item.get('type') if item_type is None or not isinstance(item_type, str): data = item.get('data') if isinstance(data, dict): item_type = data.get('type') if not item_type: item_type = 'unknown' item_type = str(item_type).strip() # 处理 quality：同样标准化 quality = item.get('quality') if quality is None: quality = '普通' else: quality = str(quality).strip() if not quality: quality = '普通' # 再次检查有效性 if not name or not item_type: continue key = f"{name}_{item_type}_{quality}" if key not in craftable: craftable[key] = { 'name': name, 'type': item_type, 'quality': quality, 'count': 0, 'items': [] } craftable[key]['count'] += 1 craftable[key]['items'].append({ 'uuid': item['uuid'], 'quality': quality, 'level': item.get('level', 1) }) # 仅返回 count >= 3 的分组 result = [info for info in craftable.values() if info['count'] >= 3] return result