1112 lines
No EOL
45 KiB
Python
1112 lines
No EOL
45 KiB
Python
"""
|
|
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) |