""" 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.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" ) # 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()} logger.info(f"Enum database loaded successfully: {len(int_values)} int values, {len(spells)} spells, {len(object_classes)} object classes") 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, '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 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), }, '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), # Add missing combat stats 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)), }, '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': item_data.get('ItemSet'), }, 'ratings': { 'damage_rating': item_data.get('DamRating', -1), 'crit_rating': item_data.get('CritRating', -1), 'heal_boost_rating': item_data.get('HealBoostRating', -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 item_stmt = sa.insert(Item).values( character_name=inventory.character_name, item_id=item_id, timestamp=timestamp, name=properties['basic']['name'], icon=properties['basic']['icon'], object_class=properties['basic']['object_class'], value=properties['basic']['value'], burden=properties['basic']['burden'], has_id_data=properties['basic']['has_id_data'], last_id_time=item_data.get('LastIdTime', 0) ).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 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 comprehensive ratings if 'ratings' in translated_props: ratings = translated_props['ratings'] if ratings.get('damage_rating', -1) != -1: processed_item['damage_rating'] = ratings['damage_rating'] if ratings.get('crit_rating', -1) != -1: processed_item['crit_rating'] = ratings['crit_rating'] if ratings.get('heal_boost_rating', -1) != -1: processed_item['heal_boost_rating'] = ratings['heal_boost_rating'] # 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("/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 } if __name__ == "__main__": import uvicorn uvicorn.run(app, host="0.0.0.0", port=8000)