""" Inventory Service - Separate microservice for item data processing and queries. Handles enum translation, data normalization, and provides structured item data APIs. """ import json import logging from pathlib import Path from typing import Dict, List, Optional, Any from datetime import datetime from fastapi import FastAPI, HTTPException, Depends, Query from fastapi.middleware.cors import CORSMiddleware from fastapi.responses import JSONResponse from pydantic import BaseModel import databases import sqlalchemy as sa from database import ( Base, Item, ItemCombatStats, ItemRequirements, ItemEnhancements, ItemRatings, ItemSpells, ItemRawData, DATABASE_URL, create_indexes ) # Configure logging logging.basicConfig(level=logging.INFO) logger = logging.getLogger(__name__) # FastAPI app app = FastAPI( title="Inventory Service", description="Microservice for Asheron's Call item data processing and queries", version="1.0.0" ) # Configure CORS app.add_middleware( CORSMiddleware, allow_origins=["*"], # Allow all origins for development allow_credentials=True, allow_methods=["*"], allow_headers=["*"], ) # Database connection database = databases.Database(DATABASE_URL) engine = sa.create_engine(DATABASE_URL) # Load comprehensive enum mappings def load_comprehensive_enums(): """Load complete enum database with all translations.""" logger.info("Loading comprehensive enum database...") try: # Try new comprehensive database first logger.info("Attempting to load comprehensive_enum_database_v2.json") with open('comprehensive_enum_database_v2.json', 'r') as f: enum_db = json.load(f) logger.info("Successfully loaded comprehensive_enum_database_v2.json") except FileNotFoundError: logger.warning("comprehensive_enum_database_v2.json not found, trying fallback") try: with open('complete_enum_database.json', 'r') as f: enum_db = json.load(f) logger.info("Successfully loaded complete_enum_database.json") except FileNotFoundError as e: logger.error(f"No enum database found: {e}") return {'int_values': {}, 'materials': {}, 'item_types': {}, 'skills': {}, 'spell_categories': {}, 'spells': {}, 'object_classes': {}, 'coverage_masks': {}} except Exception as e: logger.error(f"Error reading enum database file: {e}") return {'int_values': {}, 'materials': {}, 'item_types': {}, 'skills': {}, 'spell_categories': {}, 'spells': {}, 'object_classes': {}, 'coverage_masks': {}} # Extract specific enum mappings for easy access from new format logger.info("Processing loaded enum database...") enums = enum_db.get('enums', {}) spells_data = enum_db.get('spells', {}) object_classes_data = enum_db.get('object_classes', {}) # Convert IntValueKey to integer-keyed dict for fast lookup int_values = {} if 'IntValueKey' in enums: for k, v in enums['IntValueKey']['values'].items(): try: int_values[int(k)] = v except (ValueError, TypeError): pass # Skip non-numeric keys # Material types mapping materials = {} if 'MaterialType' in enums: for k, v in enums['MaterialType']['values'].items(): try: materials[int(k)] = v except (ValueError, TypeError): pass # Item types mapping item_types = {} if 'ItemType' in enums: for k, v in enums['ItemType']['values'].items(): try: item_types[int(k)] = v except (ValueError, TypeError): pass # Skills mapping skills = {} if 'Skill' in enums: skill_data = enums['Skill']['values'] for k, v in skill_data.items(): try: if isinstance(v, dict): skills[int(k)] = v.get('name', str(v)) else: skills[int(k)] = str(v) except (ValueError, TypeError): pass # Spell categories mapping spell_categories = {} if 'SpellCategory' in enums: for k, v in enums['SpellCategory']['values'].items(): try: spell_categories[int(k)] = v except (ValueError, TypeError): pass # Coverage mask mapping coverage_masks = {} if 'CoverageMask' in enums: for k, v in enums['CoverageMask']['values'].items(): try: coverage_masks[int(k)] = v except (ValueError, TypeError): pass # Object classes mapping object_classes = {} if object_classes_data and 'values' in object_classes_data: for k, v in object_classes_data['values'].items(): try: object_classes[int(k)] = v except (ValueError, TypeError): pass # Spells mapping spells = {} if spells_data and 'values' in spells_data: spells = {int(k): v for k, v in spells_data['values'].items() if k.isdigit()} # AttributeSetInfo - Equipment Set Names (CRITICAL for armor set detection) attribute_sets = {} # Check in dictionaries section first, then enums as fallback if 'dictionaries' in enum_db and 'AttributeSetInfo' in enum_db['dictionaries']: for k, v in enum_db['dictionaries']['AttributeSetInfo']['values'].items(): attribute_sets[k] = v # String key try: attribute_sets[int(k)] = v # Also int key except (ValueError, TypeError): pass elif 'AttributeSetInfo' in enums: for k, v in enums['AttributeSetInfo']['values'].items(): attribute_sets[k] = v # String key try: attribute_sets[int(k)] = v # Also int key except (ValueError, TypeError): pass logger.info(f"Enum database loaded successfully: {len(int_values)} int values, {len(spells)} spells, {len(object_classes)} object classes, {len(attribute_sets)} equipment sets") return { 'int_values': int_values, 'materials': materials, 'item_types': item_types, 'skills': skills, 'spell_categories': spell_categories, 'coverage_masks': coverage_masks, 'object_classes': object_classes, 'spells': spells, 'equipment_sets': attribute_sets, # Backward compatibility 'dictionaries': { 'AttributeSetInfo': {'values': attribute_sets} }, 'AttributeSetInfo': {'values': attribute_sets}, # Also add in the format expected by the endpoint 'full_database': enum_db } ENUM_MAPPINGS = load_comprehensive_enums() # Pydantic models class InventoryItem(BaseModel): """Raw inventory item from plugin.""" character_name: str timestamp: datetime items: List[Dict[str, Any]] class ProcessedItem(BaseModel): """Processed item with translated properties.""" name: str icon: int object_class: int value: int burden: int # Add other fields as needed # Startup/shutdown events @app.on_event("startup") async def startup(): """Initialize database connection and create tables.""" await database.connect() # Create tables if they don't exist Base.metadata.create_all(engine) # Create performance indexes create_indexes(engine) logger.info("Inventory service started successfully") @app.on_event("shutdown") async def shutdown(): """Close database connection.""" await database.disconnect() # Enhanced translation functions def translate_int_values(int_values: Dict[str, int]) -> Dict[str, Any]: """Translate IntValues enum keys to human-readable names using comprehensive database.""" translated = {} int_enum_map = ENUM_MAPPINGS.get('int_values', {}) for key_str, value in int_values.items(): try: key_int = int(key_str) if key_int in int_enum_map: enum_name = int_enum_map[key_int] translated[enum_name] = value else: # Keep unknown keys with numeric identifier translated[f"unknown_{key_int}"] = value except ValueError: # Skip non-numeric keys translated[key_str] = value return translated def translate_material_type(material_id: int) -> str: """Translate material type ID to human-readable name.""" materials = ENUM_MAPPINGS.get('materials', {}) return materials.get(material_id, f"Unknown_Material_{material_id}") def translate_item_type(item_type_id: int) -> str: """Translate item type ID to human-readable name.""" item_types = ENUM_MAPPINGS.get('item_types', {}) return item_types.get(item_type_id, f"Unknown_ItemType_{item_type_id}") def translate_object_class(object_class_id: int, item_data: dict = None) -> str: """Translate object class ID to human-readable name with context-aware detection.""" # Use the extracted ObjectClass enum first object_classes = ENUM_MAPPINGS.get('object_classes', {}) if object_class_id in object_classes: base_name = object_classes[object_class_id] # Context-aware classification for Gem class (ID 11) if base_name == "Gem" and object_class_id == 11 and item_data: # Check item name and properties to distinguish types item_name = item_data.get('Name', '').lower() # Mana stones and crystals if any(keyword in item_name for keyword in ['mana stone', 'crystal', 'gem']): if 'mana stone' in item_name: return "Mana Stone" elif 'crystal' in item_name: return "Crystal" else: return "Gem" # Aetheria detection - check for specific properties int_values = item_data.get('IntValues', {}) if isinstance(int_values, dict): # Check for Aetheria-specific properties has_item_set = '265' in int_values or 265 in int_values # EquipmentSetId has_aetheria_level = '218103840' in int_values or 218103840 in int_values # ItemMaxLevel if has_item_set or has_aetheria_level or 'aetheria' in item_name: return "Aetheria" # Default to Gem for other items in this class return "Gem" return base_name # Fallback to WeenieType enum weenie_types = {} if 'WeenieType' in ENUM_MAPPINGS.get('full_database', {}).get('enums', {}): weenie_data = ENUM_MAPPINGS['full_database']['enums']['WeenieType']['values'] for k, v in weenie_data.items(): try: weenie_types[int(k)] = v except (ValueError, TypeError): pass return weenie_types.get(object_class_id, f"Unknown_ObjectClass_{object_class_id}") def translate_skill(skill_id: int) -> str: """Translate skill ID to skill name.""" skills = ENUM_MAPPINGS.get('skills', {}) return skills.get(skill_id, f"Unknown_Skill_{skill_id}") def translate_spell_category(category_id: int) -> str: """Translate spell category ID to spell category name.""" spell_categories = ENUM_MAPPINGS.get('spell_categories', {}) return spell_categories.get(category_id, f"Unknown_SpellCategory_{category_id}") def translate_spell(spell_id: int) -> Dict[str, Any]: """Translate spell ID to spell data including name, description, school, etc.""" spells = ENUM_MAPPINGS.get('spells', {}) spell_data = spells.get(spell_id) if spell_data: return { 'id': spell_id, 'name': spell_data.get('name', f'Unknown_Spell_{spell_id}'), 'description': spell_data.get('description', ''), 'school': spell_data.get('school', ''), 'difficulty': spell_data.get('difficulty', ''), 'duration': spell_data.get('duration', ''), 'mana': spell_data.get('mana', ''), 'family': spell_data.get('family', '') } else: return { 'id': spell_id, 'name': f'Unknown_Spell_{spell_id}', 'description': '', 'school': '', 'difficulty': '', 'duration': '', 'mana': '', 'family': '' } def translate_coverage_mask(coverage_value: int) -> List[str]: """Translate coverage mask value to list of body parts covered.""" coverage_masks = ENUM_MAPPINGS.get('coverage_masks', {}) covered_parts = [] # Coverage masks are flags, so we need to check each bit for mask_value, part_name in coverage_masks.items(): if coverage_value & mask_value: # Convert technical names to display names display_name = part_name.replace('Outerwear', '').replace('Underwear', '').strip() if display_name and display_name not in covered_parts: covered_parts.append(display_name) # Map technical names to user-friendly names name_mapping = { 'UpperLegs': 'Upper Legs', 'LowerLegs': 'Lower Legs', 'UpperArms': 'Upper Arms', 'LowerArms': 'Lower Arms', 'Abdomen': 'Abdomen', 'Chest': 'Chest', 'Head': 'Head', 'Hands': 'Hands', 'Feet': 'Feet', 'Cloak': 'Cloak' } return [name_mapping.get(part, part) for part in covered_parts if part] def get_total_bits_set(value: int) -> int: """Count the number of bits set in a value.""" bits_set = 0 while value != 0: if (value & 1) == 1: bits_set += 1 value >>= 1 return bits_set def is_body_armor_equip_mask(value: int) -> bool: """Check if EquipMask value represents body armor.""" return (value & 0x00007F21) != 0 def is_body_armor_coverage_mask(value: int) -> bool: """Check if CoverageMask value represents body armor.""" return (value & 0x0001FF00) != 0 def get_coverage_reduction_options(coverage_value: int) -> list: """ Get the reduction options for a coverage mask, based on Mag-SuitBuilder logic. This determines which individual slots a multi-slot armor piece can be tailored to fit. """ # CoverageMask values from Mag-Plugins OuterwearUpperArms = 4096 # 0x00001000 OuterwearLowerArms = 8192 # 0x00002000 OuterwearUpperLegs = 256 # 0x00000100 OuterwearLowerLegs = 512 # 0x00000200 OuterwearChest = 1024 # 0x00000400 OuterwearAbdomen = 2048 # 0x00000800 Head = 16384 # 0x00004000 Hands = 32768 # 0x00008000 Feet = 65536 # 0x00010000 options = [] # If single bit or not body armor, return as-is if get_total_bits_set(coverage_value) <= 1 or not is_body_armor_coverage_mask(coverage_value): options.append(coverage_value) else: # Implement Mag-SuitBuilder reduction logic if coverage_value == (OuterwearUpperArms | OuterwearLowerArms): options.extend([OuterwearUpperArms, OuterwearLowerArms]) elif coverage_value == (OuterwearUpperLegs | OuterwearLowerLegs): options.extend([OuterwearUpperLegs, OuterwearLowerLegs]) elif coverage_value == (OuterwearLowerLegs | Feet): options.append(Feet) elif coverage_value == (OuterwearChest | OuterwearAbdomen): options.append(OuterwearChest) elif coverage_value == (OuterwearChest | OuterwearAbdomen | OuterwearUpperArms): options.append(OuterwearChest) elif coverage_value == (OuterwearChest | OuterwearUpperArms | OuterwearLowerArms): options.append(OuterwearChest) elif coverage_value == (OuterwearChest | OuterwearUpperArms): options.append(OuterwearChest) elif coverage_value == (OuterwearAbdomen | OuterwearUpperLegs | OuterwearLowerLegs): options.extend([OuterwearAbdomen, OuterwearUpperLegs, OuterwearLowerLegs]) elif coverage_value == (OuterwearChest | OuterwearAbdomen | OuterwearUpperArms | OuterwearLowerArms): options.append(OuterwearChest) elif coverage_value == (OuterwearAbdomen | OuterwearUpperLegs): # Pre-2010 retail guidelines - assume abdomen reduction only options.append(OuterwearAbdomen) else: # If no specific reduction pattern, return original options.append(coverage_value) return options def coverage_to_equip_mask(coverage_value: int) -> int: """Convert a CoverageMask value to its corresponding EquipMask slot.""" # Coverage to EquipMask mapping from Mag-SuitBuilder coverage_to_slot = { 16384: 1, # Head -> HeadWear 1024: 512, # OuterwearChest -> ChestArmor 4096: 2048, # OuterwearUpperArms -> UpperArmArmor 8192: 4096, # OuterwearLowerArms -> LowerArmArmor 32768: 32, # Hands -> HandWear 2048: 1024, # OuterwearAbdomen -> AbdomenArmor 256: 8192, # OuterwearUpperLegs -> UpperLegArmor 512: 16384, # OuterwearLowerLegs -> LowerLegArmor 65536: 256, # Feet -> FootWear } return coverage_to_slot.get(coverage_value, coverage_value) def get_sophisticated_slot_options(equippable_slots: int, coverage_value: int, has_material: bool = True) -> list: """ Get sophisticated slot options using Mag-SuitBuilder logic. This handles armor reduction for tailorable pieces. """ # Special case: shoes that cover feet + lower legs but only go in feet slot LowerLegWear = 128 FootWear = 256 if equippable_slots == (LowerLegWear | FootWear): return [FootWear] # If it's body armor with multiple slots if is_body_armor_equip_mask(equippable_slots) and get_total_bits_set(equippable_slots) > 1: if not has_material: # Can't reduce non-loot gen pieces, return all slots return [equippable_slots] else: # Use coverage reduction options reduction_options = get_coverage_reduction_options(coverage_value) slot_options = [] for option in reduction_options: equip_slot = coverage_to_equip_mask(option) slot_options.append(equip_slot) return slot_options if slot_options else [equippable_slots] else: # Single slot or non-armor return [equippable_slots] def convert_slot_name_to_friendly(slot_name: str) -> str: """Convert technical slot names to user-friendly names.""" name_mapping = { 'HeadWear': 'Head', 'ChestWear': 'Chest', 'ChestArmor': 'Chest', 'AbdomenWear': 'Abdomen', 'AbdomenArmor': 'Abdomen', 'UpperArmWear': 'Upper Arms', 'UpperArmArmor': 'Upper Arms', 'LowerArmWear': 'Lower Arms', 'LowerArmArmor': 'Lower Arms', 'HandWear': 'Hands', 'UpperLegWear': 'Upper Legs', 'UpperLegArmor': 'Upper Legs', 'LowerLegWear': 'Lower Legs', 'LowerLegArmor': 'Lower Legs', 'FootWear': 'Feet', 'NeckWear': 'Neck', 'WristWearLeft': 'Left Wrist', 'WristWearRight': 'Right Wrist', 'FingerWearLeft': 'Left Ring', 'FingerWearRight': 'Right Ring', 'MeleeWeapon': 'Melee Weapon', 'Shield': 'Shield', 'MissileWeapon': 'Missile Weapon', 'MissileAmmo': 'Ammo', 'Held': 'Held', 'TwoHanded': 'Two-Handed', 'TrinketOne': 'Trinket', 'Cloak': 'Cloak', 'SigilOne': 'Aetheria Blue', 'SigilTwo': 'Aetheria Yellow', 'SigilThree': 'Aetheria Red' } return name_mapping.get(slot_name, slot_name) def translate_equipment_slot(wielded_location: int) -> str: """Translate equipment slot mask to human-readable slot name(s), handling bit flags.""" if wielded_location == 0: return "Inventory" # Get EquipMask enum from database equip_mask_map = {} if 'EquipMask' in ENUM_MAPPINGS.get('full_database', {}).get('enums', {}): equip_data = ENUM_MAPPINGS['full_database']['enums']['EquipMask']['values'] for k, v in equip_data.items(): try: # Skip expression values (EXPR:...) if not k.startswith('EXPR:'): equip_mask_map[int(k)] = v except (ValueError, TypeError): pass # Check for exact match first if wielded_location in equip_mask_map: slot_name = equip_mask_map[wielded_location] # Convert technical names to user-friendly names name_mapping = { 'HeadWear': 'Head', 'ChestWear': 'Chest', 'ChestArmor': 'Chest', 'AbdomenWear': 'Abdomen', 'AbdomenArmor': 'Abdomen', 'UpperArmWear': 'Upper Arms', 'UpperArmArmor': 'Upper Arms', 'LowerArmWear': 'Lower Arms', 'LowerArmArmor': 'Lower Arms', 'HandWear': 'Hands', 'UpperLegWear': 'Upper Legs', 'UpperLegArmor': 'Upper Legs', 'LowerLegWear': 'Lower Legs', 'LowerLegArmor': 'Lower Legs', 'FootWear': 'Feet', 'NeckWear': 'Neck', 'WristWearLeft': 'Left Wrist', 'WristWearRight': 'Right Wrist', 'FingerWearLeft': 'Left Ring', 'FingerWearRight': 'Right Ring', 'MeleeWeapon': 'Melee Weapon', 'Shield': 'Shield', 'MissileWeapon': 'Missile Weapon', 'MissileAmmo': 'Ammo', 'Held': 'Held', 'TwoHanded': 'Two-Handed', 'TrinketOne': 'Trinket', 'Cloak': 'Cloak' } return name_mapping.get(slot_name, slot_name) # If no exact match, decode bit flags slot_parts = [] for mask_value, slot_name in equip_mask_map.items(): if mask_value > 0 and (wielded_location & mask_value) == mask_value: slot_parts.append(slot_name) if slot_parts: # Convert technical names to user-friendly names name_mapping = { 'HeadWear': 'Head', 'ChestWear': 'Chest', 'ChestArmor': 'Chest', 'AbdomenWear': 'Abdomen', 'AbdomenArmor': 'Abdomen', 'UpperArmWear': 'Upper Arms', 'UpperArmArmor': 'Upper Arms', 'LowerArmWear': 'Lower Arms', 'LowerArmArmor': 'Lower Arms', 'HandWear': 'Hands', 'UpperLegWear': 'Upper Legs', 'UpperLegArmor': 'Upper Legs', 'LowerLegWear': 'Lower Legs', 'LowerLegArmor': 'Lower Legs', 'FootWear': 'Feet', 'NeckWear': 'Neck', 'WristWearLeft': 'Left Wrist', 'WristWearRight': 'Right Wrist', 'FingerWearLeft': 'Left Ring', 'FingerWearRight': 'Right Ring', 'MeleeWeapon': 'Melee Weapon', 'Shield': 'Shield', 'MissileWeapon': 'Missile Weapon', 'MissileAmmo': 'Ammo', 'Held': 'Held', 'TwoHanded': 'Two-Handed', 'TrinketOne': 'Trinket', 'Cloak': 'Cloak' } translated_parts = [name_mapping.get(part, part) for part in slot_parts] return ', '.join(translated_parts) # Handle special cases for high values (like Aetheria slots) if wielded_location >= 268435456: # 2^28 and higher - likely Aetheria or special slots if wielded_location == 268435456: return "Aetheria Blue" elif wielded_location == 536870912: return "Aetheria Yellow" elif wielded_location == 1073741824: return "Aetheria Red" else: return f"Special Slot ({wielded_location})" return f"Unknown Slot ({wielded_location})" def translate_workmanship(workmanship_value: int) -> str: """Translate workmanship value to descriptive text.""" if workmanship_value <= 0: return "" elif workmanship_value == 1: return "Pitiful (1)" elif workmanship_value == 2: return "Poor (2)" elif workmanship_value == 3: return "Below Average (3)" elif workmanship_value == 4: return "Average (4)" elif workmanship_value == 5: return "Above Average (5)" elif workmanship_value == 6: return "Nearly flawless (6)" elif workmanship_value == 7: return "Flawless (7)" elif workmanship_value >= 8: return "Utterly flawless (8)" else: return f"Quality ({workmanship_value})" def format_damage_resistance(armor_level: int, damage_type: str) -> str: """Format damage resistance values with descriptive text.""" if armor_level <= 0: return "" # Rough categorization based on armor level if armor_level < 200: category = "Poor" elif armor_level < 300: category = "Below Average" elif armor_level < 400: category = "Average" elif armor_level < 500: category = "Above Average" elif armor_level < 600: category = "Excellent" else: category = "Superior" return f"{category} ({armor_level})" def get_damage_range_and_type(item_data: Dict[str, Any]) -> Dict[str, Any]: """Calculate damage range and determine damage type for weapons.""" damage_info = {} int_values = item_data.get('IntValues', {}) double_values = item_data.get('DoubleValues', {}) # Max damage max_damage = None if '218103842' in int_values: max_damage = int_values['218103842'] elif 218103842 in int_values: max_damage = int_values[218103842] # Variance for damage range calculation variance = None if '167772171' in double_values: variance = double_values['167772171'] elif 167772171 in double_values: variance = double_values[167772171] # Damage bonus damage_bonus = None if '167772174' in double_values: damage_bonus = double_values['167772174'] elif 167772174 in double_values: damage_bonus = double_values[167772174] if max_damage and variance: # Calculate min damage: max_damage * (2 - variance) / 2 min_damage = max_damage * (2 - variance) / 2 damage_info['damage_range'] = f"{min_damage:.2f} - {max_damage}" damage_info['min_damage'] = min_damage damage_info['max_damage'] = max_damage elif max_damage: damage_info['damage_range'] = str(max_damage) damage_info['max_damage'] = max_damage # Determine damage type from item name or properties item_name = item_data.get('Name', '').lower() if 'flaming' in item_name or 'fire' in item_name: damage_info['damage_type'] = 'Fire' elif 'frost' in item_name or 'ice' in item_name: damage_info['damage_type'] = 'Cold' elif 'lightning' in item_name or 'electric' in item_name: damage_info['damage_type'] = 'Electrical' elif 'acid' in item_name: damage_info['damage_type'] = 'Acid' else: # Check for elemental damage bonus elemental_bonus = None if '204' in int_values: elemental_bonus = int_values['204'] elif 204 in int_values: elemental_bonus = int_values[204] if elemental_bonus and elemental_bonus > 0: damage_info['damage_type'] = 'Elemental' else: damage_info['damage_type'] = 'Physical' return damage_info def get_weapon_speed(item_data: Dict[str, Any]) -> Dict[str, Any]: """Get weapon speed information.""" speed_info = {} int_values = item_data.get('IntValues', {}) # Weapon speed (check multiple possible keys) speed_value = None speed_keys = ['169', 169, 'WeapSpeed_Decal'] for key in speed_keys: if key in int_values: speed_value = int_values[key] break # Also check translated enum properties for speed translated_ints = translate_int_values(int_values) if 'WeaponSpeed' in translated_ints: speed_value = translated_ints['WeaponSpeed'] elif 'WeapSpeed' in translated_ints: speed_value = translated_ints['WeapSpeed'] elif 'WeapSpeed_Decal' in translated_ints: speed_value = translated_ints['WeapSpeed_Decal'] if speed_value: speed_info['speed_value'] = speed_value # Convert to descriptive text if speed_value <= 20: speed_info['speed_text'] = f"Very Fast ({speed_value})" elif speed_value <= 30: speed_info['speed_text'] = f"Fast ({speed_value})" elif speed_value <= 40: speed_info['speed_text'] = f"Average ({speed_value})" elif speed_value <= 50: speed_info['speed_text'] = f"Slow ({speed_value})" else: speed_info['speed_text'] = f"Very Slow ({speed_value})" return speed_info def get_mana_and_spellcraft(item_data: Dict[str, Any]) -> Dict[str, Any]: """Get mana and spellcraft information for items.""" mana_info = {} int_values = item_data.get('IntValues', {}) # Get translated enum properties first translated_ints = translate_int_values(int_values) # Current mana - check translated properties first current_mana = None if 'ItemCurMana' in translated_ints: current_mana = translated_ints['ItemCurMana'] elif '73' in int_values: current_mana = int_values['73'] elif 73 in int_values: current_mana = int_values[73] # Max mana - check translated properties first max_mana = None if 'ItemMaxMana' in translated_ints: max_mana = translated_ints['ItemMaxMana'] elif '72' in int_values: max_mana = int_values['72'] elif 72 in int_values: max_mana = int_values[72] # Spellcraft - check translated properties first spellcraft = None if 'ItemSpellcraft' in translated_ints: spellcraft = translated_ints['ItemSpellcraft'] elif '106' in int_values: spellcraft = int_values['106'] elif 106 in int_values: spellcraft = int_values[106] if current_mana is not None and max_mana is not None: mana_info['mana_display'] = f"{current_mana} / {max_mana}" mana_info['current_mana'] = current_mana mana_info['max_mana'] = max_mana if spellcraft: mana_info['spellcraft'] = spellcraft return mana_info def get_comprehensive_translations(item_data: Dict[str, Any]) -> Dict[str, Any]: """Get comprehensive translations for all aspects of an item.""" translations = {} # Translate IntValues int_values = item_data.get('IntValues', {}) if int_values: translations['int_properties'] = translate_int_values(int_values) # Translate material if present (check multiple locations) material_id = item_data.get('MaterialType') if material_id is None: # Check IntValues for material (key 131) int_values = item_data.get('IntValues', {}) if isinstance(int_values, dict) and '131' in int_values: material_id = int_values['131'] elif isinstance(int_values, dict) and 131 in int_values: material_id = int_values[131] if material_id is not None and material_id != 0: translations['material_name'] = translate_material_type(material_id) translations['material_id'] = material_id # Translate item type if present item_type_id = item_data.get('ItemType') if item_type_id is not None: translations['item_type_name'] = translate_item_type(item_type_id) # Translate object class using WeenieType enum object_class = item_data.get('ObjectClass') if object_class is not None: translations['object_class_name'] = translate_object_class(object_class, item_data) return translations def extract_item_properties(item_data: Dict[str, Any]) -> Dict[str, Any]: """Extract and categorize item properties from raw JSON.""" # Get raw values for comprehensive extraction int_values = item_data.get('IntValues', {}) double_values = item_data.get('DoubleValues', {}) # Start with processed fields (if available) properties = { 'basic': { 'name': item_data.get('Name', ''), 'icon': item_data.get('Icon', 0), 'object_class': item_data.get('ObjectClass', 0), 'value': item_data.get('Value', 0), 'burden': item_data.get('Burden', 0), 'has_id_data': item_data.get('HasIdData', False), # Equipment status - handle string keys properly 'current_wielded_location': int(int_values.get('10', int_values.get(10, '0'))), # Item state 'bonded': int_values.get('33', int_values.get(33, 0)), 'attuned': int_values.get('114', int_values.get(114, 0)), 'unique': int_values.get('279', int_values.get(279, 0)) != 0, # Stack/Container properties 'stack_size': int_values.get('12', int_values.get(12, 1)), 'max_stack_size': int_values.get('11', int_values.get(11, 1)), 'items_capacity': int_values.get('6', int_values.get(6, -1)), 'containers_capacity': int_values.get('7', int_values.get(7, -1)), # Durability 'structure': int_values.get('92', int_values.get(92, -1)), 'max_structure': int_values.get('91', int_values.get(91, -1)), # Special item flags 'rare_id': int_values.get('17', int_values.get(17, -1)), 'lifespan': int_values.get('267', int_values.get(267, -1)), 'remaining_lifespan': int_values.get('268', int_values.get(268, -1)), }, 'combat': { 'max_damage': item_data.get('MaxDamage', -1), 'armor_level': item_data.get('ArmorLevel', -1), 'damage_bonus': item_data.get('DamageBonus', -1.0), 'attack_bonus': item_data.get('AttackBonus', -1.0), # Defense bonuses from raw values 'melee_defense_bonus': double_values.get('29', double_values.get(29, -1.0)), 'magic_defense_bonus': double_values.get('150', double_values.get(150, -1.0)), 'missile_defense_bonus': double_values.get('149', double_values.get(149, -1.0)), 'elemental_damage_vs_monsters': double_values.get('152', double_values.get(152, -1.0)), 'mana_conversion_bonus': double_values.get('144', double_values.get(144, -1.0)), # Advanced damage properties 'cleaving': int_values.get('292', int_values.get(292, -1)), 'elemental_damage_bonus': int_values.get('204', int_values.get(204, -1)), 'crit_damage_rating': int_values.get('314', int_values.get(314, -1)), 'damage_over_time': int_values.get('318', int_values.get(318, -1)), # Resistances 'resist_magic': int_values.get('36', int_values.get(36, -1)), 'crit_resist_rating': int_values.get('315', int_values.get(315, -1)), 'crit_damage_resist_rating': int_values.get('316', int_values.get(316, -1)), 'dot_resist_rating': int_values.get('350', int_values.get(350, -1)), 'life_resist_rating': int_values.get('351', int_values.get(351, -1)), 'nether_resist_rating': int_values.get('331', int_values.get(331, -1)), # Healing/Recovery 'heal_over_time': int_values.get('312', int_values.get(312, -1)), 'healing_resist_rating': int_values.get('317', int_values.get(317, -1)), # PvP properties 'pk_damage_rating': int_values.get('381', int_values.get(381, -1)), 'pk_damage_resist_rating': int_values.get('382', int_values.get(382, -1)), 'gear_pk_damage_rating': int_values.get('383', int_values.get(383, -1)), 'gear_pk_damage_resist_rating': int_values.get('384', int_values.get(384, -1)), }, 'requirements': { 'wield_level': item_data.get('WieldLevel', -1), 'skill_level': item_data.get('SkillLevel', -1), 'lore_requirement': item_data.get('LoreRequirement', -1), 'equip_skill': item_data.get('EquipSkill'), }, 'enhancements': { 'material': item_data.get('Material'), 'imbue': item_data.get('Imbue'), 'tinks': item_data.get('Tinks', -1), 'workmanship': item_data.get('Workmanship', -1.0), 'item_set': int_values.get('265', int_values.get(265)) if int_values.get('265', int_values.get(265)) else None, # Advanced tinkering 'num_times_tinkered': int_values.get('171', int_values.get(171, -1)), 'free_tinkers_bitfield': int_values.get('264', int_values.get(264, -1)), 'num_items_in_material': int_values.get('170', int_values.get(170, -1)), # Additional imbue effects 'imbue_attempts': int_values.get('205', int_values.get(205, -1)), 'imbue_successes': int_values.get('206', int_values.get(206, -1)), 'imbued_effect2': int_values.get('303', int_values.get(303, -1)), 'imbued_effect3': int_values.get('304', int_values.get(304, -1)), 'imbued_effect4': int_values.get('305', int_values.get(305, -1)), 'imbued_effect5': int_values.get('306', int_values.get(306, -1)), 'imbue_stacking_bits': int_values.get('311', int_values.get(311, -1)), # Set information 'equipment_set_extra': int_values.get('321', int_values.get(321, -1)), # Special properties 'aetheria_bitfield': int_values.get('322', int_values.get(322, -1)), 'heritage_specific_armor': int_values.get('324', int_values.get(324, -1)), # Cooldowns 'shared_cooldown': int_values.get('280', int_values.get(280, -1)), }, 'ratings': { 'damage_rating': int_values.get('307', int_values.get(307, -1)), # DamageRating 'crit_rating': int_values.get('313', int_values.get(313, -1)), # CritRating 'crit_damage_rating': int_values.get('314', int_values.get(314, -1)), # CritDamageRating 'heal_boost_rating': int_values.get('323', int_values.get(323, -1)), # HealingBoostRating # Additional ratings 'weakness_rating': int_values.get('329', int_values.get(329, -1)), 'nether_over_time': int_values.get('330', int_values.get(330, -1)), # Gear totals 'gear_damage': int_values.get('370', int_values.get(370, -1)), 'gear_damage_resist': int_values.get('371', int_values.get(371, -1)), 'gear_crit': int_values.get('372', int_values.get(372, -1)), 'gear_crit_resist': int_values.get('373', int_values.get(373, -1)), 'gear_crit_damage': int_values.get('374', int_values.get(374, -1)), 'gear_crit_damage_resist': int_values.get('375', int_values.get(375, -1)), 'gear_healing_boost': int_values.get('376', int_values.get(376, -1)), 'gear_max_health': int_values.get('379', int_values.get(379, -1)), 'gear_nether_resist': int_values.get('377', int_values.get(377, -1)), 'gear_life_resist': int_values.get('378', int_values.get(378, -1)), 'gear_overpower': int_values.get('388', int_values.get(388, -1)), 'gear_overpower_resist': int_values.get('389', int_values.get(389, -1)), } } # Get comprehensive translations translations = get_comprehensive_translations(item_data) if translations: properties['translations'] = translations # Translate raw enum values if available int_values = item_data.get('IntValues', {}) if int_values: translated_ints = translate_int_values(int_values) properties['translated_ints'] = translated_ints # Extract spell information spells = item_data.get('Spells', []) active_spells = item_data.get('ActiveSpells', []) # Translate spell IDs to spell data translated_spells = [] for spell_id in spells: translated_spells.append(translate_spell(spell_id)) translated_active_spells = [] for spell_id in active_spells: translated_active_spells.append(translate_spell(spell_id)) # Get spell-related properties from IntValues int_values = item_data.get('IntValues', {}) spell_info = { 'spell_ids': spells, 'active_spell_ids': active_spells, 'spell_count': len(spells), 'active_spell_count': len(active_spells), 'spells': translated_spells, 'active_spells': translated_active_spells } # Extract spell-related IntValues for key_str, value in int_values.items(): key_int = int(key_str) if key_str.isdigit() else None if key_int: # Check for spell-related properties if key_int == 94: # TargetType/SpellDID spell_info['spell_target_type'] = value elif key_int == 106: # ItemSpellcraft spell_info['item_spellcraft'] = value elif key_int in [218103816, 218103838, 218103848]: # Spell decals spell_info[f'spell_decal_{key_int}'] = value properties['spells'] = spell_info # Add weapon-specific information damage_info = get_damage_range_and_type(item_data) if damage_info: properties['weapon_damage'] = damage_info speed_info = get_weapon_speed(item_data) if speed_info: properties['weapon_speed'] = speed_info mana_info = get_mana_and_spellcraft(item_data) if mana_info: properties['mana_info'] = mana_info return properties # API endpoints @app.post("/process-inventory") async def process_inventory(inventory: InventoryItem): """Process raw inventory data and store in normalized format.""" processed_count = 0 error_count = 0 async with database.transaction(): # First, delete all existing items for this character from all related tables # Get item IDs to delete item_ids_query = "SELECT id FROM items WHERE character_name = :character_name" item_ids = await database.fetch_all(item_ids_query, {"character_name": inventory.character_name}) if item_ids: id_list = [str(row['id']) for row in item_ids] id_placeholder = ','.join(id_list) # Delete from all related tables first await database.execute(f"DELETE FROM item_raw_data WHERE item_id IN ({id_placeholder})") await database.execute(f"DELETE FROM item_combat_stats WHERE item_id IN ({id_placeholder})") await database.execute(f"DELETE FROM item_requirements WHERE item_id IN ({id_placeholder})") await database.execute(f"DELETE FROM item_enhancements WHERE item_id IN ({id_placeholder})") await database.execute(f"DELETE FROM item_ratings WHERE item_id IN ({id_placeholder})") await database.execute(f"DELETE FROM item_spells WHERE item_id IN ({id_placeholder})") # Finally delete from main items table await database.execute( "DELETE FROM items WHERE character_name = :character_name", {"character_name": inventory.character_name} ) # Then insert the new complete inventory for item_data in inventory.items: try: # Extract properties properties = extract_item_properties(item_data) # Create core item record item_id = item_data.get('Id') if item_id is None: logger.warning(f"Skipping item without ID: {item_data.get('Name', 'Unknown')}") error_count += 1 continue # Insert or update core item (handle timezone-aware timestamps) timestamp = inventory.timestamp if timestamp.tzinfo is not None: timestamp = timestamp.replace(tzinfo=None) # Simple INSERT since we cleared the table first basic = properties['basic'] item_stmt = sa.insert(Item).values( character_name=inventory.character_name, item_id=item_id, timestamp=timestamp, name=basic['name'], icon=basic['icon'], object_class=basic['object_class'], value=basic['value'], burden=basic['burden'], has_id_data=basic['has_id_data'], last_id_time=item_data.get('LastIdTime', 0), # Equipment status current_wielded_location=basic['current_wielded_location'], # Item state bonded=basic['bonded'], attuned=basic['attuned'], unique=basic['unique'], # Stack/Container properties stack_size=basic['stack_size'], max_stack_size=basic['max_stack_size'], items_capacity=basic['items_capacity'] if basic['items_capacity'] != -1 else None, containers_capacity=basic['containers_capacity'] if basic['containers_capacity'] != -1 else None, # Durability structure=basic['structure'] if basic['structure'] != -1 else None, max_structure=basic['max_structure'] if basic['max_structure'] != -1 else None, # Special item flags rare_id=basic['rare_id'] if basic['rare_id'] != -1 else None, lifespan=basic['lifespan'] if basic['lifespan'] != -1 else None, remaining_lifespan=basic['remaining_lifespan'] if basic['remaining_lifespan'] != -1 else None, ).returning(Item.id) result = await database.fetch_one(item_stmt) db_item_id = result['id'] # Store combat stats if applicable combat = properties['combat'] if any(v != -1 and v != -1.0 for v in combat.values()): combat_stmt = sa.dialects.postgresql.insert(ItemCombatStats).values( item_id=db_item_id, **{k: v if v != -1 and v != -1.0 else None for k, v in combat.items()} ).on_conflict_do_update( index_elements=['item_id'], set_=dict(**{k: v if v != -1 and v != -1.0 else None for k, v in combat.items()}) ) await database.execute(combat_stmt) # Store requirements if applicable requirements = properties['requirements'] if any(v not in [-1, None, ''] for v in requirements.values()): req_stmt = sa.dialects.postgresql.insert(ItemRequirements).values( item_id=db_item_id, **{k: v if v not in [-1, None, ''] else None for k, v in requirements.items()} ).on_conflict_do_update( index_elements=['item_id'], set_=dict(**{k: v if v not in [-1, None, ''] else None for k, v in requirements.items()}) ) await database.execute(req_stmt) # Store enhancements - always create record to capture item_set data enhancements = properties['enhancements'] enh_stmt = sa.dialects.postgresql.insert(ItemEnhancements).values( item_id=db_item_id, **{k: v if v not in [-1, -1.0, None, ''] else None for k, v in enhancements.items()} ).on_conflict_do_update( index_elements=['item_id'], set_=dict(**{k: v if v not in [-1, -1.0, None, ''] else None for k, v in enhancements.items()}) ) await database.execute(enh_stmt) # Store ratings if applicable ratings = properties['ratings'] if any(v not in [-1, -1.0, None] for v in ratings.values()): rat_stmt = sa.dialects.postgresql.insert(ItemRatings).values( item_id=db_item_id, **{k: v if v not in [-1, -1.0, None] else None for k, v in ratings.items()} ).on_conflict_do_update( index_elements=['item_id'], set_=dict(**{k: v if v not in [-1, -1.0, None] else None for k, v in ratings.items()}) ) await database.execute(rat_stmt) # Store spell data if applicable spells = item_data.get('Spells', []) active_spells = item_data.get('ActiveSpells', []) all_spells = set(spells + active_spells) if all_spells: # First delete existing spells for this item await database.execute( "DELETE FROM item_spells WHERE item_id = :item_id", {"item_id": db_item_id} ) # Insert all spells for this item for spell_id in all_spells: is_active = spell_id in active_spells spell_stmt = sa.dialects.postgresql.insert(ItemSpells).values( item_id=db_item_id, spell_id=spell_id, is_active=is_active ).on_conflict_do_nothing() await database.execute(spell_stmt) # Store raw data for completeness raw_stmt = sa.dialects.postgresql.insert(ItemRawData).values( item_id=db_item_id, int_values=item_data.get('IntValues', {}), double_values=item_data.get('DoubleValues', {}), string_values=item_data.get('StringValues', {}), bool_values=item_data.get('BoolValues', {}), original_json=item_data ).on_conflict_do_update( index_elements=['item_id'], set_=dict( int_values=item_data.get('IntValues', {}), double_values=item_data.get('DoubleValues', {}), string_values=item_data.get('StringValues', {}), bool_values=item_data.get('BoolValues', {}), original_json=item_data ) ) await database.execute(raw_stmt) processed_count += 1 except Exception as e: logger.error(f"Error processing item {item_data.get('Id', 'unknown')}: {e}") error_count += 1 logger.info(f"Inventory processing complete for {inventory.character_name}: {processed_count} processed, {error_count} errors, {len(inventory.items)} total items received") return { "status": "completed", "processed": processed_count, "errors": error_count, "total_received": len(inventory.items), "character": inventory.character_name } @app.get("/inventory/{character_name}") async def get_character_inventory( character_name: str, limit: int = Query(1000, le=5000), offset: int = Query(0, ge=0) ): """Get processed inventory for a character with structured data and comprehensive translations.""" query = """ SELECT i.id, i.name, i.icon, i.object_class, i.value, i.burden, i.has_id_data, i.timestamp, cs.max_damage, cs.armor_level, cs.damage_bonus, cs.attack_bonus, r.wield_level, r.skill_level, r.equip_skill, r.lore_requirement, e.material, e.imbue, e.item_set, e.tinks, e.workmanship, rt.damage_rating, rt.crit_rating, rt.heal_boost_rating, rd.int_values, rd.double_values, rd.string_values, rd.bool_values, rd.original_json FROM items i LEFT JOIN item_combat_stats cs ON i.id = cs.item_id LEFT JOIN item_requirements r ON i.id = r.item_id LEFT JOIN item_enhancements e ON i.id = e.item_id LEFT JOIN item_ratings rt ON i.id = rt.item_id LEFT JOIN item_raw_data rd ON i.id = rd.item_id WHERE i.character_name = :character_name ORDER BY i.name LIMIT :limit OFFSET :offset """ items = await database.fetch_all(query, { "character_name": character_name, "limit": limit, "offset": offset }) if not items: raise HTTPException(status_code=404, detail=f"No inventory found for {character_name}") # Convert to structured format with enhanced translations processed_items = [] for item in items: processed_item = dict(item) # Get comprehensive translations from original_json if processed_item.get('original_json'): original_json = processed_item['original_json'] # Handle case where original_json might be stored as string if isinstance(original_json, str): try: original_json = json.loads(original_json) except (json.JSONDecodeError, TypeError): original_json = {} if original_json: # Extract properties and get translations properties = extract_item_properties(original_json) # Add translated properties to the item processed_item['translated_properties'] = properties # Add material translation if processed_item.get('material'): processed_item['material_name'] = translate_material_type(processed_item['material']) # Add object class translation if processed_item.get('object_class'): processed_item['object_class_name'] = translate_object_class(processed_item['object_class'], original_json) # Add skill translation if processed_item.get('equip_skill'): processed_item['equip_skill_name'] = translate_skill(processed_item['equip_skill']) # Flatten the structure - move translated properties to top level if 'translated_properties' in processed_item: translated_props = processed_item.pop('translated_properties') # Move spells to top level if 'spells' in translated_props: processed_item['spells'] = translated_props['spells'] # Move translated_ints to top level for enhanced tooltips if 'translated_ints' in translated_props: processed_item['enhanced_properties'] = translated_props['translated_ints'] # Add weapon damage information if 'weapon_damage' in translated_props: weapon_damage = translated_props['weapon_damage'] if weapon_damage.get('damage_range'): processed_item['damage_range'] = weapon_damage['damage_range'] if weapon_damage.get('damage_type'): processed_item['damage_type'] = weapon_damage['damage_type'] if weapon_damage.get('min_damage'): processed_item['min_damage'] = weapon_damage['min_damage'] # Add weapon speed information if 'weapon_speed' in translated_props: speed_info = translated_props['weapon_speed'] if speed_info.get('speed_text'): processed_item['speed_text'] = speed_info['speed_text'] if speed_info.get('speed_value'): processed_item['speed_value'] = speed_info['speed_value'] # Add mana and spellcraft information if 'mana_info' in translated_props: mana_info = translated_props['mana_info'] if mana_info.get('mana_display'): processed_item['mana_display'] = mana_info['mana_display'] if mana_info.get('spellcraft'): processed_item['spellcraft'] = mana_info['spellcraft'] if mana_info.get('current_mana'): processed_item['current_mana'] = mana_info['current_mana'] if mana_info.get('max_mana'): processed_item['max_mana'] = mana_info['max_mana'] # Add icon overlay/underlay information from translated properties if 'translated_ints' in translated_props: translated_ints = translated_props['translated_ints'] # Icon overlay - check for the proper property name if 'IconOverlay_Decal_DID' in translated_ints and translated_ints['IconOverlay_Decal_DID'] > 0: processed_item['icon_overlay_id'] = translated_ints['IconOverlay_Decal_DID'] # Icon underlay - check for the proper property name if 'IconUnderlay_Decal_DID' in translated_ints and translated_ints['IconUnderlay_Decal_DID'] > 0: processed_item['icon_underlay_id'] = translated_ints['IconUnderlay_Decal_DID'] # Add comprehensive combat stats if 'combat' in translated_props: combat_stats = translated_props['combat'] if combat_stats.get('max_damage', -1) != -1: processed_item['max_damage'] = combat_stats['max_damage'] if combat_stats.get('armor_level', -1) != -1: processed_item['armor_level'] = combat_stats['armor_level'] if combat_stats.get('damage_bonus', -1.0) != -1.0: processed_item['damage_bonus'] = combat_stats['damage_bonus'] if combat_stats.get('attack_bonus', -1.0) != -1.0: processed_item['attack_bonus'] = combat_stats['attack_bonus'] # Add missing combat bonuses if combat_stats.get('melee_defense_bonus', -1.0) != -1.0: processed_item['melee_defense_bonus'] = combat_stats['melee_defense_bonus'] if combat_stats.get('magic_defense_bonus', -1.0) != -1.0: processed_item['magic_defense_bonus'] = combat_stats['magic_defense_bonus'] if combat_stats.get('missile_defense_bonus', -1.0) != -1.0: processed_item['missile_defense_bonus'] = combat_stats['missile_defense_bonus'] if combat_stats.get('elemental_damage_vs_monsters', -1.0) != -1.0: processed_item['elemental_damage_vs_monsters'] = combat_stats['elemental_damage_vs_monsters'] if combat_stats.get('mana_conversion_bonus', -1.0) != -1.0: processed_item['mana_conversion_bonus'] = combat_stats['mana_conversion_bonus'] # Add comprehensive requirements if 'requirements' in translated_props: requirements = translated_props['requirements'] if requirements.get('wield_level', -1) != -1: processed_item['wield_level'] = requirements['wield_level'] if requirements.get('skill_level', -1) != -1: processed_item['skill_level'] = requirements['skill_level'] if requirements.get('lore_requirement', -1) != -1: processed_item['lore_requirement'] = requirements['lore_requirement'] if requirements.get('equip_skill'): processed_item['equip_skill'] = requirements['equip_skill'] # Add comprehensive enhancements if 'enhancements' in translated_props: enhancements = translated_props['enhancements'] if enhancements.get('material'): processed_item['material'] = enhancements['material'] # Add material information from translations if 'translations' in translated_props: trans = translated_props['translations'] if trans.get('material_name'): processed_item['material_name'] = trans['material_name'] if trans.get('material_id'): processed_item['material_id'] = trans['material_id'] # Continue with other enhancements if 'enhancements' in translated_props: enhancements = translated_props['enhancements'] if enhancements.get('imbue'): processed_item['imbue'] = enhancements['imbue'] if enhancements.get('tinks', -1) != -1: processed_item['tinks'] = enhancements['tinks'] if enhancements.get('workmanship', -1.0) != -1.0: processed_item['workmanship'] = enhancements['workmanship'] processed_item['workmanship_text'] = translate_workmanship(int(enhancements['workmanship'])) if enhancements.get('item_set'): processed_item['item_set'] = enhancements['item_set'] # Add equipment set name translation set_id = str(enhancements['item_set']).strip() dictionaries = ENUM_MAPPINGS.get('dictionaries', {}) attribute_set_info = dictionaries.get('AttributeSetInfo', {}).get('values', {}) if set_id in attribute_set_info: processed_item['item_set_name'] = attribute_set_info[set_id] else: processed_item['item_set_name'] = f"Set {set_id}" # Add comprehensive ratings (use gear totals as fallback for armor/clothing) if 'ratings' in translated_props: ratings = translated_props['ratings'] # Damage rating: use individual rating or gear total if ratings.get('damage_rating', -1) != -1: processed_item['damage_rating'] = ratings['damage_rating'] elif ratings.get('gear_damage', -1) > 0: processed_item['damage_rating'] = ratings['gear_damage'] # Crit rating if ratings.get('crit_rating', -1) != -1: processed_item['crit_rating'] = ratings['crit_rating'] elif ratings.get('gear_crit', -1) > 0: processed_item['crit_rating'] = ratings['gear_crit'] # Crit damage rating: use individual rating or gear total if ratings.get('crit_damage_rating', -1) != -1: processed_item['crit_damage_rating'] = ratings['crit_damage_rating'] elif ratings.get('gear_crit_damage', -1) > 0: processed_item['crit_damage_rating'] = ratings['gear_crit_damage'] # Heal boost rating: use individual rating or gear total if ratings.get('heal_boost_rating', -1) != -1: processed_item['heal_boost_rating'] = ratings['heal_boost_rating'] elif ratings.get('gear_healing_boost', -1) > 0: processed_item['heal_boost_rating'] = ratings['gear_healing_boost'] # Apply material prefix to item name if material exists if processed_item.get('material_name') and processed_item.get('name'): original_name = processed_item['name'] material_name = processed_item['material_name'] # Don't add prefix if name already starts with the material if not original_name.lower().startswith(material_name.lower()): processed_item['name'] = f"{material_name} {original_name}" processed_item['original_name'] = original_name # Preserve original for reference # Remove raw data from response (keep clean output) processed_item.pop('int_values', None) processed_item.pop('double_values', None) processed_item.pop('string_values', None) processed_item.pop('bool_values', None) processed_item.pop('original_json', None) # Remove null values for cleaner response processed_item = {k: v for k, v in processed_item.items() if v is not None} processed_items.append(processed_item) return { "character_name": character_name, "item_count": len(processed_items), "items": processed_items } @app.get("/health") async def health_check(): """Health check endpoint.""" return {"status": "healthy", "service": "inventory-service"} @app.get("/sets/list") async def list_equipment_sets(): """Get all unique equipment set names from the database.""" try: # Get equipment set IDs (the numeric collection sets) query = """ SELECT DISTINCT enh.item_set, COUNT(*) as item_count FROM item_enhancements enh WHERE enh.item_set IS NOT NULL AND enh.item_set != '' GROUP BY enh.item_set ORDER BY item_count DESC, enh.item_set """ rows = await database.fetch_all(query) # Get AttributeSetInfo mapping from enum database attribute_set_info = ENUM_MAPPINGS.get('AttributeSetInfo', {}).get('values', {}) # Map equipment sets to proper names equipment_sets = [] for row in rows: set_id = row["item_set"] item_count = row["item_count"] # Get proper name from AttributeSetInfo mapping set_name = attribute_set_info.get(set_id, f"Unknown Set {set_id}") equipment_sets.append({ "id": set_id, "name": set_name, "item_count": item_count }) return { "equipment_sets": equipment_sets, "total_sets": len(equipment_sets) } except Exception as e: logger.error(f"Failed to list equipment sets: {e}") raise HTTPException(status_code=500, detail="Failed to list equipment sets") @app.get("/enum-info") async def get_enum_info(): """Get information about available enum translations.""" if ENUM_MAPPINGS is None: return {"error": "Enum database not loaded"} return { "available_enums": list(ENUM_MAPPINGS.keys()), "int_values_count": len(ENUM_MAPPINGS.get('int_values', {})), "materials_count": len(ENUM_MAPPINGS.get('materials', {})), "item_types_count": len(ENUM_MAPPINGS.get('item_types', {})), "skills_count": len(ENUM_MAPPINGS.get('skills', {})), "spell_categories_count": len(ENUM_MAPPINGS.get('spell_categories', {})), "spells_count": len(ENUM_MAPPINGS.get('spells', {})), "object_classes_count": len(ENUM_MAPPINGS.get('object_classes', {})), "coverage_masks_count": len(ENUM_MAPPINGS.get('coverage_masks', {})), "database_version": ENUM_MAPPINGS.get('full_database', {}).get('metadata', {}).get('version', 'unknown') } @app.get("/translate/{enum_type}/{value}") async def translate_enum_value(enum_type: str, value: int): """Translate a specific enum value to human-readable name.""" if enum_type == "material": return {"translation": translate_material_type(value)} elif enum_type == "item_type": return {"translation": translate_item_type(value)} elif enum_type == "skill": return {"translation": translate_skill(value)} elif enum_type == "spell_category": return {"translation": translate_spell_category(value)} elif enum_type == "spell": return {"translation": translate_spell(value)} elif enum_type == "int_value": int_enums = ENUM_MAPPINGS.get('int_values', {}) return {"translation": int_enums.get(value, f"unknown_{value}")} else: raise HTTPException(status_code=400, detail=f"Unknown enum type: {enum_type}") @app.get("/inventory/{character_name}/raw") async def get_character_inventory_raw(character_name: str): """Get raw inventory data including comprehensive translations.""" query = """ SELECT i.name, i.icon, i.object_class, i.value, i.burden, i.timestamp, rd.int_values, rd.double_values, rd.string_values, rd.bool_values, rd.original_json FROM items i LEFT JOIN item_raw_data rd ON i.id = rd.item_id WHERE i.character_name = :character_name ORDER BY i.name LIMIT 100 """ items = await database.fetch_all(query, {"character_name": character_name}) if not items: raise HTTPException(status_code=404, detail=f"No inventory found for {character_name}") # Process items with comprehensive translations processed_items = [] for item in items: item_dict = dict(item) # Add comprehensive translations for raw data if item_dict.get('original_json'): original_json = item_dict['original_json'] # Handle case where original_json might be stored as string if isinstance(original_json, str): try: original_json = json.loads(original_json) except (json.JSONDecodeError, TypeError): logger.warning(f"Failed to parse original_json as JSON for item") original_json = {} if original_json: translations = get_comprehensive_translations(original_json) item_dict['comprehensive_translations'] = translations processed_items.append(item_dict) return { "character_name": character_name, "item_count": len(processed_items), "items": processed_items } # =================================================================== # INVENTORY SEARCH API ENDPOINTS # =================================================================== @app.get("/search/items") async def search_items( # Text search text: str = Query(None, description="Search item names, descriptions, or properties"), character: str = Query(None, description="Limit search to specific character"), characters: str = Query(None, description="Comma-separated list of character names"), include_all_characters: bool = Query(False, description="Search across all characters"), # Equipment filtering equipment_status: str = Query(None, description="equipped, unequipped, or all"), equipment_slot: int = Query(None, description="Equipment slot mask (e.g., 1=head, 512=chest)"), # Item category filtering armor_only: bool = Query(False, description="Show only armor items"), jewelry_only: bool = Query(False, description="Show only jewelry items"), weapon_only: bool = Query(False, description="Show only weapon items"), # Spell filtering has_spell: str = Query(None, description="Must have this specific spell (by name)"), spell_contains: str = Query(None, description="Spell name contains this text"), legendary_cantrips: str = Query(None, description="Comma-separated list of legendary cantrip names"), # Combat properties min_damage: int = Query(None, description="Minimum damage"), max_damage: int = Query(None, description="Maximum damage"), min_armor: int = Query(None, description="Minimum armor level"), max_armor: int = Query(None, description="Maximum armor level"), min_attack_bonus: float = Query(None, description="Minimum attack bonus"), min_crit_damage_rating: int = Query(None, description="Minimum critical damage rating"), min_damage_rating: int = Query(None, description="Minimum damage rating"), min_heal_boost_rating: int = Query(None, description="Minimum heal boost rating"), # Requirements max_level: int = Query(None, description="Maximum wield level requirement"), min_level: int = Query(None, description="Minimum wield level requirement"), # Enhancements material: str = Query(None, description="Material type (partial match)"), min_workmanship: float = Query(None, description="Minimum workmanship"), has_imbue: bool = Query(None, description="Has imbue effects"), item_set: str = Query(None, description="Item set ID (single set)"), item_sets: str = Query(None, description="Comma-separated list of item set IDs"), min_tinks: int = Query(None, description="Minimum tinker count"), # Item state bonded: bool = Query(None, description="Bonded status"), attuned: bool = Query(None, description="Attuned status"), unique: bool = Query(None, description="Unique item status"), is_rare: bool = Query(None, description="Rare item status"), min_condition: int = Query(None, description="Minimum condition percentage"), # Value/utility min_value: int = Query(None, description="Minimum item value"), max_value: int = Query(None, description="Maximum item value"), max_burden: int = Query(None, description="Maximum burden"), # Sorting and pagination sort_by: str = Query("name", description="Sort field: name, value, damage, armor, workmanship, damage_rating, crit_damage_rating, level"), sort_dir: str = Query("asc", description="Sort direction: asc or desc"), page: int = Query(1, ge=1, description="Page number"), limit: int = Query(200, ge=1, le=2000, description="Items per page") ): """ Search items across characters with comprehensive filtering options. """ try: # Build base query - include raw data for comprehensive translations query_parts = [""" SELECT DISTINCT i.id as db_item_id, i.character_name, i.name, i.icon, i.object_class, i.value, i.burden, i.current_wielded_location, i.bonded, i.attuned, i.unique, i.stack_size, i.max_stack_size, i.structure, i.max_structure, i.rare_id, COALESCE(cs.max_damage, -1) as max_damage, COALESCE(cs.armor_level, -1) as armor_level, COALESCE(cs.attack_bonus, -1.0) as attack_bonus, GREATEST( COALESCE((rd.int_values->>'314')::int, -1), COALESCE((rd.int_values->>'374')::int, -1) ) as crit_damage_rating, GREATEST( COALESCE((rd.int_values->>'307')::int, -1), COALESCE((rd.int_values->>'370')::int, -1) ) as damage_rating, GREATEST( COALESCE((rd.int_values->>'323')::int, -1), COALESCE((rd.int_values->>'376')::int, -1) ) as heal_boost_rating, COALESCE(req.wield_level, -1) as wield_level, COALESCE(enh.material, '') as material, COALESCE(enh.workmanship, -1.0) as workmanship, COALESCE(enh.imbue, '') as imbue, COALESCE(enh.tinks, -1) as tinks, COALESCE(enh.item_set, '') as item_set, rd.original_json FROM items i LEFT JOIN item_combat_stats cs ON i.id = cs.item_id LEFT JOIN item_requirements req ON i.id = req.item_id LEFT JOIN item_enhancements enh ON i.id = enh.item_id LEFT JOIN item_ratings rt ON i.id = rt.item_id LEFT JOIN item_raw_data rd ON i.id = rd.item_id """] conditions = [] params = {} # Character filtering if character: conditions.append("i.character_name = :character") params["character"] = character elif characters: # Handle comma-separated list of characters character_list = [char.strip() for char in characters.split(',') if char.strip()] if character_list: # Create parameterized IN clause char_params = [] for i, char_name in enumerate(character_list): param_name = f"char_{i}" char_params.append(f":{param_name}") params[param_name] = char_name conditions.append(f"i.character_name IN ({', '.join(char_params)})") else: return { "error": "Empty characters list provided", "items": [], "total_count": 0 } elif not include_all_characters: # Default to requiring character parameter if not searching all return { "error": "Must specify character, characters, or set include_all_characters=true", "items": [], "total_count": 0 } # Text search (name) if text: conditions.append("i.name ILIKE :text") params["text"] = f"%{text}%" # Item category filtering if armor_only: # Armor: ObjectClass 2 (Clothing) or 3 (Armor) with armor_level > 0 conditions.append("(i.object_class IN (2, 3) AND COALESCE(cs.armor_level, 0) > 0)") elif jewelry_only: # Jewelry: ObjectClass 4 (Jewelry) - rings, bracelets, necklaces, amulets conditions.append("i.object_class = 4") elif weapon_only: # Weapons: ObjectClass 6 (MeleeWeapon), 7 (MissileWeapon), 8 (Caster) with max_damage > 0 conditions.append("(i.object_class IN (6, 7, 8) AND COALESCE(cs.max_damage, 0) > 0)") # Spell filtering - need to join with item_spells and use spell database spell_join_added = False if has_spell or spell_contains or legendary_cantrips: query_parts[0] = query_parts[0].replace( "LEFT JOIN item_ratings rt ON i.id = rt.item_id", """LEFT JOIN item_ratings rt ON i.id = rt.item_id LEFT JOIN item_spells sp ON i.id = sp.item_id""" ) spell_join_added = True spell_conditions = [] if has_spell: # Look up spell ID by exact name match in ENUM_MAPPINGS spell_id = None spells = ENUM_MAPPINGS.get('spells', {}) for sid, spell_data in spells.items(): if isinstance(spell_data, dict) and spell_data.get('name', '').lower() == has_spell.lower(): spell_id = sid break if spell_id: spell_conditions.append("sp.spell_id = :has_spell_id") params["has_spell_id"] = spell_id else: # If spell not found by exact name, no results conditions.append("1 = 0") if spell_contains: # Find all spell IDs that contain the text matching_spell_ids = [] spells = ENUM_MAPPINGS.get('spells', {}) for sid, spell_data in spells.items(): if isinstance(spell_data, dict): spell_name = spell_data.get('name', '').lower() if spell_contains.lower() in spell_name: matching_spell_ids.append(sid) if matching_spell_ids: spell_conditions.append(f"sp.spell_id IN ({','.join(map(str, matching_spell_ids))})") else: # If no spells found containing the text, no results conditions.append("1 = 0") if legendary_cantrips: # Parse comma-separated list of cantrip names cantrip_names = [name.strip() for name in legendary_cantrips.split(',')] matching_spell_ids = [] spells = ENUM_MAPPINGS.get('spells', {}) logger.info(f"Looking for cantrips: {cantrip_names}") for cantrip_name in cantrip_names: found_match = False for sid, spell_data in spells.items(): if isinstance(spell_data, dict): spell_name = spell_data.get('name', '').lower() # Match cantrip name (flexible matching) if cantrip_name.lower() in spell_name or spell_name in cantrip_name.lower(): matching_spell_ids.append(sid) found_match = True logger.info(f"Found spell match: {spell_name} (ID: {sid}) for cantrip: {cantrip_name}") if not found_match: logger.warning(f"No spell found matching cantrip: {cantrip_name}") if matching_spell_ids: # Remove duplicates matching_spell_ids = list(set(matching_spell_ids)) spell_conditions.append(f"sp.spell_id IN ({','.join(map(str, matching_spell_ids))})") logger.info(f"Found {len(matching_spell_ids)} matching spell IDs: {matching_spell_ids}") else: # If no matching cantrips found, this will return no results logger.warning("No matching spells found for any cantrips - search will return empty results") spell_conditions.append("sp.spell_id = -1") # Use impossible condition instead of 1=0 # Combine spell conditions with OR - only add if we have conditions if spell_conditions: conditions.append(f"({' OR '.join(spell_conditions)})") # Equipment status if equipment_status == "equipped": conditions.append("i.current_wielded_location > 0") elif equipment_status == "unequipped": conditions.append("i.current_wielded_location = 0") # Equipment slot if equipment_slot is not None: conditions.append("i.current_wielded_location = :equipment_slot") params["equipment_slot"] = equipment_slot # Combat properties if min_damage is not None: conditions.append("cs.max_damage >= :min_damage") params["min_damage"] = min_damage if max_damage is not None: conditions.append("cs.max_damage <= :max_damage") params["max_damage"] = max_damage if min_armor is not None: conditions.append("cs.armor_level >= :min_armor") params["min_armor"] = min_armor if max_armor is not None: conditions.append("cs.armor_level <= :max_armor") params["max_armor"] = max_armor if min_attack_bonus is not None: conditions.append("cs.attack_bonus >= :min_attack_bonus") params["min_attack_bonus"] = min_attack_bonus if min_crit_damage_rating is not None: # Check both individual rating (314) and gear total (374) conditions.append("""( COALESCE((rd.int_values->>'314')::int, 0) >= :min_crit_damage_rating OR COALESCE((rd.int_values->>'374')::int, 0) >= :min_crit_damage_rating )""") params["min_crit_damage_rating"] = min_crit_damage_rating if min_damage_rating is not None: # Check both individual rating (307) and gear total (370) conditions.append("""( COALESCE((rd.int_values->>'307')::int, 0) >= :min_damage_rating OR COALESCE((rd.int_values->>'370')::int, 0) >= :min_damage_rating )""") params["min_damage_rating"] = min_damage_rating if min_heal_boost_rating is not None: # Check both individual rating (323) and gear total (376) conditions.append("""( COALESCE((rd.int_values->>'323')::int, 0) >= :min_heal_boost_rating OR COALESCE((rd.int_values->>'376')::int, 0) >= :min_heal_boost_rating )""") params["min_heal_boost_rating"] = min_heal_boost_rating # Requirements if max_level is not None: conditions.append("(req.wield_level <= :max_level OR req.wield_level IS NULL)") params["max_level"] = max_level if min_level is not None: conditions.append("req.wield_level >= :min_level") params["min_level"] = min_level # Enhancements if material: conditions.append("enh.material ILIKE :material") params["material"] = f"%{material}%" if min_workmanship is not None: conditions.append("enh.workmanship >= :min_workmanship") params["min_workmanship"] = min_workmanship if has_imbue is not None: if has_imbue: conditions.append("enh.imbue IS NOT NULL AND enh.imbue != ''") else: conditions.append("(enh.imbue IS NULL OR enh.imbue = '')") if item_set: conditions.append("enh.item_set = :item_set") params["item_set"] = item_set elif item_sets: # Handle comma-separated list of item set IDs set_list = [set_id.strip() for set_id in item_sets.split(',') if set_id.strip()] if set_list: # Create parameterized IN clause set_params = [] for i, set_id in enumerate(set_list): param_name = f"set_{i}" set_params.append(f":{param_name}") params[param_name] = set_id conditions.append(f"enh.item_set IN ({', '.join(set_params)})") else: # Empty sets list - no results conditions.append("1 = 0") if min_tinks is not None: conditions.append("enh.tinks >= :min_tinks") params["min_tinks"] = min_tinks # Item state if bonded is not None: conditions.append("i.bonded > 0" if bonded else "i.bonded = 0") if attuned is not None: conditions.append("i.attuned > 0" if attuned else "i.attuned = 0") if unique is not None: conditions.append("i.unique = :unique") params["unique"] = unique if is_rare is not None: if is_rare: conditions.append("i.rare_id IS NOT NULL AND i.rare_id > 0") else: conditions.append("(i.rare_id IS NULL OR i.rare_id <= 0)") if min_condition is not None: conditions.append("((i.structure * 100.0 / NULLIF(i.max_structure, 0)) >= :min_condition OR i.max_structure IS NULL)") params["min_condition"] = min_condition # Value/utility if min_value is not None: conditions.append("i.value >= :min_value") params["min_value"] = min_value if max_value is not None: conditions.append("i.value <= :max_value") params["max_value"] = max_value if max_burden is not None: conditions.append("i.burden <= :max_burden") params["max_burden"] = max_burden # Build WHERE clause if conditions: query_parts.append("WHERE " + " AND ".join(conditions)) # Add ORDER BY sort_mapping = { "name": "i.name", "value": "i.value", "damage": "cs.max_damage", "armor": "cs.armor_level", "workmanship": "enh.workmanship", "level": "req.wield_level", "damage_rating": "damage_rating", "crit_damage_rating": "crit_damage_rating", "heal_boost_rating": "heal_boost_rating" } sort_field = sort_mapping.get(sort_by, "i.name") sort_direction = "DESC" if sort_dir.lower() == "desc" else "ASC" query_parts.append(f"ORDER BY {sort_field} {sort_direction}") # Add pagination offset = (page - 1) * limit query_parts.append(f"LIMIT {limit} OFFSET {offset}") # Execute query query = "\n".join(query_parts) rows = await database.fetch_all(query, params) # Get total count for pagination - build separate count query count_query = """ SELECT COUNT(DISTINCT i.id) FROM items i LEFT JOIN item_combat_stats cs ON i.id = cs.item_id LEFT JOIN item_requirements req ON i.id = req.item_id LEFT JOIN item_enhancements enh ON i.id = enh.item_id LEFT JOIN item_ratings rt ON i.id = rt.item_id LEFT JOIN item_raw_data rd ON i.id = rd.item_id """ # Add spell join to count query if needed if spell_join_added: count_query += "\n LEFT JOIN item_spells sp ON i.id = sp.item_id" if conditions: count_query += "\nWHERE " + " AND ".join(conditions) count_result = await database.fetch_one(count_query, params) total_count = int(count_result[0]) if count_result else 0 # Format results with comprehensive translations (like individual inventory endpoint) items = [] for row in rows: item = dict(row) # Add computed properties item['is_equipped'] = item['current_wielded_location'] > 0 item['is_bonded'] = item['bonded'] > 0 item['is_attuned'] = item['attuned'] > 0 item['is_rare'] = (item['rare_id'] or 0) > 0 # Calculate condition percentage if item['max_structure'] and item['max_structure'] > 0: item['condition_percent'] = round((item['structure'] or 0) * 100.0 / item['max_structure'], 1) else: item['condition_percent'] = None # Apply comprehensive translations from original_json (like individual inventory endpoint) if item.get('original_json'): original_json = item['original_json'] # Handle case where original_json might be stored as string if isinstance(original_json, str): try: original_json = json.loads(original_json) except (json.JSONDecodeError, TypeError): original_json = {} if original_json: # Extract properties and get comprehensive translations properties = extract_item_properties(original_json) # Add material translation and prefixing if item.get('material') or properties.get('translations', {}).get('material_name'): material_name = None if item.get('material'): material_name = translate_material_type(item['material']) elif properties.get('translations', {}).get('material_name'): material_name = properties['translations']['material_name'] if material_name: item['material_name'] = material_name # Apply material prefix to item name original_name = item['name'] if not original_name.lower().startswith(material_name.lower()): item['name'] = f"{material_name} {original_name}" item['original_name'] = original_name # Add object class translation if item.get('object_class'): item['object_class_name'] = translate_object_class(item['object_class'], original_json) # Add spell information if 'spells' in properties: spell_info = properties['spells'] if spell_info.get('spells'): item['spells'] = spell_info['spells'] item['spell_names'] = [spell.get('name', '') for spell in spell_info['spells'] if spell.get('name')] if spell_info.get('active_spells'): item['active_spells'] = spell_info['active_spells'] # Add coverage calculation from coverage mask int_values = original_json.get('IntValues', {}) coverage_value = None # Check for coverage mask in correct location (218103821 = Coverage_Decal) if '218103821' in int_values: coverage_value = int_values['218103821'] elif 218103821 in int_values: coverage_value = int_values[218103821] if coverage_value and coverage_value > 0: coverage_parts = translate_coverage_mask(coverage_value) if coverage_parts: item['coverage'] = ', '.join(coverage_parts) else: item['coverage'] = f"Coverage_{coverage_value}" else: item['coverage'] = None # Add sophisticated equipment slot translation using Mag-SuitBuilder logic # Use both EquipableSlots_Decal and Coverage for armor reduction if original_json and 'IntValues' in original_json: equippable_slots = int_values.get('218103822', int_values.get(218103822, 0)) coverage_value = int_values.get('218103821', int_values.get(218103821, 0)) # Add debug info to help troubleshoot slot translation issues if 'legging' in item['name'].lower() or 'greave' in item['name'].lower(): item['debug_slot_info'] = { 'equippable_slots': equippable_slots, 'coverage_value': coverage_value, 'current_wielded_location': item.get('current_wielded_location', 0) } if equippable_slots and int(equippable_slots) > 0: # Check if item has material (can be tailored) has_material = bool(item.get('material_name') and item.get('material_name') != '') # Get sophisticated slot options using Mag-SuitBuilder logic slot_options = get_sophisticated_slot_options( int(equippable_slots), int(coverage_value) if coverage_value else 0, has_material ) # Translate all slot options to friendly names slot_names = [] for slot_option in slot_options: slot_name = translate_equipment_slot(slot_option) if slot_name and slot_name not in slot_names: slot_names.append(slot_name) item['slot_name'] = ', '.join(slot_names) if slot_names else f"Slot_{equippable_slots}" else: item['slot_name'] = "Unknown" else: item['slot_name'] = "Unknown" # Use gear totals as display ratings when individual ratings don't exist # For armor/clothing, ratings are often stored as gear totals (370, 372, 374) if item.get('damage_rating', -1) == -1 and 'gear_damage' in properties.get('ratings', {}): gear_damage = properties['ratings'].get('gear_damage', -1) if gear_damage > 0: item['damage_rating'] = gear_damage else: item['damage_rating'] = None elif item.get('damage_rating', -1) == -1: item['damage_rating'] = None if item.get('crit_damage_rating', -1) == -1 and 'gear_crit_damage' in properties.get('ratings', {}): gear_crit_damage = properties['ratings'].get('gear_crit_damage', -1) if gear_crit_damage > 0: item['crit_damage_rating'] = gear_crit_damage else: item['crit_damage_rating'] = None elif item.get('crit_damage_rating', -1) == -1: item['crit_damage_rating'] = None if item.get('heal_boost_rating', -1) == -1 and 'gear_healing_boost' in properties.get('ratings', {}): gear_healing_boost = properties['ratings'].get('gear_healing_boost', -1) if gear_healing_boost > 0: item['heal_boost_rating'] = gear_healing_boost else: item['heal_boost_rating'] = None elif item.get('heal_boost_rating', -1) == -1: item['heal_boost_rating'] = None # Add equipment set name translation if item.get('item_set') and str(item['item_set']).strip(): set_id = str(item['item_set']).strip() # Get dictionary from enum database dictionaries = ENUM_MAPPINGS.get('dictionaries', {}) attribute_set_info = dictionaries.get('AttributeSetInfo', {}).get('values', {}) if set_id in attribute_set_info: item['item_set_name'] = attribute_set_info[set_id] else: # Try checking if it's in the alternative location (equipment_sets) equipment_sets = ENUM_MAPPINGS.get('equipment_sets', {}) if set_id in equipment_sets: item['item_set_name'] = equipment_sets[set_id] else: item['item_set_name'] = f"Set {set_id}" # Clean up - remove raw data from response item.pop('original_json', None) item.pop('db_item_id', None) items.append(item) return { "items": items, "total_count": total_count, "page": page, "limit": limit, "total_pages": (total_count + limit - 1) // limit, "search_criteria": { "text": text, "character": character, "include_all_characters": include_all_characters, "equipment_status": equipment_status, "filters_applied": len(conditions) } } except Exception as e: logger.error(f"Search error: {e}", exc_info=True) raise HTTPException(status_code=500, detail=f"Search failed: {str(e)}") @app.get("/search/equipped/{character_name}") async def search_equipped_items( character_name: str, slot: int = Query(None, description="Specific equipment slot mask") ): """Get all equipped items for a character, optionally filtered by slot.""" try: conditions = ["i.character_name = :character_name", "i.current_wielded_location > 0"] params = {"character_name": character_name} if slot is not None: conditions.append("i.current_wielded_location = :slot") params["slot"] = slot query = """ SELECT i.*, COALESCE(cs.max_damage, -1) as max_damage, COALESCE(cs.armor_level, -1) as armor_level, COALESCE(cs.attack_bonus, -1.0) as attack_bonus, COALESCE(req.wield_level, -1) as wield_level, COALESCE(enh.material, '') as material, COALESCE(enh.workmanship, -1.0) as workmanship, COALESCE(enh.item_set, '') as item_set FROM items i LEFT JOIN item_combat_stats cs ON i.id = cs.item_id LEFT JOIN item_requirements req ON i.id = req.item_id LEFT JOIN item_enhancements enh ON i.id = enh.item_id WHERE """ + " AND ".join(conditions) + """ ORDER BY i.current_wielded_location, i.name """ rows = await database.fetch_all(query, params) # Load EquipMask enum for slot names equip_mask_map = {} if 'EquipMask' in ENUM_MAPPINGS.get('full_database', {}).get('enums', {}): equip_data = ENUM_MAPPINGS['full_database']['enums']['EquipMask']['values'] for k, v in equip_data.items(): try: equip_mask_map[int(k)] = v except (ValueError, TypeError): pass items = [] for row in rows: item = dict(row) item['slot_name'] = equip_mask_map.get(item['current_wielded_location'], f"Slot_{item['current_wielded_location']}") items.append(item) return { "character_name": character_name, "equipped_items": items, "slot_filter": slot, "item_count": len(items) } except Exception as e: logger.error(f"Equipped items search error: {e}", exc_info=True) raise HTTPException(status_code=500, detail=f"Equipped items search failed: {str(e)}") @app.get("/search/upgrades/{character_name}/{slot}") async def find_equipment_upgrades( character_name: str, slot: int, upgrade_type: str = Query("damage", description="What to optimize for: damage, armor, workmanship, value") ): """Find potential equipment upgrades for a specific slot.""" try: # Get currently equipped item in this slot current_query = """ SELECT i.*, cs.max_damage, cs.armor_level, enh.workmanship FROM items i LEFT JOIN item_combat_stats cs ON i.id = cs.item_id LEFT JOIN item_enhancements enh ON i.id = enh.item_id WHERE i.character_name = :character_name AND i.current_wielded_location = :slot """ current_item = await database.fetch_one(current_query, { "character_name": character_name, "slot": slot }) # Find all unequipped items that could be equipped in this slot # Check ValidLocations or infer from similar equipped items upgrade_query = """ SELECT DISTINCT i.character_name, i.name, i.value, i.burden, COALESCE(cs.max_damage, -1) as max_damage, COALESCE(cs.armor_level, -1) as armor_level, COALESCE(enh.workmanship, -1.0) as workmanship, COALESCE(req.wield_level, -1) as wield_level FROM items i LEFT JOIN item_combat_stats cs ON i.id = cs.item_id LEFT JOIN item_enhancements enh ON i.id = enh.item_id LEFT JOIN item_requirements req ON i.id = req.item_id WHERE i.current_wielded_location = 0 AND i.object_class = :object_class """ params = {} if current_item: params["object_class"] = current_item["object_class"] # Add upgrade criteria based on current item if upgrade_type == "damage" and current_item.get("max_damage", -1) > 0: upgrade_query += " AND cs.max_damage > :current_damage" params["current_damage"] = current_item["max_damage"] elif upgrade_type == "armor" and current_item.get("armor_level", -1) > 0: upgrade_query += " AND cs.armor_level > :current_armor" params["current_armor"] = current_item["armor_level"] elif upgrade_type == "workmanship" and current_item.get("workmanship", -1) > 0: upgrade_query += " AND enh.workmanship > :current_workmanship" params["current_workmanship"] = current_item["workmanship"] elif upgrade_type == "value": upgrade_query += " AND i.value > :current_value" params["current_value"] = current_item["value"] else: # No current item, show all available items for this slot type # We'll need to infer object class from slot - this is a simplified approach params["object_class"] = 1 # Default to generic # Add sorting based on upgrade type if upgrade_type == "damage": upgrade_query += " ORDER BY cs.max_damage DESC" elif upgrade_type == "armor": upgrade_query += " ORDER BY cs.armor_level DESC" elif upgrade_type == "workmanship": upgrade_query += " ORDER BY enh.workmanship DESC" else: upgrade_query += " ORDER BY i.value DESC" upgrade_query += " LIMIT 20" # Limit to top 20 upgrades upgrades = await database.fetch_all(upgrade_query, params) return { "character_name": character_name, "slot": slot, "upgrade_type": upgrade_type, "current_item": dict(current_item) if current_item else None, "potential_upgrades": [dict(row) for row in upgrades], "upgrade_count": len(upgrades) } except Exception as e: logger.error(f"Equipment upgrades search error: {e}", exc_info=True) raise HTTPException(status_code=500, detail=f"Equipment upgrades search failed: {str(e)}") @app.get("/characters/list") async def list_inventory_characters(): """List all characters that have inventory data.""" try: query = """ SELECT character_name, COUNT(*) as item_count, MAX(timestamp) as last_updated FROM items GROUP BY character_name ORDER BY character_name """ rows = await database.fetch_all(query) characters = [] for row in rows: characters.append({ "character_name": row["character_name"], "item_count": row["item_count"], "last_updated": row["last_updated"] }) return { "characters": characters, "total_characters": len(characters) } except Exception as e: logger.error(f"Failed to list inventory characters: {e}") raise HTTPException(status_code=500, detail="Failed to list inventory characters") if __name__ == "__main__": import uvicorn uvicorn.run(app, host="0.0.0.0", port=8000)