# Description: LogWatcher implementing Observer Pattern for real-time log parsing # Updated: Swedish language support based on real game logs # Monitors chat.log asynchronously with minimal CPU impact (Rule #3: 60+ FPS) import asyncio import re import os from pathlib import Path from typing import Callable, List, Dict, Any, Optional from dataclasses import dataclass, field from datetime import datetime from decimal import Decimal import logging import time logger = logging.getLogger(__name__) @dataclass class LogEvent: """Represents a parsed event from chat.log.""" timestamp: datetime event_type: str raw_line: str data: Dict[str, Any] @dataclass class LootItem: """Represents a single loot item.""" name: str quantity: int value_ped: Decimal is_shrapnel: bool = False is_universal_ammo: bool = False @dataclass class HuntingSessionStats: """Statistics for a hunting session tracked from chat.log.""" # Financial tracking total_loot_ped: Decimal = Decimal('0.0') total_shrapnel_ped: Decimal = Decimal('0.0') total_universal_ammo_ped: Decimal = Decimal('0.0') total_other_loot_ped: Decimal = Decimal('0.0') # Non-shrapnel, non-UA loot # Cost tracking weapon_cost_ped: Decimal = Decimal('0.0') armor_cost_ped: Decimal = Decimal('0.0') healing_cost_ped: Decimal = Decimal('0.0') plates_cost_ped: Decimal = Decimal('0.0') total_cost_ped: Decimal = Decimal('0.0') # Combat tracking damage_dealt: Decimal = Decimal('0.0') damage_taken: Decimal = Decimal('0.0') healing_done: Decimal = Decimal('0.0') shots_fired: int = 0 kills: int = 0 # Special events globals_count: int = 0 hofs_count: int = 0 personal_globals: List[Dict[str, Any]] = field(default_factory=list) # Calculated metrics @property def net_profit_ped(self) -> Decimal: """Calculate net profit (excluding shrapnel from loot).""" return self.total_other_loot_ped - self.total_cost_ped @property def return_percentage(self) -> Decimal: """Calculate return percentage (loot/cost * 100).""" if self.total_cost_ped > 0: return (self.total_other_loot_ped / self.total_cost_ped) * Decimal('100') return Decimal('0.0') @property def cost_per_kill(self) -> Decimal: """Calculate cost per kill.""" if self.kills > 0: return self.total_cost_ped / self.kills return Decimal('0.0') @property def dpp(self) -> Decimal: """Calculate Damage Per PED (efficiency metric).""" if self.total_cost_ped > 0: return self.damage_dealt / self.total_cost_ped return Decimal('0.0') @property def damage_per_kill(self) -> Decimal: """Calculate average damage per kill.""" if self.kills > 0: return self.damage_dealt / self.kills return Decimal('0.0') class LogWatcher: """ Watches Entropia Universe chat.log and notifies observers of events. Supports multiple languages (English, Swedish) based on real game logs. Implements Observer Pattern: Multiple modules can subscribe to specific event types without tight coupling. """ # ======================================================================== # REGEX PATTERNS - ENGLISH & SWEDISH (from real game logs) # ======================================================================== # LOOT PATTERNS # English: "You received [Shrapnel] x (67) Value: 0.0067 PED" # English alt: "You received Shrapnel x 123 (Value: 1.23 PED)" # Swedish: "Du fick Shrapnel x (4627) Värde: 0.4627 PED" PATTERN_LOOT_EN = re.compile( r'^(\d{4}-\d{2}-\d{2}\s+\d{2}:\d{2}:\d{2})\s+\[System\]:?\s*\[?\]?\s*' r'You\s+received\s+\[?([\w\s\-()]+?)\]?\s+x\s*\((\d+)\)\s*' r'Value:\s+(\d+(?:\.\d+)?)\s+PED', re.IGNORECASE ) PATTERN_LOOT_SV = re.compile( r'^(\d{4}-\d{2}-\d{2}\s+\d{2}:\d{2}:\d{2})\s+\[System\]\s+\[?\]?\s*' r'Du\s+fick\s+([\w\s\-()]+?)\s+x\s*\((\d+)\)\s*' r'Värde:\s+(\d+(?:\.\d+)?)\s+PED', re.IGNORECASE ) # LOOT PATTERN WITHOUT VALUE (some items don't show value) # English: "You received Animal Thyroid Oil x 5" PATTERN_LOOT_NO_VALUE_EN = re.compile( r'^(\d{4}-\d{2}-\d{2}\s+\d{2}:\d{2}:\d{2})\s+\[System\]:?\s*\[?\]?\s*' r'You\s+received\s+\[?([\w\s\-()]+?)\]?\s+x\s*(\d+)', re.IGNORECASE ) PATTERN_LOOT_NO_VALUE_SV = re.compile( r'^(\d{4}-\d{2}-\d{2}\s+\d{2}:\d{2}:\d{2})\s+\[System\]\s+\[?\]?\s*' r'Du\s+fick\s+([\w\s\-()]+?)\s+x\s*(\d+)', re.IGNORECASE ) # KILL PATTERNS - "You killed" messages (for accurate kill counting) # English: "You killed [Creature Name]" # Swedish: "Du dödade [Creature Name]" PATTERN_KILL_EN = re.compile( r'^(\d{4}-\d{2}-\d{2}\s+\d{2}:\d{2}:\d{2})\s+\[System\]:?\s*\[?\]?\s*' r'You\s+killed\s+\[?([\w\s\-()]+?)\]?', re.IGNORECASE ) PATTERN_KILL_SV = re.compile( r'^(\d{4}-\d{2}-\d{2}\s+\d{2}:\d{2}:\d{2})\s+\[System\]\s+\[?\]?\s*' r'Du\s+dödade\s+\[?([\w\s\-()]+?)\]?', re.IGNORECASE ) # GLOBAL PATTERNS (Other players) # English: "PlayerName globals in Zone for 150.00 PED" # Swedish: "PlayerName hittade en avsättning (Item) med ett värde av X PED" PATTERN_GLOBAL_EN = re.compile( r'^(\d{4}-\d{2}-\d{2}\s+\d{2}:\d{2}:\d{2})\s+\[Globala?\]\s+\[?\]?\s*' r'([\w\s]+?)\s+(?:globals?|found)\s+(?:in|a)\s+([\w\s]+?)\s+' r'(?:for|with\s+a\s+value\s+of)\s+(\d+(?:\.\d+)?)\s+PED', re.IGNORECASE ) PATTERN_GLOBAL_SV = re.compile( r'^(\d{4}-\d{2}-\d{2}\s+\d{2}:\d{2}:\d{2})\s+\[Globala?\]\s+\[?\]?\s*' r'([\w\s]+?)\s+hittade\s+en\s+avsättning\s+\(([^)]+)\)\s+' r'med\s+ett\s+värde\s+av\s+(\d+(?:\.\d+)?)\s+PED', re.IGNORECASE ) # PERSONAL GLOBAL (when YOU get a global - different from seeing others) # English: "[Globals] [Player] killed a creature (Creature) with a value of X PED" # Swedish: "[Globala] [Player] dödade ett kreatur (Creature) med ett värde av X PED" PATTERN_PERSONAL_GLOBAL_EN = re.compile( r'^(\d{4}-\d{2}-\d{2}\s+\d{2}:\d{2}:\d{2})\s+\[Globals\]\s+\[?\]?\s*' r'([\w\s]+?)\s+killed\s+a\s+creature\s+\(([^)]+)\)\s+' r'(?:with\s+a\s+value\s+of|for)\s+(\d+(?:\.\d+)?)\s+PED', re.IGNORECASE ) PATTERN_PERSONAL_GLOBAL_SV = re.compile( r'^(\d{4}-\d{2}-\d{2}\s+\d{2}:\d{2}:\d{2})\s+\[Globala\]\s+\[?\]?\s*' r'([\w\s]+?)\s+dödade\s+ett\s+kreatur\s+\(([^)]+)\)\s+' r'med\s+ett\s+värde\s+av\s+(\d+(?:\.\d+)?)\s+PED', re.IGNORECASE ) # HALL OF FAME PATTERNS # Swedish: "...En post har lagts till i Hall of Fame!" # English: "[Hall of Fame] Player killed a creature (Creature) for X PED" PATTERN_HOF_EN = re.compile( r'^(\d{4}-\d{2}-\d{2}\s+\d{2}:\d{2}:\d{2})\s+\[Hall\s+of\s+Fame\]\s+\[?\]?\s*' r'([\w\s]+?)\s+killed\s+a\s+creature\s+\(([^)]+)\)\s+' r'(?:for|with\s+a\s+value\s+of)\s+(\d+(?:\.\d+)?)\s+PED', re.IGNORECASE ) PATTERN_HOF_MARKER = re.compile( r'^(\d{4}-\d{2}-\d{2}\s+\d{2}:\d{2}:\d{2})\s+\[\w+\]\s+\[?\]?\s*' r'.*?Hall\s+of\s+Fame', re.IGNORECASE ) # SKILL GAIN PATTERNS # English: "You have gained 1.1466 experience in your Whip skill" # English (alt): "You gained 0.45 experience in your Rifle skill" # Swedish: "Du har fått 0.3238 erfarenhet i din Translocation färdighet" PATTERN_SKILL_EN = re.compile( r'^(\d{4}-\d{2}-\d{2}\s+\d{2}:\d{2}:\d{2})\s+\[System\]\s+\[?\]?\s*' r'You\s+(?:have\s+)?gained\s+(\d+(?:\.\d+)?)\s+experience\s+in\s+your\s+([\w\s]+?)\s+skill', re.IGNORECASE ) PATTERN_SKILL_SV = re.compile( r'^(\d{4}-\d{2}-\d{2}\s+\d{2}:\d{2}:\d{2})\s+\[System\]\s+\[?\]?\s*' r'Du\s+har\s+fått\s+(\d+(?:\.\d+)?)\s+erfarenhet\s+i\s+din\s+([\w\s]+?)\s+färdighet', re.IGNORECASE ) # SKILL LEVEL UP # English: "You have advanced to level 45 in Rifle" # Swedish: "Du har avancerat till nivå 45 i Rifle" PATTERN_LEVEL_UP_EN = re.compile( r'^(\d{4}-\d{2}-\d{2}\s+\d{2}:\d{2}:\d{2})\s+\[System\]\s+\[?\]?\s*' r'You\s+have\s+advanced\s+to\s+level\s+(\d+)\s+in\s+([\w\s]+)', re.IGNORECASE ) PATTERN_LEVEL_UP_SV = re.compile( r'^(\d{4}-\d{2}-\d{2}\s+\d{2}:\d{2}:\d{2})\s+\[System\]\s+\[?\]?\s*' r'Du\s+har\s+avancerat\s+till\s+nivå\s+(\d+)\s+i\s+([\w\s]+)', re.IGNORECASE ) # DAMAGE DEALT - Swedish & English # Swedish: "Du orsakade 13.5 poäng skada" # English: "You inflicted 4.4 points of damage" PATTERN_DAMAGE_DEALT_SV = re.compile( r'^(\d{4}-\d{2}-\d{2}\s+\d{2}:\d{2}:\d{2})\s+\[System\]\s+\[?\]?\s*' r'Du\s+orsakade\s+(\d+(?:\.\d+)?)\s+poäng\s+skada', re.IGNORECASE ) PATTERN_DAMAGE_DEALT_EN = re.compile( r'^(\d{4}-\d{2}-\d{2}\s+\d{2}:\d{2}:\d{2})\s+\[System\]\s+\[?\]?\s*' r'You\s+inflicted\s+(\d+(?:\.\d+)?)\s+points?\s+of\s+damage', re.IGNORECASE ) # CRITICAL HIT # Swedish: "Kritisk träff - Extra skada! Du orsakade 44.4 poäng skada" # English: "Critical hit - Additional damage! You inflicted 49.6 points of damage" PATTERN_CRITICAL_SV = re.compile( r'^(\d{4}-\d{2}-\d{2}\s+\d{2}:\d{2}:\d{2})\s+\[System\]\s+\[?\]?\s*' r'Kritisk\s+träff.*?Du\s+orsakade\s+(\d+(?:\.\d+)?)\s+poäng\s+skada', re.IGNORECASE ) PATTERN_CRITICAL_EN = re.compile( r'^(\d{4}-\d{2}-\d{2}\s+\d{2}:\d{2}:\d{2})\s+\[System\]\s+\[?\]?\s*' r'Critical\s+hit.*?You\s+inflicted\s+(\d+(?:\.\d+)?)\s+points?\s+of\s+damage', re.IGNORECASE ) # DAMAGE TAKEN # Swedish: "Du tog 31.5 poäng skada" # English: "You took 7.4 points of damage" PATTERN_DAMAGE_TAKEN_SV = re.compile( r'^(\d{4}-\d{2}-\d{2}\s+\d{2}:\d{2}:\d{2})\s+\[System\]\s+\[?\]?\s*' r'Du\s+tog\s+(\d+(?:\.\d+)?)\s+poäng\s+skada', re.IGNORECASE ) PATTERN_DAMAGE_TAKEN_EN = re.compile( r'^(\d{4}-\d{2}-\d{2}\s+\d{2}:\d{2}:\d{2})\s+\[System\]\s+\[?\]?\s*' r'You\s+took\s+(\d+(?:\.\d+)?)\s+points?\s+of\s+damage', re.IGNORECASE ) # HEALING # Swedish: "Du läkte dig själv 4.0 poäng" # English: "You healed yourself 25.5 points" PATTERN_HEAL_SV = re.compile( r'^(\d{4}-\d{2}-\d{2}\s+\d{2}:\d{2}:\d{2})\s+\[System\]\s+\[?\]?\s*' r'Du\s+läkte\s+dig\själv\s+(\d+(?:\.\d+)?)\s+poäng', re.IGNORECASE ) PATTERN_HEAL_EN = re.compile( r'^(\d{4}-\d{2}-\d{2}\s+\d{2}:\d{2}:\d{2})\s+\[System\]\s+\[?\]?\s*' r'You\s+healed\s+yourself\s+(\d+(?:\.\d+)?)\s+points?', re.IGNORECASE ) # WEAPON TIER/LEVEL UP # Swedish: "Din Piron PBP-17 (L) har nått nivå 0.38" # English: "Your Piron PBP-17 (L) has reached tier 2.68" PATTERN_WEAPON_TIER_SV = re.compile( r'^(\d{4}-\d{2}-\d{2}\s+\d{2}:\d{2}:\d{2})\s+\[System\]\s+\[?\]?\s*' r'Din\s+([\w\s\-()]+?)\s+har\snått\s+nivå\s+(\d+(?:\.\d+)?)', re.IGNORECASE ) PATTERN_WEAPON_TIER_EN = re.compile( r'^(\d{4}-\d{2}-\d{2}\s+\d{2}:\d{2}:\d{2})\s+\[System\]\s+\[?\]?\s*' r'Your\s+([\w\s\-()]+?)\s+has\s+reached\s+tier\s+(\d+(?:\.\d+)?)', re.IGNORECASE ) # COMBAT EVADE/DODGE/MISS # English: "You Evaded", "The target Evaded your attack", "The attack missed you" # Swedish: "Du undvek", "Målet undvek din attack" PATTERN_EVADE_EN = re.compile( r'^(\d{4}-\d{2}-\d{2}\s+\d{2}:\d{2}:\d{2})\s+\[System\]\s+\[?\]?\s*' r'(You\s+Evaded|You\s+dodged|The\s+target\s+Evaded\s+your\s+attack|The\s+target\s+Dodged|The\s+attack\s+missed\s+you)', re.IGNORECASE ) PATTERN_EVADE_SV = re.compile( r'^(\d{4}-\d{2}-\d{2}\s+\d{2}:\d{2}:\d{2})\s+\[System\]\s+\[?\]?\s*' r'(Du\s+undvek|Målet\s+undvek\s+din\s+attack|Attacken\s+missade\s+dig)', re.IGNORECASE ) # DECAY (when weapon durability decreases) # English: "Your Omegaton M2100 has decayed 15 PEC" # Swedish: "Din Piron PBP-17 (L) har nått minimalt skick" PATTERN_DECAY_EN = re.compile( r'^(\d{4}-\d{2}-\d{2}\s+\d{2}:\d{2}:\d{2})\s+\[System\]\s+\[?\]?\s*' r'Your\s+([\w\s\-()]+?)\s+has\s+decayed\s+(\d+(?:\.\d+)?)\s+PEC', re.IGNORECASE ) PATTERN_DECAY_SV = re.compile( r'^(\d{4}-\d{2}-\d{2}\s+\d{2}:\d{2}:\d{2})\s+\[System\]\s+\[?\]?\s*' r'Din\s+([\w\s\-()]+?)\s+har\s+decayed\s+(\d+(?:\.\d+)?)\s+PEC', re.IGNORECASE ) PATTERN_WEAPON_BROKEN_SV = re.compile( r'^(\d{4}-\d{2}-\d{2}\s+\d{2}:\d{2}:\d{2})\s+\[System\]\s+\[?\]?\s*' r'Din\s+([\w\s\-()]+?)\s+har\s+nått\s+minimalt\s+skick', re.IGNORECASE ) # BROKEN ENHANCERS # English: "Your enhancer Weapon Damage Enhancer 1 on your Piron PBP-17 (L) broke" PATTERN_ENHANCER_BROKEN_EN = re.compile( r'^(\d{4}-\d{2}-\d{2}\s+\d{2}:\d{2}:\d{2})\s+\[System\]\s+\[?\]?\s*' r'Your\s+enhancer\s+([\w\s]+?)\s+on\s+your\s+([\w\s\-()]+?)\s+broke', re.IGNORECASE ) # PED TRANSFER # Swedish: "Överföring slutförd! 3.38000 PED har överförts till ditt PED-kort." PATTERN_PED_TRANSFER_SV = re.compile( r'^(\d{4}-\d{2}-\d{2}\s+\d{2}:\d{2}:\d{2})\s+\[System\]\s+' r'Överföring\s+slutförd.*?((\d+(?:\.\d+)?))\s+PED', re.IGNORECASE ) PATTERN_PED_TRANSFER_EN = re.compile( r'^(\d{4}-\d{2}-\d{2}\s+\d{2}:\d{2}:\d{2})\s+\[System\]\s+' r'Transfer\s+complete.*?((\d+(?:\.\d+)?))\s+PED', re.IGNORECASE ) # ATTRIBUTE GAIN (Agility, etc) # Swedish: "Din Agility har förbättrats med 0.0001" # English: "Your Agility has improved by 0.0001" OR "You gained 0.0001 Agility" PATTERN_ATTRIBUTE_SV = re.compile( r'^(\d{4}-\d{2}-\d{2}\s+\d{2}:\d{2}:\d{2})\s+\[System\]\s+\[?\]?\s*' r'Din\s+(\w+)\s+har\s+förbättrats\s+med\s+(\d+(?:\.\d+)?)', re.IGNORECASE ) PATTERN_ATTRIBUTE_EN = re.compile( r'^(\d{4}-\d{2}-\d{2}\s+\d{2}:\d{2}:\d{2})\s+\[System\]\s+\[?\]?\s*' r'(?:Your\s+(\w+)\s+has\s+improved\s+by|You\s+gained)\s+(\d+(?:\.\d+)?)\s+(\w+)', re.IGNORECASE ) # TEAM HUNT PATTERNS # "You received 0.1234 PED from your teammates' activity." PATTERN_TEAM_SHARE_EN = re.compile( r'^(\d{4}-\d{2}-\d{2}\s+\d{2}:\d{2}:\d{2})\s+\[System\]\s+\[?\]?\s*' r'You\s+received\s+(\d+(?:\.\d+)?)\s+PED\s+from\s+your\s+teammates', re.IGNORECASE ) PATTERN_TEAM_SHARE_SV = re.compile( r'^(\d{4}-\d{2}-\d{2}\s+\d{2}:\d{2}:\d{2})\s+\[System\]\s+\[?\]?\s*' r'Du\s+fick\s+(\d+(?:\.\d+)?)\s+PED\s+från\s+dina\s+lagkamrater', re.IGNORECASE ) EVENT_PATTERNS = { 'loot_en': PATTERN_LOOT_EN, 'loot_sv': PATTERN_LOOT_SV, 'loot_no_value_en': PATTERN_LOOT_NO_VALUE_EN, 'loot_no_value_sv': PATTERN_LOOT_NO_VALUE_SV, 'kill_en': PATTERN_KILL_EN, 'kill_sv': PATTERN_KILL_SV, 'global_en': PATTERN_GLOBAL_EN, 'global_sv': PATTERN_GLOBAL_SV, 'personal_global_en': PATTERN_PERSONAL_GLOBAL_EN, 'personal_global_sv': PATTERN_PERSONAL_GLOBAL_SV, 'hof_en': PATTERN_HOF_EN, 'hof_marker': PATTERN_HOF_MARKER, 'skill_en': PATTERN_SKILL_EN, 'skill_sv': PATTERN_SKILL_SV, 'damage_dealt_sv': PATTERN_DAMAGE_DEALT_SV, 'damage_dealt_en': PATTERN_DAMAGE_DEALT_EN, 'critical_hit_sv': PATTERN_CRITICAL_SV, 'critical_hit_en': PATTERN_CRITICAL_EN, 'damage_taken_sv': PATTERN_DAMAGE_TAKEN_SV, 'damage_taken_en': PATTERN_DAMAGE_TAKEN_EN, 'heal_sv': PATTERN_HEAL_SV, 'heal_en': PATTERN_HEAL_EN, 'weapon_tier_sv': PATTERN_WEAPON_TIER_SV, 'weapon_tier_en': PATTERN_WEAPON_TIER_EN, 'evade_en': PATTERN_EVADE_EN, 'evade_sv': PATTERN_EVADE_SV, 'decay_en': PATTERN_DECAY_EN, 'decay_sv': PATTERN_DECAY_SV, 'weapon_broken_sv': PATTERN_WEAPON_BROKEN_SV, 'enhancer_broken_en': PATTERN_ENHANCER_BROKEN_EN, 'ped_transfer_sv': PATTERN_PED_TRANSFER_SV, 'ped_transfer_en': PATTERN_PED_TRANSFER_EN, 'attribute_sv': PATTERN_ATTRIBUTE_SV, 'attribute_en': PATTERN_ATTRIBUTE_EN, 'team_share_en': PATTERN_TEAM_SHARE_EN, 'team_share_sv': PATTERN_TEAM_SHARE_SV, } def __init__(self, log_path: Optional[str] = None, poll_interval: float = 1.0, mock_mode: bool = False): """Initialize LogWatcher.""" self.mock_mode = mock_mode if log_path is None: if mock_mode: core_dir = Path(__file__).parent log_path = core_dir.parent / "test-data" / "mock-chat.log" else: log_path = self._find_eu_log_path() self.log_path = Path(log_path) self.poll_interval = poll_interval self.observers: Dict[str, List[Callable]] = { 'loot': [], 'global': [], 'hof': [], 'skill': [], 'damage_dealt': [], 'damage_taken': [], 'heal': [], 'weapon_tier': [], 'evade': [], 'decay': [], 'critical_hit': [], 'ped_transfer': [], 'attribute': [], 'kill': [], 'team_share': [], 'any': [], } self._running = False self._file_position = 0 self._last_file_size = 0 self._task: Optional[asyncio.Task] = None logger.info(f"LogWatcher initialized: {self.log_path} (mock={mock_mode})") def _find_eu_log_path(self) -> Path: """Attempt to find Entropia Universe chat.log.""" possible_paths = [ Path.home() / "Documents" / "Entropia Universe" / "chat.log", Path("C:") / "Users" / os.getenv("USERNAME", "User") / "Documents" / "Entropia Universe" / "chat.log", ] wine_prefix = Path.home() / ".wine" / "drive_c" possible_paths.extend([ wine_prefix / "users" / os.getenv("USER", "user") / "Documents" / "Entropia Universe" / "chat.log", ]) for path in possible_paths: if path.exists(): logger.info(f"Found EU log: {path}") return path fallback = Path(__file__).parent.parent / "test-data" / "chat.log" logger.warning(f"EU log not found, using fallback: {fallback}") return fallback def subscribe(self, event_type: str, callback: Callable[[LogEvent], None]) -> None: """Subscribe to an event type.""" if event_type not in self.observers: self.observers[event_type] = [] self.observers[event_type].append(callback) logger.debug(f"Subscribed to {event_type}: {callback.__name__}") def unsubscribe(self, event_type: str, callback: Callable[[LogEvent], None]) -> None: """Unsubscribe from an event type.""" if event_type in self.observers: if callback in self.observers[event_type]: self.observers[event_type].remove(callback) logger.debug(f"Unsubscribed from {event_type}: {callback.__name__}") def _notify(self, event: LogEvent) -> None: """Notify all observers of an event.""" if event.event_type in self.observers: for callback in self.observers[event.event_type]: try: callback(event) except Exception as e: logger.error(f"Observer error for {event.event_type}: {e}") for callback in self.observers['any']: try: callback(event) except Exception as e: logger.error(f"Observer error for 'any': {e}") def _parse_timestamp(self, ts_str: str) -> datetime: """Parse EU timestamp format.""" return datetime.strptime(ts_str, "%Y-%m-%d %H:%M:%S") def _is_shrapnel(self, item_name: str) -> bool: """Check if item is Shrapnel.""" return item_name.strip().lower() == 'shrapnel' def _is_universal_ammo(self, item_name: str) -> bool: """Check if item is Universal Ammo.""" name = item_name.strip().lower() return name == 'universal ammo' or name == 'universell ammunition' def _categorize_loot(self, item_name: str, value_ped: Decimal) -> LootItem: """Categorize loot item and return LootItem.""" is_shrapnel = self._is_shrapnel(item_name) is_ua = self._is_universal_ammo(item_name) return LootItem( name=item_name.strip(), quantity=1, # Will be set by caller value_ped=value_ped, is_shrapnel=is_shrapnel, is_universal_ammo=is_ua ) def _parse_line(self, line: str) -> Optional[LogEvent]: """ Parse a single log line. Returns LogEvent if parsed, None otherwise. """ line = line.strip() if not line: return None # Try each pattern in priority order # KILL - Swedish match = self.PATTERN_KILL_SV.match(line) if match: return self._create_kill_event(match, line, 'swedish') # KILL - English match = self.PATTERN_KILL_EN.match(line) if match: return self._create_kill_event(match, line, 'english') # LOOT - Swedish (prioritize based on your game client) match = self.PATTERN_LOOT_SV.match(line) if match: return self._create_loot_event_sv(match, line) # LOOT - English match = self.PATTERN_LOOT_EN.match(line) if match: return self._create_loot_event_en(match, line) # LOOT WITHOUT VALUE - Swedish match = self.PATTERN_LOOT_NO_VALUE_SV.match(line) if match: return self._create_loot_event_no_value(match, line, 'swedish') # LOOT WITHOUT VALUE - English match = self.PATTERN_LOOT_NO_VALUE_EN.match(line) if match: return self._create_loot_event_no_value(match, line, 'english') # GLOBAL - Swedish match = self.PATTERN_GLOBAL_SV.match(line) if match: return self._create_global_event_sv(match, line) # GLOBAL - English match = self.PATTERN_GLOBAL_EN.match(line) if match: return self._create_global_event_en(match, line) # PERSONAL GLOBAL - Swedish (when YOU get a global) match = self.PATTERN_PERSONAL_GLOBAL_SV.match(line) if match: return self._create_personal_global_event(match, line, 'swedish') # PERSONAL GLOBAL - English (when YOU get a global) match = self.PATTERN_PERSONAL_GLOBAL_EN.match(line) if match: return self._create_personal_global_event(match, line, 'english') # HOF - English match = self.PATTERN_HOF_EN.match(line) if match: return self._create_hof_event(match, line, 'english') # HOF Marker match = self.PATTERN_HOF_MARKER.match(line) if match: return LogEvent( timestamp=self._parse_timestamp(match.group(1)), event_type='hof', raw_line=line, data={'message': 'Hall of Fame entry'} ) # SKILL - Swedish match = self.PATTERN_SKILL_SV.match(line) if match: return LogEvent( timestamp=self._parse_timestamp(match.group(1)), event_type='skill', raw_line=line, data={ 'gained': Decimal(match.group(2)), 'skill_name': match.group(3).strip(), 'language': 'swedish' } ) # SKILL - English match = self.PATTERN_SKILL_EN.match(line) if match: return LogEvent( timestamp=self._parse_timestamp(match.group(1)), event_type='skill', raw_line=line, data={ 'gained': Decimal(match.group(2)), 'skill_name': match.group(3).strip(), 'language': 'english' } ) # LEVEL UP - English match = self.PATTERN_LEVEL_UP_EN.match(line) if match: return LogEvent( timestamp=self._parse_timestamp(match.group(1)), event_type='level_up', raw_line=line, data={ 'new_level': int(match.group(2)), 'skill_name': match.group(3).strip(), 'language': 'english' } ) # LEVEL UP - Swedish match = self.PATTERN_LEVEL_UP_SV.match(line) if match: return LogEvent( timestamp=self._parse_timestamp(match.group(1)), event_type='level_up', raw_line=line, data={ 'new_level': int(match.group(2)), 'skill_name': match.group(3).strip(), 'language': 'swedish' } ) # DAMAGE DEALT - Swedish match = self.PATTERN_DAMAGE_DEALT_SV.match(line) if match: return LogEvent( timestamp=self._parse_timestamp(match.group(1)), event_type='damage_dealt', raw_line=line, data={'damage': Decimal(match.group(2)), 'language': 'swedish'} ) # DAMAGE DEALT - English match = self.PATTERN_DAMAGE_DEALT_EN.match(line) if match: return LogEvent( timestamp=self._parse_timestamp(match.group(1)), event_type='damage_dealt', raw_line=line, data={'damage': Decimal(match.group(2)), 'language': 'english'} ) # CRITICAL HIT - Swedish match = self.PATTERN_CRITICAL_SV.match(line) if match: return LogEvent( timestamp=self._parse_timestamp(match.group(1)), event_type='critical_hit', raw_line=line, data={'damage': Decimal(match.group(2)), 'language': 'swedish'} ) # CRITICAL HIT - English match = self.PATTERN_CRITICAL_EN.match(line) if match: return LogEvent( timestamp=self._parse_timestamp(match.group(1)), event_type='critical_hit', raw_line=line, data={'damage': Decimal(match.group(2)), 'language': 'english'} ) # DAMAGE TAKEN - Swedish match = self.PATTERN_DAMAGE_TAKEN_SV.match(line) if match: return LogEvent( timestamp=self._parse_timestamp(match.group(1)), event_type='damage_taken', raw_line=line, data={'damage': Decimal(match.group(2)), 'language': 'swedish'} ) # DAMAGE TAKEN - English match = self.PATTERN_DAMAGE_TAKEN_EN.match(line) if match: return LogEvent( timestamp=self._parse_timestamp(match.group(1)), event_type='damage_taken', raw_line=line, data={'damage': Decimal(match.group(2)), 'language': 'english'} ) # HEALING - Swedish match = self.PATTERN_HEAL_SV.match(line) if match: return LogEvent( timestamp=self._parse_timestamp(match.group(1)), event_type='heal', raw_line=line, data={'heal_amount': Decimal(match.group(2)), 'language': 'swedish'} ) # HEALING - English match = self.PATTERN_HEAL_EN.match(line) if match: return LogEvent( timestamp=self._parse_timestamp(match.group(1)), event_type='heal', raw_line=line, data={'heal_amount': Decimal(match.group(2)), 'language': 'english'} ) # WEAPON TIER/LEVEL - Swedish match = self.PATTERN_WEAPON_TIER_SV.match(line) if match: return LogEvent( timestamp=self._parse_timestamp(match.group(1)), event_type='weapon_tier', raw_line=line, data={ 'weapon_name': match.group(2).strip(), 'new_tier': Decimal(match.group(3)), 'language': 'swedish' } ) # WEAPON TIER/LEVEL - English match = self.PATTERN_WEAPON_TIER_EN.match(line) if match: return LogEvent( timestamp=self._parse_timestamp(match.group(1)), event_type='weapon_tier', raw_line=line, data={ 'weapon_name': match.group(2).strip(), 'new_tier': Decimal(match.group(3)), 'language': 'english' } ) # EVADE/DODGE/MISS - English match = self.PATTERN_EVADE_EN.match(line) if match: return LogEvent( timestamp=self._parse_timestamp(match.group(1)), event_type='evade', raw_line=line, data={'type': match.group(2), 'language': 'english'} ) # EVADE/DODGE/MISS - Swedish match = self.PATTERN_EVADE_SV.match(line) if match: return LogEvent( timestamp=self._parse_timestamp(match.group(1)), event_type='evade', raw_line=line, data={'type': match.group(2), 'language': 'swedish'} ) # DECAY - English match = self.PATTERN_DECAY_EN.match(line) if match: return LogEvent( timestamp=self._parse_timestamp(match.group(1)), event_type='decay', raw_line=line, data={ 'item': match.group(2).strip(), 'amount_pec': Decimal(match.group(3)), 'language': 'english' } ) # DECAY - Swedish match = self.PATTERN_DECAY_SV.match(line) if match: return LogEvent( timestamp=self._parse_timestamp(match.group(1)), event_type='decay', raw_line=line, data={ 'item': match.group(2).strip(), 'amount_pec': Decimal(match.group(3)), 'language': 'swedish' } ) # BROKEN ENHANCER - English match = self.PATTERN_ENHANCER_BROKEN_EN.match(line) if match: return LogEvent( timestamp=self._parse_timestamp(match.group(1)), event_type='enhancer_broken', raw_line=line, data={ 'enhancer_type': match.group(2).strip(), 'weapon': match.group(3).strip(), 'language': 'english' } ) # ATTRIBUTE GAIN - Swedish match = self.PATTERN_ATTRIBUTE_SV.match(line) if match: return LogEvent( timestamp=self._parse_timestamp(match.group(1)), event_type='attribute', raw_line=line, data={ 'attribute': match.group(2), 'increase': Decimal(match.group(3)), 'language': 'swedish' } ) # ATTRIBUTE GAIN - English match = self.PATTERN_ATTRIBUTE_EN.match(line) if match: return LogEvent( timestamp=self._parse_timestamp(match.group(1)), event_type='attribute', raw_line=line, data={ 'attribute': match.group(4) if match.group(4) else match.group(2), 'increase': Decimal(match.group(3)), 'language': 'english' } ) # TEAM SHARE - English match = self.PATTERN_TEAM_SHARE_EN.match(line) if match: return LogEvent( timestamp=self._parse_timestamp(match.group(1)), event_type='team_share', raw_line=line, data={ 'amount_ped': Decimal(match.group(2)), 'language': 'english' } ) # TEAM SHARE - Swedish match = self.PATTERN_TEAM_SHARE_SV.match(line) if match: return LogEvent( timestamp=self._parse_timestamp(match.group(1)), event_type='team_share', raw_line=line, data={ 'amount_ped': Decimal(match.group(2)), 'language': 'swedish' } ) return None def _create_loot_event_sv(self, match: re.Match, line: str) -> LogEvent: """Create loot event from Swedish pattern.""" item_name = match.group(2).strip() quantity = int(match.group(3)) value_ped = Decimal(match.group(4)) loot_item = self._categorize_loot(item_name, value_ped) loot_item.quantity = quantity return LogEvent( timestamp=self._parse_timestamp(match.group(1)), event_type='loot', raw_line=line, data={ 'item_name': loot_item.name, 'quantity': loot_item.quantity, 'value_ped': loot_item.value_ped, 'is_shrapnel': loot_item.is_shrapnel, 'is_universal_ammo': loot_item.is_universal_ammo, 'language': 'swedish' } ) def _create_loot_event_en(self, match: re.Match, line: str) -> LogEvent: """Create loot event from English pattern.""" item_name = match.group(2).strip() quantity = int(match.group(3)) value_ped = Decimal(match.group(4)) if match.group(4) else Decimal('0') loot_item = self._categorize_loot(item_name, value_ped) loot_item.quantity = quantity return LogEvent( timestamp=self._parse_timestamp(match.group(1)), event_type='loot', raw_line=line, data={ 'item_name': loot_item.name, 'quantity': loot_item.quantity, 'value_ped': loot_item.value_ped, 'is_shrapnel': loot_item.is_shrapnel, 'is_universal_ammo': loot_item.is_universal_ammo, 'language': 'english' } ) def _create_loot_event_no_value(self, match: re.Match, line: str, language: str) -> LogEvent: """Create loot event without value.""" item_name = match.group(2).strip() quantity = int(match.group(3)) loot_item = self._categorize_loot(item_name, Decimal('0')) loot_item.quantity = quantity return LogEvent( timestamp=self._parse_timestamp(match.group(1)), event_type='loot', raw_line=line, data={ 'item_name': loot_item.name, 'quantity': loot_item.quantity, 'value_ped': loot_item.value_ped, 'is_shrapnel': loot_item.is_shrapnel, 'is_universal_ammo': loot_item.is_universal_ammo, 'language': language } ) def _create_kill_event(self, match: re.Match, line: str, language: str) -> LogEvent: """Create kill event.""" return LogEvent( timestamp=self._parse_timestamp(match.group(1)), event_type='kill', raw_line=line, data={ 'creature_name': match.group(2).strip(), 'language': language } ) def _create_global_event_sv(self, match: re.Match, line: str) -> LogEvent: """Create global event from Swedish pattern.""" return LogEvent( timestamp=self._parse_timestamp(match.group(1)), event_type='global', raw_line=line, data={ 'player_name': match.group(2).strip(), 'item': match.group(3).strip(), 'value_ped': Decimal(match.group(4)), 'language': 'swedish' } ) def _create_global_event_en(self, match: re.Match, line: str) -> LogEvent: """Create global event from English pattern.""" return LogEvent( timestamp=self._parse_timestamp(match.group(1)), event_type='global', raw_line=line, data={ 'player_name': match.group(2).strip(), 'zone': match.group(3).strip(), 'value_ped': Decimal(match.group(4)), 'language': 'english' } ) def _create_personal_global_event(self, match: re.Match, line: str, language: str) -> LogEvent: """Create personal global event (when YOU get a global).""" return LogEvent( timestamp=self._parse_timestamp(match.group(1)), event_type='personal_global', raw_line=line, data={ 'player_name': match.group(2).strip(), 'creature': match.group(3).strip(), 'value_ped': Decimal(match.group(4)), 'language': language } ) def _create_hof_event(self, match: re.Match, line: str, language: str) -> LogEvent: """Create Hall of Fame event.""" return LogEvent( timestamp=self._parse_timestamp(match.group(1)), event_type='hof', raw_line=line, data={ 'player_name': match.group(2).strip(), 'creature': match.group(3).strip(), 'value_ped': Decimal(match.group(4)), 'language': language } ) # ======================================================================== # ASYNC POLLING LOOP # ======================================================================== async def start(self) -> None: """Start watching log file asynchronously.""" if self._running: logger.warning("LogWatcher already running") return self._running = True if self.log_path.exists(): self._last_file_size = self.log_path.stat().st_size self._file_position = self._last_file_size self._task = asyncio.create_task(self._watch_loop()) logger.info("LogWatcher started") async def stop(self) -> None: """Stop watching log file.""" self._running = False if self._task: self._task.cancel() try: await self._task except asyncio.CancelledError: pass logger.info("LogWatcher stopped") async def _watch_loop(self) -> None: """Main watching loop.""" while self._running: try: await self._poll_once() except Exception as e: logger.error(f"Poll error: {e}") await asyncio.sleep(self.poll_interval) async def _poll_once(self) -> None: """Single poll iteration.""" if not self.log_path.exists(): return current_size = self.log_path.stat().st_size if current_size < self._file_position: logger.info("Log file truncated, resetting position") self._file_position = 0 if current_size == self._file_position: return with open(self.log_path, 'r', encoding='utf-8', errors='ignore') as f: f.seek(self._file_position) new_lines = f.readlines() self._file_position = f.tell() for line in new_lines: event = self._parse_line(line) if event: self._notify(event) self._last_file_size = current_size # ============================================================================ # HUNTING SESSION TRACKER # ============================================================================ class HuntingSessionTracker: """ Tracks hunting session statistics from LogWatcher events. This class accumulates all hunting-related data and provides real-time metrics like profit/loss, return percentage, etc. """ def __init__(self): self.stats = HuntingSessionStats() self._session_start: Optional[datetime] = None self._session_active = False # Callbacks for real-time updates self._on_stats_update: Optional[Callable] = None def start_session(self): """Start a new hunting session.""" self._session_start = datetime.now() self._session_active = True self.stats = HuntingSessionStats() logger.info("Hunting session started") def end_session(self) -> HuntingSessionStats: """End the current hunting session and return final stats.""" self._session_active = False logger.info("Hunting session ended") return self.stats def is_active(self) -> bool: """Check if session is active.""" return self._session_active def set_stats_callback(self, callback: Callable[[HuntingSessionStats], None]): """Set callback for real-time stats updates.""" self._on_stats_update = callback def _notify_update(self): """Notify listeners of stats update.""" if self._on_stats_update: try: self._on_stats_update(self.stats) except Exception as e: logger.error(f"Stats callback error: {e}") def on_loot(self, event: LogEvent): """Process loot event.""" if not self._session_active: return data = event.data value_ped = data.get('value_ped', Decimal('0.0')) is_shrapnel = data.get('is_shrapnel', False) is_ua = data.get('is_universal_ammo', False) self.stats.total_loot_ped += value_ped if is_shrapnel: self.stats.total_shrapnel_ped += value_ped elif is_ua: self.stats.total_universal_ammo_ped += value_ped else: self.stats.total_other_loot_ped += value_ped self._notify_update() def on_kill(self, event: LogEvent): """Process kill event.""" if not self._session_active: return self.stats.kills += 1 self._notify_update() def on_damage_dealt(self, event: LogEvent): """Process damage dealt event.""" if not self._session_active: return damage = event.data.get('damage', Decimal('0.0')) self.stats.damage_dealt += damage self.stats.shots_fired += 1 # Each damage event = 1 shot self._notify_update() def on_damage_taken(self, event: LogEvent): """Process damage taken event.""" if not self._session_active: return damage = event.data.get('damage', Decimal('0.0')) self.stats.damage_taken += damage self._notify_update() def on_heal(self, event: LogEvent): """Process heal event.""" if not self._session_active: return heal_amount = event.data.get('heal_amount', Decimal('0.0')) self.stats.healing_done += heal_amount self._notify_update() def on_global(self, event: LogEvent): """Process global event.""" if not self._session_active: return self.stats.globals_count += 1 # Store personal global details if event.event_type == 'personal_global': self.stats.personal_globals.append({ 'timestamp': event.timestamp, 'creature': event.data.get('creature', 'Unknown'), 'value_ped': event.data.get('value_ped', Decimal('0.0')) }) self._notify_update() def on_hof(self, event: LogEvent): """Process Hall of Fame event.""" if not self._session_active: return self.stats.hofs_count += 1 # Store HoF details if 'creature' in event.data: self.stats.personal_globals.append({ 'timestamp': event.timestamp, 'creature': event.data.get('creature', 'Unknown'), 'value_ped': event.data.get('value_ped', Decimal('0.0')), 'is_hof': True }) self._notify_update() def on_decay(self, event: LogEvent): """Process decay event.""" if not self._session_active: return # Convert PEC to PED amount_pec = event.data.get('amount_pec', Decimal('0.0')) amount_ped = amount_pec / Decimal('100') self.stats.weapon_cost_ped += amount_ped self.stats.total_cost_ped += amount_ped self._notify_update() def add_weapon_cost(self, cost_ped: Decimal): """Manually add weapon cost (for calculated decay).""" if not self._session_active: return self.stats.weapon_cost_ped += cost_ped self.stats.total_cost_ped += cost_ped self._notify_update() def add_armor_cost(self, cost_ped: Decimal): """Manually add armor cost.""" if not self._session_active: return self.stats.armor_cost_ped += cost_ped self.stats.total_cost_ped += cost_ped self._notify_update() def add_healing_cost(self, cost_ped: Decimal): """Manually add healing cost.""" if not self._session_active: return self.stats.healing_cost_ped += cost_ped self.stats.total_cost_ped += cost_ped self._notify_update() def get_stats(self) -> HuntingSessionStats: """Get current stats.""" return self.stats def get_summary(self) -> Dict[str, Any]: """Get session summary as dictionary.""" return { 'session_active': self._session_active, 'session_start': self._session_start.isoformat() if self._session_start else None, 'total_loot_ped': float(self.stats.total_loot_ped), 'total_shrapnel_ped': float(self.stats.total_shrapnel_ped), 'total_universal_ammo_ped': float(self.stats.total_universal_ammo_ped), 'total_other_loot_ped': float(self.stats.total_other_loot_ped), 'total_cost_ped': float(self.stats.total_cost_ped), 'weapon_cost_ped': float(self.stats.weapon_cost_ped), 'armor_cost_ped': float(self.stats.armor_cost_ped), 'healing_cost_ped': float(self.stats.healing_cost_ped), 'net_profit_ped': float(self.stats.net_profit_ped), 'return_percentage': float(self.stats.return_percentage), 'cost_per_kill': float(self.stats.cost_per_kill), 'dpp': float(self.stats.dpp), 'damage_dealt': float(self.stats.damage_dealt), 'damage_taken': float(self.stats.damage_taken), 'healing_done': float(self.stats.healing_done), 'shots_fired': self.stats.shots_fired, 'kills': self.stats.kills, 'globals_count': self.stats.globals_count, 'hofs_count': self.stats.hofs_count, 'damage_per_kill': float(self.stats.damage_per_kill), } # ============================================================================ # MOCK MODE SUPPORT # ============================================================================ class MockLogGenerator: """Generates mock log entries for testing.""" MOCK_LINES = [ "2026-02-08 14:23:15 [System] You received Shrapnel x 123 (Value: 1.23 PED)", "2026-02-08 14:23:45 [System] You gained 0.45 experience in your Rifle skill", "2026-02-08 14:24:02 [System] Your Omegaton M2100 has decayed 15 PEC", "2026-02-08 14:25:30 [Globals] PlayerOne globals in Twin Peaks for 150.00 PED", "2026-02-08 14:26:10 [System] You received Animal Thyroid Oil x 5 (Value: 2.50 PED)", "2026-02-08 14:26:15 [System] You killed Araneatrox Young", "2026-02-08 14:27:55 [System] Congratulations! You have advanced to level 45 in Rifle", "2026-02-08 14:28:30 [Globals] You killed a creature (Cornundacauda) with a value of 75.00 PED", "2026-02-08 14:30:00 [Hall of Fame] PlayerTwo killed a creature (Atrox) for 2500.00 PED", "2026-02-08 14:31:15 [System] You received Universal Ammo x 50 (Value: 0.50 PED)", "2026-02-08 14:32:20 [System] You inflicted 45.5 points of damage", "2026-02-08 14:32:25 [System] You took 12.3 points of damage", "2026-02-08 14:33:00 [System] You healed yourself 25.0 points", "2026-02-08 14:34:10 [System] Critical hit - Additional damage! You inflicted 89.2 points of damage", # Swedish examples "2025-09-23 19:36:43 [System] Du fick Shrapnel x (4627) Värde: 0.4627 PED", "2025-09-23 19:36:08 [System] Du har fått 0.3238 erfarenhet i din Translocation färdighet", "2025-09-23 19:36:18 [System] Du orsakade 13.5 poäng skada", "2025-09-23 19:37:00 [System] Du dödade Araneatrox Young", ] @classmethod def create_mock_file(cls, path: Path, lines: int = 100) -> None: """Create a mock chat.log file.""" path.parent.mkdir(parents=True, exist_ok=True) with open(path, 'w') as f: for i in range(lines): line = cls.MOCK_LINES[i % len(cls.MOCK_LINES)] f.write(f"{line}\n") logger.info(f"Created mock log: {path} ({lines} lines)") # ============================================================================ # MODULE EXPORTS # ============================================================================ __all__ = [ 'LogWatcher', 'LogEvent', 'LootItem', 'HuntingSessionStats', 'HuntingSessionTracker', 'MockLogGenerator' ]