From 28b8921efa7e36dcd7f24c05e36d4d3a57852b9b Mon Sep 17 00:00:00 2001 From: LemonNexus Date: Sun, 8 Feb 2026 16:56:40 +0000 Subject: [PATCH] feat(core): implement ProjectManager with Data Principle - Create ProjectManager class enforcing Data Principle - Implement ProjectData, SessionData, LootEvent dataclasses - Add create_project, load_project, archive_project methods - Implement session lifecycle (start_session, end_session) - Add loot recording with auto-screenshot trigger (>50 PED) - Include analytics: get_project_summary, compare_to_historical - All PED calculations use Decimal for precision (Rule #4) Every session is a Project: auto-saved, archivable, comparable. --- core/project_manager.py | 489 ++++++++++++++++++++++++++++++++++++++++ 1 file changed, 489 insertions(+) create mode 100644 core/project_manager.py diff --git a/core/project_manager.py b/core/project_manager.py new file mode 100644 index 0000000..2e55df7 --- /dev/null +++ b/core/project_manager.py @@ -0,0 +1,489 @@ +# 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 +from typing import Optional, Dict, Any, List +from dataclasses import dataclass, asdict +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 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") + creature_name: Optional[str] = None + zone_name: Optional[str] = None + raw_log_line: Optional[str] = None + screenshot_path: 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 + """ + + 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 + + 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 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 + + # 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 + + # ======================================================================== + # 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 _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 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', + 'LootEvent' +]