644 lines
23 KiB
Python
644 lines
23 KiB
Python
"""
|
|
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()
|