""" EU-Utility - Session Exporter Plugin Export hunting/mining/crafting sessions to CSV/JSON formats. Tracks loot, skills gained, globals, and other session data. """ import json import csv from datetime import datetime from pathlib import Path from typing import Dict, List, Any, Optional from dataclasses import dataclass, field, asdict from PyQt6.QtWidgets import ( QWidget, QVBoxLayout, QHBoxLayout, QLabel, QPushButton, QTableWidget, QTableWidgetItem, QHeaderView, QComboBox, QFileDialog, QMessageBox, QGroupBox, QSpinBox, QCheckBox, QTabWidget, QTextEdit, QSplitter ) from PyQt6.QtCore import Qt, QTimer from PyQt6.QtGui import QColor from plugins.base_plugin import BasePlugin from core.event_bus import ( LootEvent, SkillGainEvent, GlobalEvent, DamageEvent, EventCategory, get_event_bus ) @dataclass class SessionEntry: """Single session entry representing an event.""" timestamp: datetime event_type: str description: str value: float = 0.0 details: Dict[str, Any] = field(default_factory=dict) @dataclass class SessionSummary: """Summary statistics for a session.""" start_time: datetime end_time: Optional[datetime] = None total_loot_tt: float = 0.0 total_loot_count: int = 0 globals_count: int = 0 hofs_count: int = 0 skill_gains: int = 0 total_damage_dealt: float = 0.0 total_damage_taken: float = 0.0 unique_items: List[str] = field(default_factory=list) class SessionExporterPlugin(BasePlugin): """ Plugin for exporting hunting/mining sessions to various formats. Features: - Automatic session tracking from event bus - Export to CSV or JSON - Session statistics and summaries - Configurable auto-export """ name = "Session Exporter" version = "1.0.0" author = "EU-Utility" description = "Export hunting/mining sessions to CSV/JSON" icon = "📊" hotkey = "ctrl+shift+e" def __init__(self, overlay_window, config): super().__init__(overlay_window, config) # Session state self.session_active = False self.session_start_time: Optional[datetime] = None self.session_entries: List[SessionEntry] = [] self.session_summary = None # Event subscriptions self._subscriptions: List[str] = [] # Auto-export settings self.auto_export_enabled = self.get_config('auto_export', False) self.auto_export_interval = self.get_config('auto_export_interval', 300) # 5 minutes self.auto_export_format = self.get_config('auto_export_format', 'json') # Export settings self.export_directory = self.get_config('export_directory', str(Path.home() / "Documents" / "EU-Sessions")) Path(self.export_directory).mkdir(parents=True, exist_ok=True) # UI references self._ui = None self.session_table = None self.status_label = None self.stats_label = None # Auto-export timer self._export_timer = None def initialize(self) -> None: """Initialize plugin and subscribe to events.""" self.log_info("Initializing Session Exporter") # Subscribe to events self._subscriptions.append( self.subscribe_typed(LootEvent, self._on_loot) ) self._subscriptions.append( self.subscribe_typed(SkillGainEvent, self._on_skill_gain) ) self._subscriptions.append( self.subscribe_typed(GlobalEvent, self._on_global) ) self._subscriptions.append( self.subscribe_typed(DamageEvent, self._on_damage) ) # Start session automatically self.start_session() # Setup auto-export timer if enabled if self.auto_export_enabled: self._setup_auto_export() self.log_info("Session Exporter initialized") def get_ui(self) -> QWidget: """Return the plugin's UI widget.""" if self._ui is None: self._ui = self._create_ui() return self._ui def _create_ui(self) -> QWidget: """Create the plugin UI.""" widget = QWidget() layout = QVBoxLayout(widget) layout.setSpacing(12) layout.setContentsMargins(16, 16, 16, 16) # Header header = QLabel("📊 Session Exporter") header.setStyleSheet(""" font-size: 18px; font-weight: bold; color: white; padding-bottom: 8px; """) layout.addWidget(header) # Status section status_group = QGroupBox("Session Status") status_layout = QVBoxLayout(status_group) self.status_label = QLabel("Session: Active") self.status_label.setStyleSheet("color: #4caf50; font-weight: bold;") status_layout.addWidget(self.status_label) self.stats_label = QLabel(self._get_stats_text()) self.stats_label.setStyleSheet("color: rgba(255, 255, 255, 150);") status_layout.addWidget(self.stats_label) layout.addWidget(status_group) # Control buttons button_layout = QHBoxLayout() self.start_btn = QPushButton("â–ļī¸ Start New") self.start_btn.setStyleSheet(self._button_style()) self.start_btn.clicked.connect(self.start_session) button_layout.addWidget(self.start_btn) self.stop_btn = QPushButton("⏚ī¸ Stop") self.stop_btn.setStyleSheet(self._button_style()) self.stop_btn.clicked.connect(self.stop_session) button_layout.addWidget(self.stop_btn) self.export_csv_btn = QPushButton("📄 Export CSV") self.export_csv_btn.setStyleSheet(self._button_style("#2196f3")) self.export_csv_btn.clicked.connect(lambda: self.export_session('csv')) button_layout.addWidget(self.export_csv_btn) self.export_json_btn = QPushButton("📄 Export JSON") self.export_json_btn.setStyleSheet(self._button_style("#2196f3")) self.export_json_btn.clicked.connect(lambda: self.export_session('json')) button_layout.addWidget(self.export_json_btn) layout.addLayout(button_layout) # Session entries table table_group = QGroupBox("Session Entries") table_layout = QVBoxLayout(table_group) self.session_table = QTableWidget() self.session_table.setColumnCount(4) self.session_table.setHorizontalHeaderLabels(["Time", "Type", "Description", "Value"]) self.session_table.horizontalHeader().setStretchLastSection(True) self.session_table.horizontalHeader().setSectionResizeMode(QHeaderView.ResizeMode.Stretch) self.session_table.setStyleSheet(""" QTableWidget { background-color: rgba(30, 35, 45, 150); border: 1px solid rgba(100, 150, 200, 50); border-radius: 8px; color: white; } QHeaderView::section { background-color: rgba(50, 60, 75, 200); color: white; padding: 8px; border: none; } QTableWidget::item { padding: 6px; } """) table_layout.addWidget(self.session_table) layout.addWidget(table_group) # Settings settings_group = QGroupBox("Settings") settings_layout = QVBoxLayout(settings_group) # Auto-export auto_export_layout = QHBoxLayout() self.auto_export_checkbox = QCheckBox("Auto-export every") self.auto_export_checkbox.setChecked(self.auto_export_enabled) self.auto_export_checkbox.setStyleSheet("color: white;") auto_export_layout.addWidget(self.auto_export_checkbox) self.auto_export_spin = QSpinBox() self.auto_export_spin.setRange(1, 60) self.auto_export_spin.setValue(self.auto_export_interval // 60) self.auto_export_spin.setSuffix(" min") self.auto_export_spin.setStyleSheet(""" QSpinBox { background-color: rgba(50, 60, 75, 200); color: white; border: 1px solid rgba(100, 150, 200, 100); border-radius: 4px; padding: 4px; } """) auto_export_layout.addWidget(self.auto_export_spin) auto_export_layout.addStretch() settings_layout.addLayout(auto_export_layout) # Export directory dir_layout = QHBoxLayout() dir_label = QLabel("Export Directory:") dir_label.setStyleSheet("color: rgba(255, 255, 255, 150);") dir_layout.addWidget(dir_label) self.dir_display = QLabel(self.export_directory) self.dir_display.setStyleSheet("color: white;") self.dir_display.setWordWrap(True) dir_layout.addWidget(self.dir_display, 1) change_dir_btn = QPushButton("📁 Change") change_dir_btn.setStyleSheet(self._button_style()) change_dir_btn.clicked.connect(self._change_export_directory) dir_layout.addWidget(change_dir_btn) settings_layout.addLayout(dir_layout) layout.addWidget(settings_group) # Apply settings button apply_btn = QPushButton("💾 Apply Settings") apply_btn.setStyleSheet(self._button_style("#4caf50")) apply_btn.clicked.connect(self._apply_settings) layout.addWidget(apply_btn) layout.addStretch() return widget def _button_style(self, color: str = "#607d8b") -> str: """Generate button stylesheet.""" return f""" QPushButton {{ background-color: {color}; color: white; border: none; border-radius: 8px; padding: 10px 16px; font-weight: bold; }} QPushButton:hover {{ background-color: {color}dd; }} QPushButton:pressed {{ background-color: {color}aa; }} """ def start_session(self) -> None: """Start a new tracking session.""" self.session_active = True self.session_start_time = datetime.now() self.session_entries = [] self.session_summary = SessionSummary(start_time=self.session_start_time) self.log_info(f"Session started at {self.session_start_time}") if self.status_label: self.status_label.setText(f"Session: Active (started {self.session_start_time.strftime('%H:%M:%S')})") self.status_label.setStyleSheet("color: #4caf50; font-weight: bold;") self.notify("Session Started", "Tracking loot, skills, and globals") def stop_session(self) -> None: """Stop the current tracking session.""" if not self.session_active: return self.session_active = False if self.session_summary: self.session_summary.end_time = datetime.now() self.log_info("Session stopped") if self.status_label: duration = "" if self.session_summary and self.session_summary.end_time: duration_secs = (self.session_summary.end_time - self.session_start_time).total_seconds() mins, secs = divmod(int(duration_secs), 60) hours, mins = divmod(mins, 60) duration = f"Duration: {hours:02d}:{mins:02d}:{secs:02d}" self.status_label.setText(f"Session: Stopped ({duration})") self.status_label.setStyleSheet("color: #ff9800; font-weight: bold;") self.notify("Session Stopped", f"Total entries: {len(self.session_entries)}") def _on_loot(self, event: LootEvent) -> None: """Handle loot events.""" if not self.session_active: return item_names = ", ".join(event.get_item_names()) entry = SessionEntry( timestamp=event.timestamp, event_type="Loot", description=f"From {event.mob_name}: {item_names}", value=event.total_tt_value, details={ 'mob_name': event.mob_name, 'items': event.items, 'position': event.position } ) self.session_entries.append(entry) if self.session_summary: self.session_summary.total_loot_tt += event.total_tt_value self.session_summary.total_loot_count += 1 for item in event.get_item_names(): if item not in self.session_summary.unique_items: self.session_summary.unique_items.append(item) self._update_ui() def _on_skill_gain(self, event: SkillGainEvent) -> None: """Handle skill gain events.""" if not self.session_active: return entry = SessionEntry( timestamp=event.timestamp, event_type="Skill", description=f"{event.skill_name} +{event.gain_amount:.4f}", value=event.gain_amount, details={ 'skill_name': event.skill_name, 'skill_value': event.skill_value, 'gain_amount': event.gain_amount } ) self.session_entries.append(entry) if self.session_summary: self.session_summary.skill_gains += 1 self._update_ui() def _on_global(self, event: GlobalEvent) -> None: """Handle global/HOF events.""" if not self.session_active: return is_hof = event.achievement_type.lower() in ['hof', 'ath', 'discovery'] entry = SessionEntry( timestamp=event.timestamp, event_type="HOF" if is_hof else "Global", description=f"{event.player_name}: {event.item_name or 'Value'} worth {event.value:.0f} PED", value=event.value, details={ 'player_name': event.player_name, 'achievement_type': event.achievement_type, 'item_name': event.item_name } ) self.session_entries.append(entry) if self.session_summary: if is_hof: self.session_summary.hofs_count += 1 else: self.session_summary.globals_count += 1 self._update_ui() def _on_damage(self, event: DamageEvent) -> None: """Handle damage events.""" if not self.session_active: return if event.is_outgoing: if self.session_summary: self.session_summary.total_damage_dealt += event.damage_amount else: if self.session_summary: self.session_summary.total_damage_taken += event.damage_amount self._update_ui() def _update_ui(self) -> None: """Update the UI with current session data.""" if not self._ui or not self.session_table: return # Update stats label if self.stats_label: self.stats_label.setText(self._get_stats_text()) # Update table (show last 50 entries) recent_entries = self.session_entries[-50:] self.session_table.setRowCount(len(recent_entries)) type_colors = { 'Loot': '#4caf50', 'Skill': '#2196f3', 'Global': '#ff9800', 'HOF': '#f44336' } for row, entry in enumerate(recent_entries): # Time time_item = QTableWidgetItem(entry.timestamp.strftime("%H:%M:%S")) time_item.setForeground(QColor("white")) self.session_table.setItem(row, 0, time_item) # Type type_item = QTableWidgetItem(entry.event_type) type_item.setForeground(QColor(type_colors.get(entry.event_type, 'white'))) self.session_table.setItem(row, 1, type_item) # Description desc_item = QTableWidgetItem(entry.description) desc_item.setForeground(QColor("white")) self.session_table.setItem(row, 2, desc_item) # Value value_item = QTableWidgetItem(f"{entry.value:.2f}") value_item.setForeground(QColor("#ffc107")) self.session_table.setItem(row, 3, value_item) # Scroll to bottom self.session_table.scrollToBottom() def _get_stats_text(self) -> str: """Get formatted statistics text.""" if not self.session_summary: return "No active session" lines = [ f"Entries: {len(self.session_entries)}", f"Loot: {self.session_summary.total_loot_count} items, {self.session_summary.total_loot_tt:.2f} PED TT", f"Globals: {self.session_summary.globals_count} | HOFs: {self.session_summary.hofs_count}", f"Skills: {self.session_summary.skill_gains}", ] if self.session_summary.total_damage_dealt > 0: lines.append(f"Damage Dealt: {self.session_summary.total_damage_dealt:,.0f}") return " | ".join(lines) def export_session(self, format_type: str = 'json') -> Optional[Path]: """ Export the current session to a file. Args: format_type: 'json' or 'csv' Returns: Path to exported file or None if failed """ if not self.session_entries: self.notify_warning("Export Failed", "No session data to export") return None timestamp = datetime.now().strftime("%Y%m%d_%H%M%S") filename = f"session_{timestamp}.{format_type}" filepath = Path(self.export_directory) / filename try: if format_type == 'json': self._export_json(filepath) elif format_type == 'csv': self._export_csv(filepath) else: raise ValueError(f"Unknown format: {format_type}") self.notify_success("Export Complete", f"Saved to {filename}") self.log_info(f"Session exported to {filepath}") return filepath except Exception as e: self.notify_error("Export Failed", str(e)) self.log_error(f"Export failed: {e}") return None def _export_json(self, filepath: Path) -> None: """Export session to JSON format.""" data = { 'export_info': { 'version': '1.0.0', 'exported_at': datetime.now().isoformat(), 'plugin': 'Session Exporter' }, 'session': { 'start_time': self.session_start_time.isoformat() if self.session_start_time else None, 'end_time': self.session_summary.end_time.isoformat() if self.session_summary and self.session_summary.end_time else None, 'summary': asdict(self.session_summary) if self.session_summary else {}, 'entries': [ { 'timestamp': e.timestamp.isoformat(), 'event_type': e.event_type, 'description': e.description, 'value': e.value, 'details': e.details } for e in self.session_entries ] } } with open(filepath, 'w', encoding='utf-8') as f: json.dump(data, f, indent=2, ensure_ascii=False) def _export_csv(self, filepath: Path) -> None: """Export session to CSV format.""" with open(filepath, 'w', newline='', encoding='utf-8') as f: writer = csv.writer(f) # Write header writer.writerow(['timestamp', 'event_type', 'description', 'value', 'details']) # Write entries for entry in self.session_entries: writer.writerow([ entry.timestamp.isoformat(), entry.event_type, entry.description, entry.value, json.dumps(entry.details) ]) def _change_export_directory(self) -> None: """Change the export directory.""" new_dir = QFileDialog.getExistingDirectory( self._ui, "Select Export Directory", self.export_directory ) if new_dir: self.export_directory = new_dir if self.dir_display: self.dir_display.setText(new_dir) def _apply_settings(self) -> None: """Apply and save settings.""" self.auto_export_enabled = self.auto_export_checkbox.isChecked() self.auto_export_interval = self.auto_export_spin.value() * 60 self.set_config('auto_export', self.auto_export_enabled) self.set_config('auto_export_interval', self.auto_export_interval) self.set_config('export_directory', self.export_directory) # Update auto-export timer if self.auto_export_enabled: self._setup_auto_export() elif self._export_timer: self._export_timer.stop() self.notify_success("Settings Saved", "Session exporter settings updated") def _setup_auto_export(self) -> None: """Setup auto-export timer.""" if self._export_timer: self._export_timer.stop() self._export_timer = QTimer() self._export_timer.timeout.connect(lambda: self.export_session(self.auto_export_format)) self._export_timer.start(self.auto_export_interval * 1000) self.log_info(f"Auto-export enabled every {self.auto_export_interval // 60} minutes") def on_hotkey(self) -> None: """Handle hotkey press.""" if self.session_active: self.stop_session() else: self.start_session() def shutdown(self) -> None: """Cleanup on shutdown.""" # Auto-export on shutdown if enabled if self.auto_export_enabled and self.session_entries: self.export_session(self.auto_export_format) # Stop session if self.session_active: self.stop_session() # Unsubscribe from events for sub_id in self._subscriptions: self.unsubscribe_typed(sub_id) if self._export_timer: self._export_timer.stop() super().shutdown()