EU-Utility/plugins/price_alerts/plugin.py

694 lines
25 KiB
Python

"""
EU-Utility - Price Alert Plugin
Monitor Entropia Nexus API prices and get alerts when items
reach target prices (good for buying low/selling high).
"""
import json
from datetime import datetime, timedelta
from typing import Dict, List, Optional, Any
from dataclasses import dataclass, field, asdict
from PyQt6.QtWidgets import (
QWidget, QVBoxLayout, QHBoxLayout, QLabel, QPushButton,
QTableWidget, QTableWidgetItem, QHeaderView, QLineEdit,
QDoubleSpinBox, QComboBox, QMessageBox, QGroupBox,
QCheckBox, QSpinBox, QSplitter, QTextEdit
)
from PyQt6.QtCore import Qt, QTimer, pyqtSignal, QObject
from PyQt6.QtGui import QColor
from plugins.base_plugin import BasePlugin
from core.nexus_api import get_nexus_api, MarketData
@dataclass
class PriceAlert:
"""A price alert configuration."""
id: str
item_id: str
item_name: str
alert_type: str # 'below' or 'above'
target_price: float
current_price: Optional[float] = None
last_checked: Optional[datetime] = None
triggered: bool = False
trigger_count: int = 0
enabled: bool = True
created_at: datetime = field(default_factory=datetime.now)
def to_dict(self) -> Dict[str, Any]:
"""Convert to dictionary for serialization."""
return {
'id': self.id,
'item_id': self.item_id,
'item_name': self.item_name,
'alert_type': self.alert_type,
'target_price': self.target_price,
'current_price': self.current_price,
'last_checked': self.last_checked.isoformat() if self.last_checked else None,
'triggered': self.triggered,
'trigger_count': self.trigger_count,
'enabled': self.enabled,
'created_at': self.created_at.isoformat()
}
@classmethod
def from_dict(cls, data: Dict[str, Any]) -> 'PriceAlert':
"""Create from dictionary."""
alert = cls(
id=data['id'],
item_id=data['item_id'],
item_name=data['item_name'],
alert_type=data['alert_type'],
target_price=data['target_price'],
current_price=data.get('current_price'),
triggered=data.get('triggered', False),
trigger_count=data.get('trigger_count', 0),
enabled=data.get('enabled', True)
)
if data.get('last_checked'):
alert.last_checked = datetime.fromisoformat(data['last_checked'])
if data.get('created_at'):
alert.created_at = datetime.fromisoformat(data['created_at'])
return alert
def check_condition(self, current_price: float) -> bool:
"""Check if the alert condition is met."""
self.current_price = current_price
self.last_checked = datetime.now()
if self.alert_type == 'below':
return current_price <= self.target_price
elif self.alert_type == 'above':
return current_price >= self.target_price
return False
class PriceAlertSignals(QObject):
"""Signals for thread-safe UI updates."""
alert_triggered = pyqtSignal(object) # PriceAlert
prices_updated = pyqtSignal()
class PriceAlertPlugin(BasePlugin):
"""
Plugin for monitoring Entropia Nexus prices.
Features:
- Monitor any item's market price
- Set alerts for price thresholds (buy low, sell high)
- Auto-refresh at configurable intervals
- Notification when alerts trigger
- Price history tracking
"""
name = "Price Alerts"
version = "1.0.0"
author = "EU-Utility"
description = "Monitor Nexus prices and get alerts"
icon = "🔔"
hotkey = "ctrl+shift+p"
def __init__(self, overlay_window, config):
super().__init__(overlay_window, config)
# Alert storage
self.alerts: Dict[str, PriceAlert] = {}
self.price_history: Dict[str, List[Dict]] = {} # item_id -> price history
# Services
self.nexus = get_nexus_api()
self.signals = PriceAlertSignals()
# Settings
self.check_interval = self.get_config('check_interval', 300) # 5 minutes
self.notifications_enabled = self.get_config('notifications_enabled', True)
self.sound_enabled = self.get_config('sound_enabled', True)
self.history_days = self.get_config('history_days', 7)
# UI references
self._ui = None
self.alerts_table = None
self.search_results_table = None
self.search_input = None
self.status_label = None
# Timer
self._check_timer = None
# Connect signals
self.signals.alert_triggered.connect(self._on_alert_triggered)
self.signals.prices_updated.connect(self._update_alerts_table)
# Load saved alerts
self._load_alerts()
def initialize(self) -> None:
"""Initialize plugin."""
self.log_info("Initializing Price Alerts")
# Start price check timer
self._check_timer = QTimer()
self._check_timer.timeout.connect(self._check_all_prices)
self._check_timer.start(self.check_interval * 1000)
# Initial price check
self._check_all_prices()
self.log_info(f"Price Alerts initialized with {len(self.alerts)} alerts")
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("🔔 Price Alerts")
header.setStyleSheet("""
font-size: 18px;
font-weight: bold;
color: white;
padding-bottom: 8px;
""")
layout.addWidget(header)
# Status
self.status_label = QLabel(f"Active Alerts: {len(self.alerts)} | Next check: --")
self.status_label.setStyleSheet("color: rgba(255, 255, 255, 150);")
layout.addWidget(self.status_label)
# Create tabs
tabs = QSplitter(Qt.Orientation.Vertical)
# === Alerts Tab ===
alerts_widget = QWidget()
alerts_layout = QVBoxLayout(alerts_widget)
alerts_layout.setContentsMargins(0, 0, 0, 0)
alerts_header = QLabel("Your Alerts")
alerts_header.setStyleSheet("font-weight: bold; color: white;")
alerts_layout.addWidget(alerts_header)
self.alerts_table = QTableWidget()
self.alerts_table.setColumnCount(6)
self.alerts_table.setHorizontalHeaderLabels([
"Item", "Condition", "Target", "Current", "Status", "Actions"
])
self.alerts_table.horizontalHeader().setStretchLastSection(True)
self.alerts_table.horizontalHeader().setSectionResizeMode(QHeaderView.ResizeMode.Stretch)
self.alerts_table.setStyleSheet(self._table_style())
alerts_layout.addWidget(self.alerts_table)
# Alert buttons
alerts_btn_layout = QHBoxLayout()
refresh_btn = QPushButton("🔄 Check Now")
refresh_btn.setStyleSheet(self._button_style("#2196f3"))
refresh_btn.clicked.connect(self._check_all_prices)
alerts_btn_layout.addWidget(refresh_btn)
clear_triggered_btn = QPushButton("🧹 Clear Triggered")
clear_triggered_btn.setStyleSheet(self._button_style())
clear_triggered_btn.clicked.connect(self._clear_triggered)
alerts_btn_layout.addWidget(clear_triggered_btn)
alerts_btn_layout.addStretch()
alerts_layout.addLayout(alerts_btn_layout)
tabs.addWidget(alerts_widget)
# === Search Tab ===
search_widget = QWidget()
search_layout = QVBoxLayout(search_widget)
search_layout.setContentsMargins(0, 0, 0, 0)
search_header = QLabel("Search Items to Monitor")
search_header.setStyleSheet("font-weight: bold; color: white;")
search_layout.addWidget(search_header)
# Search input
search_input_layout = QHBoxLayout()
self.search_input = QLineEdit()
self.search_input.setPlaceholderText("Search for items...")
self.search_input.setStyleSheet(self._input_style())
self.search_input.returnPressed.connect(self._search_items)
search_input_layout.addWidget(self.search_input)
search_btn = QPushButton("🔍 Search")
search_btn.setStyleSheet(self._button_style("#4caf50"))
search_btn.clicked.connect(self._search_items)
search_input_layout.addWidget(search_btn)
search_layout.addLayout(search_input_layout)
# Search results
self.search_results_table = QTableWidget()
self.search_results_table.setColumnCount(4)
self.search_results_table.setHorizontalHeaderLabels(["Name", "Type", "Current Markup", "Action"])
self.search_results_table.horizontalHeader().setStretchLastSection(True)
self.search_results_table.setStyleSheet(self._table_style())
search_layout.addWidget(self.search_results_table)
tabs.addWidget(search_widget)
layout.addWidget(tabs)
# Settings
settings_group = QGroupBox("Settings")
settings_layout = QVBoxLayout(settings_group)
# Check interval
interval_layout = QHBoxLayout()
interval_label = QLabel("Check Interval:")
interval_label.setStyleSheet("color: white;")
interval_layout.addWidget(interval_label)
self.interval_spin = QSpinBox()
self.interval_spin.setRange(1, 60)
self.interval_spin.setValue(self.check_interval // 60)
self.interval_spin.setSuffix(" min")
self.interval_spin.setStyleSheet(self._spinbox_style())
interval_layout.addWidget(self.interval_spin)
interval_layout.addStretch()
settings_layout.addLayout(interval_layout)
# Notification settings
notif_layout = QHBoxLayout()
self.notif_checkbox = QCheckBox("Show Notifications")
self.notif_checkbox.setChecked(self.notifications_enabled)
self.notif_checkbox.setStyleSheet("color: white;")
notif_layout.addWidget(self.notif_checkbox)
self.sound_checkbox = QCheckBox("Play Sound")
self.sound_checkbox.setChecked(self.sound_enabled)
self.sound_checkbox.setStyleSheet("color: white;")
notif_layout.addWidget(self.sound_checkbox)
notif_layout.addStretch()
settings_layout.addLayout(notif_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)
# Initial table population
self._update_alerts_table()
return widget
def _table_style(self) -> str:
"""Generate table stylesheet."""
return """
QTableWidget {
background-color: rgba(30, 35, 45, 150);
border: 1px solid rgba(100, 150, 200, 50);
border-radius: 8px;
color: white;
gridline-color: rgba(100, 150, 200, 30);
}
QHeaderView::section {
background-color: rgba(50, 60, 75, 200);
color: white;
padding: 8px;
border: none;
}
QTableWidget::item {
padding: 6px;
}
QTableWidget::item:selected {
background-color: rgba(100, 150, 200, 100);
}
"""
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: 8px 14px;
font-weight: bold;
}}
QPushButton:hover {{
background-color: {color}dd;
}}
QPushButton:pressed {{
background-color: {color}aa;
}}
"""
def _input_style(self) -> str:
"""Generate input stylesheet."""
return """
QLineEdit {
background-color: rgba(50, 60, 75, 200);
color: white;
border: 1px solid rgba(100, 150, 200, 100);
border-radius: 8px;
padding: 8px 12px;
}
QLineEdit:focus {
border: 1px solid rgba(100, 180, 255, 150);
}
"""
def _spinbox_style(self) -> str:
"""Generate spinbox stylesheet."""
return """
QSpinBox {
background-color: rgba(50, 60, 75, 200);
color: white;
border: 1px solid rgba(100, 150, 200, 100);
border-radius: 4px;
padding: 4px;
}
"""
def _search_items(self) -> None:
"""Search for items via Nexus API."""
query = self.search_input.text().strip()
if not query:
return
self.status_label.setText(f"Searching for '{query}'...")
# Run search in background
self.run_in_background(
self.nexus.search_items,
query,
limit=20,
priority='normal',
on_complete=self._on_search_complete,
on_error=self._on_search_error
)
def _on_search_complete(self, results: List[Any]) -> None:
"""Handle search completion."""
self.search_results_table.setRowCount(len(results))
for row, item in enumerate(results):
# Name
name_item = QTableWidgetItem(item.name)
name_item.setForeground(QColor("white"))
self.search_results_table.setItem(row, 0, name_item)
# Type
type_item = QTableWidgetItem(item.type)
type_item.setForeground(QColor("#2196f3"))
self.search_results_table.setItem(row, 1, type_item)
# Current markup (will be fetched)
markup_item = QTableWidgetItem("Click to check")
markup_item.setForeground(QColor("rgba(255, 255, 255, 100)"))
self.search_results_table.setItem(row, 2, markup_item)
# Action button
add_btn = QPushButton("+ Alert")
add_btn.setStyleSheet(self._button_style("#4caf50"))
add_btn.clicked.connect(lambda checked, i=item: self._show_add_alert_dialog(i))
self.search_results_table.setCellWidget(row, 3, add_btn)
self.status_label.setText(f"Found {len(results)} items")
def _on_search_error(self, error: Exception) -> None:
"""Handle search error."""
self.status_label.setText(f"Search failed: {str(error)}")
self.notify_error("Search Failed", str(error))
def _show_add_alert_dialog(self, item: Any) -> None:
"""Show dialog to add a new alert."""
from PyQt6.QtWidgets import QDialog, QFormLayout, QDialogButtonBox
dialog = QDialog(self._ui)
dialog.setWindowTitle(f"Add Alert: {item.name}")
dialog.setStyleSheet("""
QDialog {
background-color: #2a3040;
}
QLabel {
color: white;
}
""")
layout = QFormLayout(dialog)
# Alert type
type_combo = QComboBox()
type_combo.addItems(["Price Below", "Price Above"])
type_combo.setStyleSheet(self._input_style())
layout.addRow("Alert When:", type_combo)
# Target price
price_spin = QDoubleSpinBox()
price_spin.setRange(0.01, 1000000)
price_spin.setDecimals(2)
price_spin.setValue(100.0)
price_spin.setSuffix(" %")
price_spin.setStyleSheet(self._spinbox_style())
layout.addRow("Target Price:", price_spin)
# Buttons
buttons = QDialogButtonBox(
QDialogButtonBox.StandardButton.Ok | QDialogButtonBox.StandardButton.Cancel
)
buttons.accepted.connect(dialog.accept)
buttons.rejected.connect(dialog.reject)
layout.addRow(buttons)
if dialog.exec() == QDialog.DialogCode.Accepted:
alert_type = 'below' if type_combo.currentIndex() == 0 else 'above'
self._add_alert(item.id, item.name, alert_type, price_spin.value())
def _add_alert(self, item_id: str, item_name: str, alert_type: str, target_price: float) -> None:
"""Add a new price alert."""
import uuid
alert_id = str(uuid.uuid4())[:8]
alert = PriceAlert(
id=alert_id,
item_id=item_id,
item_name=item_name,
alert_type=alert_type,
target_price=target_price
)
self.alerts[alert_id] = alert
self._save_alerts()
self._update_alerts_table()
# Check price immediately
self._check_alert_price(alert)
self.notify_success("Alert Added", f"Monitoring {item_name}")
self.log_info(f"Added price alert for {item_name}: {alert_type} {target_price}%")
def _remove_alert(self, alert_id: str) -> None:
"""Remove an alert."""
if alert_id in self.alerts:
del self.alerts[alert_id]
self._save_alerts()
self._update_alerts_table()
self.log_info(f"Removed alert {alert_id}")
def _check_all_prices(self) -> None:
"""Check prices for all enabled alerts."""
if not self.alerts:
return
self.status_label.setText("Checking prices...")
# Check each alert's price
for alert in self.alerts.values():
if alert.enabled:
self._check_alert_price(alert)
# Update UI
self._update_alerts_table()
# Schedule next check
next_check = datetime.now() + timedelta(seconds=self.check_interval)
self.status_label.setText(f"Active Alerts: {len(self.alerts)} | Next check: {next_check.strftime('%H:%M')}")
def _check_alert_price(self, alert: PriceAlert) -> None:
"""Check price for a single alert."""
try:
market_data = self.nexus.get_market_data(alert.item_id)
if market_data and market_data.current_markup is not None:
current_price = market_data.current_markup
# Update price history
if alert.item_id not in self.price_history:
self.price_history[alert.item_id] = []
self.price_history[alert.item_id].append({
'timestamp': datetime.now().isoformat(),
'price': current_price
})
# Trim history
cutoff = datetime.now() - timedelta(days=self.history_days)
self.price_history[alert.item_id] = [
h for h in self.price_history[alert.item_id]
if datetime.fromisoformat(h['timestamp']) > cutoff
]
# Check condition
if alert.check_condition(current_price):
if not alert.triggered:
alert.triggered = True
alert.trigger_count += 1
self.signals.alert_triggered.emit(alert)
else:
alert.triggered = False
alert.last_checked = datetime.now()
except Exception as e:
self.log_error(f"Error checking price for {alert.item_name}: {e}")
def _on_alert_triggered(self, alert: PriceAlert) -> None:
"""Handle triggered alert."""
condition = "dropped below" if alert.alert_type == 'below' else "rose above"
message = f"{alert.item_name} {condition} {alert.target_price:.1f}% (current: {alert.current_price:.1f}%)"
if self.notifications_enabled:
self.notify("🚨 Price Alert", message, sound=self.sound_enabled)
if self.sound_enabled:
self.play_sound('alert')
self.log_info(f"Price alert triggered: {message}")
self._save_alerts()
def _update_alerts_table(self) -> None:
"""Update the alerts table."""
if not self.alerts_table:
return
enabled_alerts = [a for a in self.alerts.values() if a.enabled]
self.alerts_table.setRowCount(len(enabled_alerts))
for row, alert in enumerate(enabled_alerts):
# Item name
name_item = QTableWidgetItem(alert.item_name)
name_item.setForeground(QColor("white"))
self.alerts_table.setItem(row, 0, name_item)
# Condition
condition_text = f"Alert when {'below' if alert.alert_type == 'below' else 'above'}"
condition_item = QTableWidgetItem(condition_text)
condition_item.setForeground(QColor("#2196f3"))
self.alerts_table.setItem(row, 1, condition_item)
# Target price
target_item = QTableWidgetItem(f"{alert.target_price:.1f}%")
target_item.setForeground(QColor("#ffc107"))
self.alerts_table.setItem(row, 2, target_item)
# Current price
current_text = f"{alert.current_price:.1f}%" if alert.current_price else "--"
current_item = QTableWidgetItem(current_text)
current_item.setForeground(QColor("#4caf50" if alert.current_price else "rgba(255,255,255,100)"))
self.alerts_table.setItem(row, 3, current_item)
# Status
status_text = "🚨 TRIGGERED" if alert.triggered else ("✅ OK" if alert.last_checked else "⏳ Pending")
status_item = QTableWidgetItem(status_text)
status_color = "#f44336" if alert.triggered else ("#4caf50" if alert.last_checked else "rgba(255,255,255,100)")
status_item.setForeground(QColor(status_color))
self.alerts_table.setItem(row, 4, status_item)
# Actions
actions_widget = QWidget()
actions_layout = QHBoxLayout(actions_widget)
actions_layout.setContentsMargins(4, 2, 4, 2)
remove_btn = QPushButton("🗑️")
remove_btn.setFixedSize(28, 28)
remove_btn.setStyleSheet("""
QPushButton {
background-color: #f44336;
color: white;
border: none;
border-radius: 4px;
}
""")
remove_btn.clicked.connect(lambda checked, aid=alert.id: self._remove_alert(aid))
actions_layout.addWidget(remove_btn)
actions_layout.addStretch()
self.alerts_table.setCellWidget(row, 5, actions_widget)
self.status_label.setText(f"Active Alerts: {len(self.alerts)}")
def _clear_triggered(self) -> None:
"""Clear all triggered alerts."""
for alert in self.alerts.values():
alert.triggered = False
self._save_alerts()
self._update_alerts_table()
def _apply_settings(self) -> None:
"""Apply and save settings."""
self.check_interval = self.interval_spin.value() * 60
self.notifications_enabled = self.notif_checkbox.isChecked()
self.sound_enabled = self.sound_checkbox.isChecked()
self.set_config('check_interval', self.check_interval)
self.set_config('notifications_enabled', self.notifications_enabled)
self.set_config('sound_enabled', self.sound_enabled)
# Update timer
if self._check_timer:
self._check_timer.setInterval(self.check_interval * 1000)
self.notify_success("Settings Saved", "Price alert settings updated")
def _save_alerts(self) -> None:
"""Save alerts to storage."""
alerts_data = {aid: alert.to_dict() for aid, alert in self.alerts.items()}
self.save_data('alerts', alerts_data)
self.save_data('price_history', self.price_history)
def _load_alerts(self) -> None:
"""Load alerts from storage."""
alerts_data = self.load_data('alerts', {})
for aid, data in alerts_data.items():
try:
self.alerts[aid] = PriceAlert.from_dict(data)
except Exception as e:
self.log_error(f"Failed to load alert {aid}: {e}")
self.price_history = self.load_data('price_history', {})
def on_hotkey(self) -> None:
"""Handle hotkey press."""
self._check_all_prices()
self.notify("Price Check", f"Checked {len(self.alerts)} items")
def shutdown(self) -> None:
"""Cleanup on shutdown."""
self._save_alerts()
if self._check_timer:
self._check_timer.stop()
super().shutdown()