1381 lines
50 KiB
Python
1381 lines
50 KiB
Python
# 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'
|
|
]
|