1008 lines
34 KiB
Python
1008 lines
34 KiB
Python
# Description: ProjectManager implementing the Data Principle
|
|
# Every session (Hunt/Mine/Craft) is a Project with auto-save, archive, reload
|
|
# Standards: Python 3.11+, type hints, Decimal precision for PED/PEC
|
|
|
|
from decimal import Decimal
|
|
from datetime import datetime, timedelta
|
|
from typing import Optional, Dict, Any, List
|
|
from dataclasses import dataclass, asdict, field
|
|
import json
|
|
import logging
|
|
|
|
from .database import DatabaseManager
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
|
|
@dataclass
|
|
class ProjectData:
|
|
"""Data class representing a Project (Data Principle core)."""
|
|
id: Optional[int] = None
|
|
name: str = ""
|
|
type: str = "hunt" # hunt, mine, craft, inventory
|
|
status: str = "active" # active, paused, completed, archived
|
|
created_at: Optional[datetime] = None
|
|
updated_at: Optional[datetime] = None
|
|
archived_at: Optional[datetime] = None
|
|
metadata: Dict[str, Any] = None
|
|
|
|
def __post_init__(self):
|
|
if self.metadata is None:
|
|
self.metadata = {}
|
|
|
|
|
|
@dataclass
|
|
class SessionData:
|
|
"""Data class representing a gameplay Session."""
|
|
id: Optional[int] = None
|
|
project_id: Optional[int] = None
|
|
started_at: Optional[datetime] = None
|
|
ended_at: Optional[datetime] = None
|
|
duration_seconds: int = 0
|
|
total_spent_ped: Decimal = Decimal("0.0")
|
|
total_return_ped: Decimal = Decimal("0.0")
|
|
net_profit_ped: Decimal = Decimal("0.0")
|
|
notes: str = ""
|
|
|
|
|
|
@dataclass
|
|
class HuntingSessionData:
|
|
"""
|
|
Extended data class for hunting sessions with detailed metrics.
|
|
|
|
This class tracks all hunting-specific statistics including:
|
|
- Loot breakdown (shrapnel, universal ammo, other loot)
|
|
- Cost breakdown (weapon, armor, healing, plates)
|
|
- Combat statistics (damage, kills, shots fired)
|
|
- Special events (globals, HoFs)
|
|
- Calculated efficiency metrics (DPP, return %, cost/kill)
|
|
"""
|
|
id: Optional[int] = None
|
|
session_id: Optional[int] = None
|
|
|
|
# Timestamps
|
|
started_at: Optional[datetime] = None
|
|
ended_at: Optional[datetime] = None
|
|
|
|
# Loot tracking (PED values)
|
|
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") # Actual marketable loot
|
|
|
|
# Cost tracking (PED values)
|
|
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")
|
|
enhancer_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
|
|
shots_missed: int = 0
|
|
evades: 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)
|
|
|
|
# Skill gains (tracked separately)
|
|
skill_gains: Dict[str, Decimal] = field(default_factory=dict)
|
|
|
|
# Equipment used
|
|
weapon_name: str = ""
|
|
weapon_dpp: Decimal = Decimal("0.0")
|
|
armor_name: str = ""
|
|
fap_name: str = ""
|
|
|
|
# Calculated properties
|
|
@property
|
|
def net_profit_ped(self) -> Decimal:
|
|
"""Net profit (excluding shrapnel from loot value)."""
|
|
return self.total_other_loot_ped - self.total_cost_ped
|
|
|
|
@property
|
|
def return_percentage(self) -> Decimal:
|
|
"""Return percentage (other_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 return_percentage_with_shrapnel(self) -> Decimal:
|
|
"""Return percentage including shrapnel."""
|
|
if self.total_cost_ped > 0:
|
|
return (self.total_loot_ped / self.total_cost_ped) * Decimal('100')
|
|
return Decimal('0.0')
|
|
|
|
@property
|
|
def cost_per_kill(self) -> Decimal:
|
|
"""Average cost per kill."""
|
|
if self.kills > 0:
|
|
return self.total_cost_ped / self.kills
|
|
return Decimal('0.0')
|
|
|
|
@property
|
|
def cost_per_damage(self) -> Decimal:
|
|
"""Cost per damage point (PED/damage)."""
|
|
if self.damage_dealt > 0:
|
|
return self.total_cost_ped / self.damage_dealt
|
|
return Decimal('0.0')
|
|
|
|
@property
|
|
def dpp(self) -> Decimal:
|
|
"""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:
|
|
"""Average damage per kill."""
|
|
if self.kills > 0:
|
|
return self.damage_dealt / self.kills
|
|
return Decimal('0.0')
|
|
|
|
@property
|
|
def loot_per_kill(self) -> Decimal:
|
|
"""Average loot per kill."""
|
|
if self.kills > 0:
|
|
return self.total_other_loot_ped / self.kills
|
|
return Decimal('0.0')
|
|
|
|
@property
|
|
def accuracy(self) -> Decimal:
|
|
"""Hit accuracy percentage."""
|
|
total_shots = self.shots_fired + self.shots_missed
|
|
if total_shots > 0:
|
|
return (self.shots_fired / total_shots) * Decimal('100')
|
|
return Decimal('0.0')
|
|
|
|
@property
|
|
def session_duration(self) -> Optional[timedelta]:
|
|
"""Calculate session duration."""
|
|
if self.started_at and self.ended_at:
|
|
return self.ended_at - self.started_at
|
|
elif self.started_at:
|
|
return datetime.now() - self.started_at
|
|
return None
|
|
|
|
@property
|
|
def kills_per_hour(self) -> Decimal:
|
|
"""Kill rate per hour."""
|
|
duration = self.session_duration
|
|
if duration and duration.total_seconds() > 0:
|
|
hours = Decimal(str(duration.total_seconds())) / Decimal('3600')
|
|
return Decimal(str(self.kills)) / hours
|
|
return Decimal('0.0')
|
|
|
|
@property
|
|
def cost_per_hour(self) -> Decimal:
|
|
"""Cost rate per hour."""
|
|
duration = self.session_duration
|
|
if duration and duration.total_seconds() > 0:
|
|
hours = Decimal(str(duration.total_seconds())) / Decimal('3600')
|
|
return self.total_cost_ped / hours
|
|
return Decimal('0.0')
|
|
|
|
@property
|
|
def profit_per_hour(self) -> Decimal:
|
|
"""Profit rate per hour."""
|
|
duration = self.session_duration
|
|
if duration and duration.total_seconds() > 0:
|
|
hours = Decimal(str(duration.total_seconds())) / Decimal('3600')
|
|
return self.net_profit_ped / hours
|
|
return Decimal('0.0')
|
|
|
|
def to_dict(self) -> Dict[str, Any]:
|
|
"""Convert to dictionary for serialization."""
|
|
return {
|
|
'id': self.id,
|
|
'session_id': self.session_id,
|
|
'started_at': self.started_at.isoformat() if self.started_at else None,
|
|
'ended_at': self.ended_at.isoformat() if self.ended_at else None,
|
|
'total_loot_ped': float(self.total_loot_ped),
|
|
'total_shrapnel_ped': float(self.total_shrapnel_ped),
|
|
'total_universal_ammo_ped': float(self.total_universal_ammo_ped),
|
|
'total_other_loot_ped': float(self.total_other_loot_ped),
|
|
'weapon_cost_ped': float(self.weapon_cost_ped),
|
|
'armor_cost_ped': float(self.armor_cost_ped),
|
|
'healing_cost_ped': float(self.healing_cost_ped),
|
|
'plates_cost_ped': float(self.plates_cost_ped),
|
|
'enhancer_cost_ped': float(self.enhancer_cost_ped),
|
|
'total_cost_ped': float(self.total_cost_ped),
|
|
'damage_dealt': float(self.damage_dealt),
|
|
'damage_taken': float(self.damage_taken),
|
|
'healing_done': float(self.healing_done),
|
|
'shots_fired': self.shots_fired,
|
|
'shots_missed': self.shots_missed,
|
|
'evades': self.evades,
|
|
'kills': self.kills,
|
|
'globals_count': self.globals_count,
|
|
'hofs_count': self.hofs_count,
|
|
'net_profit_ped': float(self.net_profit_ped),
|
|
'return_percentage': float(self.return_percentage),
|
|
'return_percentage_with_shrapnel': float(self.return_percentage_with_shrapnel),
|
|
'cost_per_kill': float(self.cost_per_kill),
|
|
'dpp': float(self.dpp),
|
|
'damage_per_kill': float(self.damage_per_kill),
|
|
'loot_per_kill': float(self.loot_per_kill),
|
|
'kills_per_hour': float(self.kills_per_hour),
|
|
'cost_per_hour': float(self.cost_per_hour),
|
|
'profit_per_hour': float(self.profit_per_hour),
|
|
'weapon_name': self.weapon_name,
|
|
'weapon_dpp': float(self.weapon_dpp),
|
|
'armor_name': self.armor_name,
|
|
'fap_name': self.fap_name,
|
|
}
|
|
|
|
|
|
@dataclass
|
|
class LootEvent:
|
|
"""Data class representing a loot event from chat.log."""
|
|
id: Optional[int] = None
|
|
session_id: Optional[int] = None
|
|
timestamp: Optional[datetime] = None
|
|
event_type: str = "regular" # global, hof, regular, skill
|
|
item_name: Optional[str] = None
|
|
quantity: int = 1
|
|
value_ped: Decimal = Decimal("0.0")
|
|
is_shrapnel: bool = False
|
|
is_universal_ammo: bool = False
|
|
creature_name: Optional[str] = None
|
|
zone_name: Optional[str] = None
|
|
raw_log_line: Optional[str] = None
|
|
screenshot_path: Optional[str] = None
|
|
|
|
|
|
@dataclass
|
|
class CombatEvent:
|
|
"""Data class representing a combat event."""
|
|
id: Optional[int] = None
|
|
session_id: Optional[int] = None
|
|
timestamp: Optional[datetime] = None
|
|
event_type: str = "damage_dealt" # damage_dealt, damage_taken, heal, evade, kill
|
|
damage_amount: Optional[Decimal] = None
|
|
heal_amount: Optional[Decimal] = None
|
|
creature_name: Optional[str] = None
|
|
weapon_name: Optional[str] = None
|
|
raw_log_line: Optional[str] = None
|
|
|
|
|
|
class ProjectManager:
|
|
"""
|
|
Manages Projects implementing the Data Principle.
|
|
|
|
The Data Principle states: Every session (Hunt/Mine/Craft) is a Project.
|
|
- Projects are auto-saved
|
|
- Projects can be archived and reloaded
|
|
- Historical data is comparable to current live data
|
|
|
|
Attributes:
|
|
db: DatabaseManager instance for persistence
|
|
current_project: Currently active ProjectData
|
|
current_session: Currently active SessionData
|
|
current_hunting_session: Currently active HuntingSessionData
|
|
"""
|
|
|
|
def __init__(self, db: Optional[DatabaseManager] = None):
|
|
"""
|
|
Initialize ProjectManager.
|
|
|
|
Args:
|
|
db: DatabaseManager instance. Creates default if None.
|
|
"""
|
|
self.db = db if db else DatabaseManager()
|
|
self.current_project: Optional[ProjectData] = None
|
|
self.current_session: Optional[SessionData] = None
|
|
self.current_hunting_session: Optional[HuntingSessionData] = None
|
|
|
|
logger.info("ProjectManager initialized")
|
|
|
|
# ========================================================================
|
|
# PROJECT OPERATIONS
|
|
# ========================================================================
|
|
|
|
def create_project(self, name: str, project_type: str,
|
|
metadata: Optional[Dict] = None) -> ProjectData:
|
|
"""
|
|
Create a new project.
|
|
|
|
Args:
|
|
name: Project name
|
|
project_type: One of: hunt, mine, craft, inventory
|
|
metadata: Optional project metadata dict
|
|
|
|
Returns:
|
|
Created ProjectData with assigned ID
|
|
"""
|
|
project = ProjectData(
|
|
name=name,
|
|
type=project_type,
|
|
status="active",
|
|
metadata=metadata or {}
|
|
)
|
|
|
|
cursor = self.db.execute(
|
|
"""
|
|
INSERT INTO projects (name, type, status, metadata)
|
|
VALUES (?, ?, ?, ?)
|
|
""",
|
|
(name, project_type, "active", json.dumps(metadata or {}))
|
|
)
|
|
self.db.commit()
|
|
|
|
project.id = cursor.lastrowid
|
|
logger.info(f"Created project: {name} (ID: {project.id}, Type: {project_type})")
|
|
|
|
return project
|
|
|
|
def load_project(self, project_id: int) -> Optional[ProjectData]:
|
|
"""
|
|
Load a project by ID.
|
|
|
|
Args:
|
|
project_id: Project ID to load
|
|
|
|
Returns:
|
|
ProjectData if found, None otherwise
|
|
"""
|
|
cursor = self.db.execute(
|
|
"SELECT * FROM projects WHERE id = ?",
|
|
(project_id,)
|
|
)
|
|
row = cursor.fetchone()
|
|
|
|
if not row:
|
|
logger.warning(f"Project not found: {project_id}")
|
|
return None
|
|
|
|
project = ProjectData(
|
|
id=row['id'],
|
|
name=row['name'],
|
|
type=row['type'],
|
|
status=row['status'],
|
|
created_at=row['created_at'],
|
|
updated_at=row['updated_at'],
|
|
archived_at=row['archived_at'],
|
|
metadata=json.loads(row['metadata']) if row['metadata'] else {}
|
|
)
|
|
|
|
self.current_project = project
|
|
logger.info(f"Loaded project: {project.name} (ID: {project.id})")
|
|
|
|
return project
|
|
|
|
def list_projects(self, project_type: Optional[str] = None,
|
|
status: Optional[str] = None) -> List[ProjectData]:
|
|
"""
|
|
List all projects with optional filtering.
|
|
|
|
Args:
|
|
project_type: Filter by type (hunt, mine, craft, inventory)
|
|
status: Filter by status (active, paused, completed, archived)
|
|
|
|
Returns:
|
|
List of ProjectData
|
|
"""
|
|
query = "SELECT * FROM projects WHERE 1=1"
|
|
params = []
|
|
|
|
if project_type:
|
|
query += " AND type = ?"
|
|
params.append(project_type)
|
|
|
|
if status:
|
|
query += " AND status = ?"
|
|
params.append(status)
|
|
|
|
query += " ORDER BY created_at DESC"
|
|
|
|
cursor = self.db.execute(query, tuple(params))
|
|
rows = cursor.fetchall()
|
|
|
|
projects = []
|
|
for row in rows:
|
|
projects.append(ProjectData(
|
|
id=row['id'],
|
|
name=row['name'],
|
|
type=row['type'],
|
|
status=row['status'],
|
|
created_at=row['created_at'],
|
|
updated_at=row['updated_at'],
|
|
archived_at=row['archived_at'],
|
|
metadata=json.loads(row['metadata']) if row['metadata'] else {}
|
|
))
|
|
|
|
return projects
|
|
|
|
def archive_project(self, project_id: int) -> bool:
|
|
"""
|
|
Archive a project.
|
|
|
|
Args:
|
|
project_id: Project ID to archive
|
|
|
|
Returns:
|
|
True if successful
|
|
"""
|
|
self.db.execute(
|
|
"""
|
|
UPDATE projects
|
|
SET status = 'archived', archived_at = CURRENT_TIMESTAMP
|
|
WHERE id = ?
|
|
""",
|
|
(project_id,)
|
|
)
|
|
self.db.commit()
|
|
logger.info(f"Archived project: {project_id}")
|
|
return True
|
|
|
|
# ========================================================================
|
|
# SESSION OPERATIONS
|
|
# ========================================================================
|
|
|
|
def start_session(self, project_id: int, notes: str = "") -> SessionData:
|
|
"""
|
|
Start a new session for a project.
|
|
|
|
Args:
|
|
project_id: Parent project ID
|
|
notes: Optional session notes
|
|
|
|
Returns:
|
|
Created SessionData
|
|
"""
|
|
cursor = self.db.execute(
|
|
"""
|
|
INSERT INTO sessions (project_id, notes)
|
|
VALUES (?, ?)
|
|
""",
|
|
(project_id, notes)
|
|
)
|
|
self.db.commit()
|
|
|
|
session = SessionData(
|
|
id=cursor.lastrowid,
|
|
project_id=project_id,
|
|
notes=notes
|
|
)
|
|
|
|
self.current_session = session
|
|
|
|
# Update project status
|
|
self.db.execute(
|
|
"UPDATE projects SET status = 'active' WHERE id = ?",
|
|
(project_id,)
|
|
)
|
|
self.db.commit()
|
|
|
|
logger.info(f"Started session {session.id} for project {project_id}")
|
|
return session
|
|
|
|
def start_hunting_session(self, project_id: int,
|
|
weapon_name: str = "",
|
|
weapon_dpp: Decimal = Decimal('0.0'),
|
|
armor_name: str = "",
|
|
fap_name: str = "") -> HuntingSessionData:
|
|
"""
|
|
Start a new hunting session with full tracking.
|
|
|
|
Args:
|
|
project_id: Parent project ID
|
|
weapon_name: Name of equipped weapon
|
|
weapon_dpp: Weapon DPP rating
|
|
armor_name: Name of equipped armor
|
|
fap_name: Name of equipped FAP
|
|
|
|
Returns:
|
|
Created HuntingSessionData
|
|
"""
|
|
# First create base session
|
|
base_session = self.start_session(project_id, notes=f"Hunting with {weapon_name}")
|
|
|
|
# Create hunting session data
|
|
hunting_session = HuntingSessionData(
|
|
session_id=base_session.id,
|
|
started_at=datetime.now(),
|
|
weapon_name=weapon_name,
|
|
weapon_dpp=weapon_dpp,
|
|
armor_name=armor_name,
|
|
fap_name=fap_name
|
|
)
|
|
|
|
self.current_hunting_session = hunting_session
|
|
|
|
# Insert into hunting_sessions table
|
|
cursor = self.db.execute(
|
|
"""
|
|
INSERT INTO hunting_sessions
|
|
(session_id, started_at, weapon_name, weapon_dpp, armor_name, fap_name)
|
|
VALUES (?, ?, ?, ?, ?, ?)
|
|
""",
|
|
(
|
|
base_session.id,
|
|
hunting_session.started_at,
|
|
weapon_name,
|
|
float(weapon_dpp),
|
|
armor_name,
|
|
fap_name
|
|
)
|
|
)
|
|
self.db.commit()
|
|
|
|
hunting_session.id = cursor.lastrowid
|
|
|
|
logger.info(f"Started hunting session {hunting_session.id} with {weapon_name}")
|
|
return hunting_session
|
|
|
|
def end_session(self, session_id: Optional[int] = None) -> bool:
|
|
"""
|
|
End a session and calculate final metrics.
|
|
|
|
Args:
|
|
session_id: Session ID (uses current_session if None)
|
|
|
|
Returns:
|
|
True if successful
|
|
"""
|
|
if session_id is None:
|
|
session_id = self.current_session.id if self.current_session else None
|
|
|
|
if not session_id:
|
|
logger.error("No session ID provided")
|
|
return False
|
|
|
|
# If we have an active hunting session, finalize it first
|
|
if self.current_hunting_session and self.current_hunting_session.session_id == session_id:
|
|
self._finalize_hunting_session()
|
|
|
|
# Calculate metrics
|
|
cursor = self.db.execute(
|
|
"""
|
|
SELECT
|
|
COALESCE(SUM(value_ped), 0) as total_return,
|
|
COALESCE(SUM(CASE WHEN event_type IN ('global', 'hof') THEN 1 ELSE 0 END), 0) as globals
|
|
FROM loot_events
|
|
WHERE session_id = ?
|
|
""",
|
|
(session_id,)
|
|
)
|
|
result = cursor.fetchone()
|
|
total_return = Decimal(str(result['total_return'])) if result else Decimal("0.0")
|
|
|
|
# Get decay (spent)
|
|
cursor = self.db.execute(
|
|
"SELECT COALESCE(SUM(decay_amount_ped), 0) as total_decay FROM decay_events WHERE session_id = ?",
|
|
(session_id,)
|
|
)
|
|
result = cursor.fetchone()
|
|
total_spent = Decimal(str(result['total_decay'])) if result else Decimal("0.0")
|
|
|
|
net_profit = total_return - total_spent
|
|
|
|
# End session
|
|
self.db.execute(
|
|
"""
|
|
UPDATE sessions
|
|
SET ended_at = CURRENT_TIMESTAMP,
|
|
total_return_ped = ?,
|
|
total_spent_ped = ?,
|
|
net_profit_ped = ?
|
|
WHERE id = ?
|
|
""",
|
|
(float(total_return), float(total_spent), float(net_profit), session_id)
|
|
)
|
|
self.db.commit()
|
|
|
|
if self.current_session and self.current_session.id == session_id:
|
|
self.current_session = None
|
|
|
|
logger.info(f"Ended session {session_id}. Profit: {net_profit} PED")
|
|
return True
|
|
|
|
def _finalize_hunting_session(self):
|
|
"""Finalize the current hunting session in the database."""
|
|
if not self.current_hunting_session:
|
|
return
|
|
|
|
hs = self.current_hunting_session
|
|
hs.ended_at = datetime.now()
|
|
|
|
self.db.execute(
|
|
"""
|
|
UPDATE hunting_sessions SET
|
|
ended_at = ?,
|
|
total_loot_ped = ?,
|
|
total_shrapnel_ped = ?,
|
|
total_universal_ammo_ped = ?,
|
|
total_other_loot_ped = ?,
|
|
weapon_cost_ped = ?,
|
|
armor_cost_ped = ?,
|
|
healing_cost_ped = ?,
|
|
plates_cost_ped = ?,
|
|
total_cost_ped = ?,
|
|
damage_dealt = ?,
|
|
damage_taken = ?,
|
|
healing_done = ?,
|
|
shots_fired = ?,
|
|
shots_missed = ?,
|
|
evades = ?,
|
|
kills = ?,
|
|
globals_count = ?,
|
|
hofs_count = ?
|
|
WHERE id = ?
|
|
""",
|
|
(
|
|
hs.ended_at,
|
|
float(hs.total_loot_ped),
|
|
float(hs.total_shrapnel_ped),
|
|
float(hs.total_universal_ammo_ped),
|
|
float(hs.total_other_loot_ped),
|
|
float(hs.weapon_cost_ped),
|
|
float(hs.armor_cost_ped),
|
|
float(hs.healing_cost_ped),
|
|
float(hs.plates_cost_ped),
|
|
float(hs.total_cost_ped),
|
|
float(hs.damage_dealt),
|
|
float(hs.damage_taken),
|
|
float(hs.healing_done),
|
|
hs.shots_fired,
|
|
hs.shots_missed,
|
|
hs.evades,
|
|
hs.kills,
|
|
hs.globals_count,
|
|
hs.hofs_count,
|
|
hs.id
|
|
)
|
|
)
|
|
self.db.commit()
|
|
|
|
# Store personal globals
|
|
for global_data in hs.personal_globals:
|
|
self.db.execute(
|
|
"""
|
|
INSERT INTO hunting_globals
|
|
(hunting_session_id, timestamp, creature_name, value_ped, is_hof)
|
|
VALUES (?, ?, ?, ?, ?)
|
|
""",
|
|
(
|
|
hs.id,
|
|
global_data.get('timestamp', datetime.now()),
|
|
global_data.get('creature', 'Unknown'),
|
|
float(global_data.get('value_ped', Decimal('0.0'))),
|
|
global_data.get('is_hof', False)
|
|
)
|
|
)
|
|
self.db.commit()
|
|
|
|
logger.info(f"Finalized hunting session {hs.id}")
|
|
|
|
def update_hunting_session(self, updates: Dict[str, Any]) -> bool:
|
|
"""
|
|
Update the current hunting session with new data.
|
|
|
|
Args:
|
|
updates: Dictionary of field updates
|
|
|
|
Returns:
|
|
True if updated successfully
|
|
"""
|
|
if not self.current_hunting_session:
|
|
logger.error("No active hunting session to update")
|
|
return False
|
|
|
|
hs = self.current_hunting_session
|
|
|
|
# Update fields
|
|
for key, value in updates.items():
|
|
if hasattr(hs, key):
|
|
setattr(hs, key, value)
|
|
|
|
# Recalculate totals
|
|
hs.total_cost_ped = (
|
|
hs.weapon_cost_ped +
|
|
hs.armor_cost_ped +
|
|
hs.healing_cost_ped +
|
|
hs.plates_cost_ped +
|
|
hs.enhancer_cost_ped
|
|
)
|
|
|
|
return True
|
|
|
|
# ========================================================================
|
|
# LOOT TRACKING
|
|
# ========================================================================
|
|
|
|
def record_loot(self, loot: LootEvent) -> bool:
|
|
"""
|
|
Record a loot event.
|
|
|
|
Args:
|
|
loot: LootEvent data
|
|
|
|
Returns:
|
|
True if recorded successfully
|
|
"""
|
|
if not self.current_session:
|
|
logger.error("No active session to record loot")
|
|
return False
|
|
|
|
loot.session_id = self.current_session.id
|
|
|
|
cursor = self.db.execute(
|
|
"""
|
|
INSERT INTO loot_events
|
|
(session_id, event_type, item_name, quantity, value_ped,
|
|
creature_name, zone_name, raw_log_line, screenshot_path)
|
|
VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)
|
|
""",
|
|
(
|
|
loot.session_id,
|
|
loot.event_type,
|
|
loot.item_name,
|
|
loot.quantity,
|
|
float(loot.value_ped),
|
|
loot.creature_name,
|
|
loot.zone_name,
|
|
loot.raw_log_line,
|
|
loot.screenshot_path
|
|
)
|
|
)
|
|
self.db.commit()
|
|
|
|
logger.debug(f"Recorded loot: {loot.item_name} ({loot.value_ped} PED)")
|
|
|
|
# Check for screenshot trigger (Rule: >50 PED)
|
|
if loot.value_ped >= Decimal("50.0") and loot.event_type in ('global', 'hof'):
|
|
self._trigger_screenshot(loot)
|
|
|
|
return True
|
|
|
|
def record_combat_event(self, event: CombatEvent) -> bool:
|
|
"""
|
|
Record a combat event.
|
|
|
|
Args:
|
|
event: CombatEvent data
|
|
|
|
Returns:
|
|
True if recorded successfully
|
|
"""
|
|
if not self.current_session:
|
|
logger.error("No active session to record combat event")
|
|
return False
|
|
|
|
event.session_id = self.current_session.id
|
|
|
|
self.db.execute(
|
|
"""
|
|
INSERT INTO combat_events
|
|
(session_id, timestamp, event_type, damage_amount, heal_amount,
|
|
creature_name, weapon_name, raw_log_line)
|
|
VALUES (?, ?, ?, ?, ?, ?, ?, ?)
|
|
""",
|
|
(
|
|
event.session_id,
|
|
event.timestamp or datetime.now(),
|
|
event.event_type,
|
|
float(event.damage_amount) if event.damage_amount else None,
|
|
float(event.heal_amount) if event.heal_amount else None,
|
|
event.creature_name,
|
|
event.weapon_name,
|
|
event.raw_log_line
|
|
)
|
|
)
|
|
self.db.commit()
|
|
|
|
return True
|
|
|
|
def _trigger_screenshot(self, loot: LootEvent) -> None:
|
|
"""
|
|
Trigger screenshot capture for high-value loot.
|
|
|
|
Args:
|
|
loot: The loot event that triggered screenshot
|
|
"""
|
|
# Placeholder - actual implementation in screenshot module
|
|
logger.info(f"Screenshot triggered for {loot.value_ped} PED {loot.event_type}")
|
|
|
|
# ========================================================================
|
|
# ANALYTICS
|
|
# ========================================================================
|
|
|
|
def get_project_summary(self, project_id: int) -> Optional[Dict[str, Any]]:
|
|
"""
|
|
Get summary statistics for a project.
|
|
|
|
Args:
|
|
project_id: Project ID
|
|
|
|
Returns:
|
|
Dictionary with summary statistics
|
|
"""
|
|
cursor = self.db.execute(
|
|
"SELECT * FROM v_project_summary WHERE id = ?",
|
|
(project_id,)
|
|
)
|
|
row = cursor.fetchone()
|
|
|
|
if not row:
|
|
return None
|
|
|
|
return {
|
|
'project_id': row['id'],
|
|
'name': row['name'],
|
|
'type': row['type'],
|
|
'status': row['status'],
|
|
'session_count': row['session_count'],
|
|
'total_spent': Decimal(str(row['total_spent'])) if row['total_spent'] else Decimal("0.0"),
|
|
'total_return': Decimal(str(row['total_return'])) if row['total_return'] else Decimal("0.0"),
|
|
'net_profit': Decimal(str(row['net_profit'])) if row['net_profit'] else Decimal("0.0"),
|
|
'global_count': row['global_count'],
|
|
'hof_count': row['hof_count']
|
|
}
|
|
|
|
def get_hunting_session_summary(self, hunting_session_id: int) -> Optional[Dict[str, Any]]:
|
|
"""
|
|
Get detailed summary for a hunting session.
|
|
|
|
Args:
|
|
hunting_session_id: Hunting session ID
|
|
|
|
Returns:
|
|
Dictionary with hunting session summary
|
|
"""
|
|
cursor = self.db.execute(
|
|
"SELECT * FROM hunting_sessions WHERE id = ?",
|
|
(hunting_session_id,)
|
|
)
|
|
row = cursor.fetchone()
|
|
|
|
if not row:
|
|
return None
|
|
|
|
# Build hunting session data
|
|
hs = HuntingSessionData(
|
|
id=row['id'],
|
|
session_id=row['session_id'],
|
|
started_at=row['started_at'],
|
|
ended_at=row['ended_at'],
|
|
total_loot_ped=Decimal(str(row['total_loot_ped'])),
|
|
total_shrapnel_ped=Decimal(str(row['total_shrapnel_ped'])),
|
|
total_universal_ammo_ped=Decimal(str(row['total_universal_ammo_ped'])),
|
|
total_other_loot_ped=Decimal(str(row['total_other_loot_ped'])),
|
|
weapon_cost_ped=Decimal(str(row['weapon_cost_ped'])),
|
|
armor_cost_ped=Decimal(str(row['armor_cost_ped'])),
|
|
healing_cost_ped=Decimal(str(row['healing_cost_ped'])),
|
|
plates_cost_ped=Decimal(str(row['plates_cost_ped'])),
|
|
total_cost_ped=Decimal(str(row['total_cost_ped'])),
|
|
damage_dealt=Decimal(str(row['damage_dealt'])),
|
|
damage_taken=Decimal(str(row['damage_taken'])),
|
|
healing_done=Decimal(str(row['healing_done'])),
|
|
shots_fired=row['shots_fired'],
|
|
shots_missed=row['shots_missed'],
|
|
evades=row['evades'],
|
|
kills=row['kills'],
|
|
globals_count=row['globals_count'],
|
|
hofs_count=row['hofs_count'],
|
|
weapon_name=row['weapon_name'],
|
|
weapon_dpp=Decimal(str(row['weapon_dpp'])),
|
|
armor_name=row['armor_name'],
|
|
fap_name=row['fap_name']
|
|
)
|
|
|
|
return hs.to_dict()
|
|
|
|
def get_hunting_history(self, project_id: int, limit: int = 10) -> List[Dict[str, Any]]:
|
|
"""
|
|
Get hunting session history for a project.
|
|
|
|
Args:
|
|
project_id: Project ID
|
|
limit: Maximum number of sessions to return
|
|
|
|
Returns:
|
|
List of hunting session summaries
|
|
"""
|
|
cursor = self.db.execute(
|
|
"""
|
|
SELECT hs.* FROM hunting_sessions hs
|
|
JOIN sessions s ON hs.session_id = s.id
|
|
WHERE s.project_id = ?
|
|
ORDER BY hs.started_at DESC
|
|
LIMIT ?
|
|
""",
|
|
(project_id, limit)
|
|
)
|
|
rows = cursor.fetchall()
|
|
|
|
history = []
|
|
for row in rows:
|
|
hs = HuntingSessionData(
|
|
id=row['id'],
|
|
session_id=row['session_id'],
|
|
started_at=row['started_at'],
|
|
ended_at=row['ended_at'],
|
|
total_loot_ped=Decimal(str(row['total_loot_ped'])),
|
|
total_other_loot_ped=Decimal(str(row['total_other_loot_ped'])),
|
|
total_cost_ped=Decimal(str(row['total_cost_ped'])),
|
|
kills=row['kills'],
|
|
globals_count=row['globals_count'],
|
|
hofs_count=row['hofs_count'],
|
|
weapon_name=row['weapon_name']
|
|
)
|
|
history.append(hs.to_dict())
|
|
|
|
return history
|
|
|
|
def compare_to_historical(self, project_id: int,
|
|
metric: str = 'net_profit') -> Dict[str, Any]:
|
|
"""
|
|
Compare current project to historical averages.
|
|
|
|
Implements Data Principle: Comparable historical data.
|
|
|
|
Args:
|
|
project_id: Current project ID
|
|
metric: Metric to compare (net_profit, roi, globals_per_hour)
|
|
|
|
Returns:
|
|
Comparison data
|
|
"""
|
|
# Get current project type
|
|
cursor = self.db.execute(
|
|
"SELECT type FROM projects WHERE id = ?",
|
|
(project_id,)
|
|
)
|
|
row = cursor.fetchone()
|
|
if not row:
|
|
return {}
|
|
|
|
project_type = row['type']
|
|
|
|
# Get historical averages for same type
|
|
cursor = self.db.execute(
|
|
"""
|
|
SELECT AVG(net_profit_ped) as avg_profit,
|
|
AVG(CASE WHEN total_spent_ped > 0
|
|
THEN (net_profit_ped / total_spent_ped) * 100
|
|
ELSE 0 END) as avg_roi
|
|
FROM sessions s
|
|
JOIN projects p ON s.project_id = p.id
|
|
WHERE p.type = ? AND p.id != ? AND p.status = 'archived'
|
|
""",
|
|
(project_type, project_id)
|
|
)
|
|
hist = cursor.fetchone()
|
|
|
|
# Get current project stats
|
|
current = self.get_project_summary(project_id)
|
|
|
|
return {
|
|
'current': current,
|
|
'historical_avg_profit': Decimal(str(hist['avg_profit'])) if hist['avg_profit'] else Decimal("0.0"),
|
|
'historical_avg_roi': hist['avg_roi'] or 0,
|
|
'comparison_metric': metric
|
|
}
|
|
|
|
|
|
# ============================================================================
|
|
# MODULE EXPORTS
|
|
# ============================================================================
|
|
|
|
__all__ = [
|
|
'ProjectManager',
|
|
'ProjectData',
|
|
'SessionData',
|
|
'HuntingSessionData',
|
|
'LootEvent',
|
|
'CombatEvent'
|
|
]
|