""" Quản Lý Vòng Đời Trang Bị Chưa Giám Định (Unidentified Item Lifecycle & Triage Manager) Giải quyết bài toán vật phẩm rơi ở dạng Chưa Giám Định (Unidentified): 1. Sàng lọc tiền giám định (Pre-Identification Triage): Đánh giá Phôi đồ (Base), iLvl, Rarity để quyết định nhặt hay bỏ. 2. Điều phối giám định 2 kênh: - Kênh 1: Giám định dã chiến bằng Cuộn Giám Định (Scroll of Wisdom) cho phôi S-Tier ngay trong map. - Kênh 2: Gom về làng/Hideout giám định hàng loạt miễn phí qua NPC The Hooded One. 3. Thẩm định hậu giám định (Post-Identification Appraisal) đưa vào bộ 'What Is Better'. 4. Tự học xác suất phẩm chất (Identification Probability & Expected Value Online Learning). """ import json import logging import os from dataclasses import dataclass from enum import Enum from typing import Any, Dict, List, Optional from src.agentic_engine.what_is_better import ( ItemComparisonResult, ItemStats, WhatIsBetterEvaluator, ) logger = logging.getLogger("ItemLifecycleManager") class ItemIdentificationState(str, Enum): UNIDENTIFIED = "UNIDENTIFIED" IDENTIFIED = "IDENTIFIED" class IdentificationChannel(str, Enum): FIELD_SCROLL_OF_WISDOM = "FIELD_SCROLL_OF_WISDOM" TOWN_THE_HOODED_ONE = "TOWN_THE_HOODED_ONE" SKIP_DO_NOT_IDENTIFY = "SKIP_DO_NOT_IDENTIFY" @dataclass class PreIdTriageResult: item_base_name: str item_level: int rarity: str slot: str recommended_channel: IdentificationChannel base_tier_score: float priority_reason: str expected_value_chaos: float @dataclass class UnidentifiedItemRecord: item_id: str base_name: str item_level: int rarity: str slot: str drop_area: str timestamp: float state: ItemIdentificationState = ItemIdentificationState.UNIDENTIFIED identified_stats: Optional[ItemStats] = None appraisal_result: Optional[ItemComparisonResult] = None class ItemLifecycleManager: """ Quản lý toàn diện quy trình xử lý đồ Chưa Giám Định (Unidentified Items) cho AutoPOE2. """ S_TIER_BASES = { "gloves": ["Spiked Fists", "Fists of Stone", "Stocky Mitts", "Facebreaker", "Iron Gauntlets", "Titan Gauntlets"], "weapon": ["Crescent Quarterstaff", "Dawn Staff", "Bone Quarterstaff", "Quarterstaff"], "ring": ["Amethyst Ring", "Diamond Ring", "Two-Stone Ring", "Prismatic Ring"], "amulet": ["Lunar Amulet", "Jade Amulet", "Amber Amulet"], "belt": ["Heavy Belt", "Utility Belt", "Vanguard Belt"], "boots": ["Windrunner Boots", "Dragonscale Boots", "Rawhide Boots", "Two-Toned Boots"], "body_armour": ["Dragonscale Doublet", "Varnished Coat", "Astral Plate", "Sadist Garb", "Vaal Regalia"] } def __init__( self, evaluator: Optional[WhatIsBetterEvaluator] = None, learning_db_path: str = "captures/unidentified_learning_db.json" ): self.evaluator = evaluator or WhatIsBetterEvaluator() self.learning_db_path = learning_db_path self.inventory_items: Dict[str, UnidentifiedItemRecord] = {} self.base_knowledge: Dict[str, Dict[str, Any]] = {} self._load_knowledge() def triage_unidentified_item( self, base_name: str, item_level: int, rarity: str, slot: str, scrolls_available: int = 40, in_combat_zone: bool = True ) -> PreIdTriageResult: slot_lower = slot.lower() s_bases = self.S_TIER_BASES.get(slot_lower, []) is_s_tier_base = any(b.lower() in base_name.lower() for b in s_bases) is_high_ilvl = item_level >= 75 is_unique = rarity.upper() == "UNIQUE" learned_stats = self.base_knowledge.get(base_name, {"id_count": 0, "upgrade_count": 0, "avg_chaos": 5.0}) historical_upgrade_rate = ( (learned_stats["upgrade_count"] / learned_stats["id_count"]) if learned_stats["id_count"] >= 5 else 0.15 ) base_score = 40.0 if is_s_tier_base: base_score += 35.0 if is_high_ilvl: base_score += 15.0 if is_unique: base_score += 20.0 base_score += min(10.0, historical_upgrade_rate * 40.0) if is_unique: if scrolls_available > 5 and in_combat_zone: channel = IdentificationChannel.FIELD_SCROLL_OF_WISDOM reason = "Đồ Unique giá trị cao: Giám định ngay bằng Scroll of Wisdom để kiểm tra roll." else: channel = IdentificationChannel.TOWN_THE_HOODED_ONE reason = "Đồ Unique: Lưu túi đồ để The Hooded One giám định miễn phí tại làng." elif is_s_tier_base and is_high_ilvl and slot_lower == "gloves": if scrolls_available >= 10 and in_combat_zone: channel = IdentificationChannel.FIELD_SCROLL_OF_WISDOM reason = f"Phôi S-Tier tối thượng ({base_name} iLvl {item_level}): Giám định ngay bằng Scroll, nếu xịn mặc luôn!" else: channel = IdentificationChannel.TOWN_THE_HOODED_ONE reason = f"Phôi S-Tier ({base_name}): Đưa về làng để The Hooded One giám định miễn phí." elif base_score >= 50.0 or rarity.upper() == "RARE": channel = IdentificationChannel.TOWN_THE_HOODED_ONE reason = f"Phôi hiếm thông thường ({base_name}): Gom vào túi đồ, về làng để The Hooded One giám định 1-click tiết kiệm Scroll." else: channel = IdentificationChannel.SKIP_DO_NOT_IDENTIFY reason = f"Phôi rác iLvl thấp ({base_name}): Bỏ qua hoặc đưa thẳng vào Bàn Thợ Rèn (SALVAGE_BENCH) lấy Vàng." expected_chaos = round(learned_stats.get("avg_chaos", 5.0) * (base_score / 50.0), 1) return PreIdTriageResult( item_base_name=base_name, item_level=item_level, rarity=rarity, slot=slot, recommended_channel=channel, base_tier_score=round(base_score, 1), priority_reason=reason, expected_value_chaos=expected_chaos ) def check_the_hooded_one_interaction( self, current_area_name: str, unidentified_items_in_inventory: int ) -> Dict[str, Any]: is_safe_hub = any( h in current_area_name.lower() for h in ["hideout", "town", "encampment", "settlement", "sanctum", "shores"] ) if not is_safe_hub: return { "can_interact": False, "message": f"Đang ở khu vực chiến đấu ({current_area_name}). Hãy dùng Scroll of Wisdom dã chiến hoặc mở portal về làng." } if unidentified_items_in_inventory <= 0: return { "can_interact": False, "message": "Không có món đồ chưa giám định nào trong túi đồ." } return { "can_interact": True, "npc_name": "The Hooded One", "items_count": unidentified_items_in_inventory, "headline": f"PHÁT HIỆN {unidentified_items_in_inventory} MÓN ĐỒ CHƯA GIÁM ĐỊNH", "action_guidance": ( f"Tương tác với The Hooded One tại {current_area_name} " f"và chọn 'Identify All' (hoặc dùng phím tắt macro) để giải mã toàn bộ {unidentified_items_in_inventory} món đồ với chi phí 0 Vàng / 0 Scroll." ) } def process_identified_item( self, item_id: str, revealed_stats: ItemStats, current_equipped: ItemStats, player_resists: Optional[Dict[str, int]] = None ) -> ItemComparisonResult: result = self.evaluator.compare_items( current_item=current_equipped, candidate_item=revealed_stats, player_current_resists=player_resists ) base_name = revealed_stats.name.split(" of ")[0] self._update_learning_database( base_name=base_name, is_upgrade=(result.decision == "EQUIP_UPGRADE"), trade_value_chaos=revealed_stats.estimated_trade_value_chaos or (50.0 if result.decision == "EQUIP_UPGRADE" else 5.0) ) logger.info( f"[ItemLifecycle] Giám định xong '{revealed_stats.name}': " f"Kết quả = {result.decision} ({result.headline_verdict})" ) return result def evaluate_field_identified_item( self, item_stats: ItemStats, current_equipped: Optional[ItemStats] = None, min_chaos_value: float = 3.0 ) -> Dict[str, Any]: """ Thẩm định dã chiến đồ vừa giám định bằng Scroll of Wisdom ngay trong map (Doc 63 SSoT). Nếu là đồ rác (không phải upgrade và giá trị chaos < min_chaos_value) -> Đề xuất vứt ra sàn (DISCARD) để giải phóng ô túi đồ cho loot xịn hơn, tránh chật túi. """ if current_equipped: comp = self.evaluator.compare_items(current_equipped, item_stats) is_upgrade = (comp.decision == "EQUIP_UPGRADE") else: is_upgrade = False est_value = item_stats.estimated_trade_value_chaos or 1.0 should_discard = (not is_upgrade) and (est_value < min_chaos_value) return { "item_name": item_stats.name, "is_upgrade": is_upgrade, "estimated_chaos": est_value, "should_discard": should_discard, "action": "FIELD_DISCARD" if should_discard else "KEEP_IN_INVENTORY", "reason": ( f"Đồ rác ({item_stats.name}): Giá trị ước tính {est_value}c < {min_chaos_value}c, " "vứt ngay ra đất dã chiến để tiết kiệm ô túi đồ." if should_discard else f"Đồ có giá trị ({item_stats.name}, {est_value}c): Giữ lại trong túi để cất Stash." ) } def _update_learning_database(self, base_name: str, is_upgrade: bool, trade_value_chaos: float) -> None: if base_name not in self.base_knowledge: self.base_knowledge[base_name] = { "id_count": 0, "upgrade_count": 0, "total_chaos_value": 0.0, "avg_chaos": 5.0 } k = self.base_knowledge[base_name] k["id_count"] += 1 if is_upgrade: k["upgrade_count"] += 1 k["total_chaos_value"] += trade_value_chaos k["avg_chaos"] = round(k["total_chaos_value"] / k["id_count"], 1) self._save_knowledge() def get_knowledge_summary(self) -> List[Dict[str, Any]]: summary = [] for base, stats in self.base_knowledge.items(): summary.append({ "base_name": base, "identified_count": stats.get("id_count", 0), "upgrades_found": stats.get("upgrade_count", 0), "upgrade_rate_pct": round((stats.get("upgrade_count", 0) / max(1, stats.get("id_count", 1))) * 100.0, 1), "avg_value_chaos": stats.get("avg_chaos", 0.0) }) summary.sort(key=lambda x: (x["upgrades_found"], x["avg_value_chaos"]), reverse=True) return summary def _save_knowledge(self) -> None: os.makedirs(os.path.dirname(self.learning_db_path) or ".", exist_ok=True) try: with open(self.learning_db_path, "w", encoding="utf-8") as f: json.dump(self.base_knowledge, f, indent=2, ensure_ascii=False) except Exception as e: logger.error(f"[ItemLifecycle] Lỗi lưu tri thức: {e}") def _load_knowledge(self) -> None: if os.path.exists(self.learning_db_path): try: with open(self.learning_db_path, "r", encoding="utf-8") as f: self.base_knowledge = json.load(f) except Exception as e: logger.error(f"[ItemLifecycle] Lỗi đọc tri thức: {e}")