282 lines
8.6 KiB
Python
282 lines
8.6 KiB
Python
"""Pushover service for sending push notifications.
|
|
|
|
This module provides push notification functionality for the AlpineBits application,
|
|
including error alerts and daily reports via Pushover.
|
|
"""
|
|
|
|
import asyncio
|
|
from datetime import datetime
|
|
from typing import Any
|
|
|
|
from pushover_complete import PushoverAPI
|
|
|
|
from .logging_config import get_logger
|
|
|
|
_LOGGER = get_logger(__name__)
|
|
|
|
|
|
class PushoverConfig:
|
|
"""Configuration for Pushover service.
|
|
|
|
Attributes:
|
|
user_key: Pushover user/group key
|
|
api_token: Pushover application API token
|
|
|
|
"""
|
|
|
|
def __init__(self, config: dict[str, Any]):
|
|
"""Initialize Pushover configuration from config dict.
|
|
|
|
Args:
|
|
config: Pushover configuration dictionary
|
|
|
|
"""
|
|
self.user_key: str | None = config.get("user_key")
|
|
self.api_token: str | None = config.get("api_token")
|
|
|
|
# Validate configuration
|
|
if not self.user_key or not self.api_token:
|
|
msg = "Both user_key and api_token are required for Pushover"
|
|
raise ValueError(msg)
|
|
|
|
|
|
class PushoverService:
|
|
"""Service for sending push notifications via Pushover.
|
|
|
|
This service handles sending notifications through the Pushover API,
|
|
including alerts and daily reports.
|
|
"""
|
|
|
|
def __init__(self, config: PushoverConfig):
|
|
"""Initialize Pushover service.
|
|
|
|
Args:
|
|
config: Pushover configuration
|
|
|
|
"""
|
|
self.config = config
|
|
self.api = PushoverAPI(config.api_token)
|
|
|
|
async def send_notification(
|
|
self,
|
|
title: str,
|
|
message: str,
|
|
priority: int = 0,
|
|
url: str | None = None,
|
|
url_title: str | None = None,
|
|
) -> bool:
|
|
"""Send a push notification via Pushover.
|
|
|
|
Args:
|
|
title: Notification title
|
|
message: Notification message
|
|
priority: Priority level (-2 to 2, default 0)
|
|
url: Optional supplementary URL
|
|
url_title: Optional title for the URL
|
|
|
|
Returns:
|
|
True if notification was sent successfully, False otherwise
|
|
|
|
"""
|
|
try:
|
|
# Send notification in thread pool (API is blocking)
|
|
loop = asyncio.get_event_loop()
|
|
await loop.run_in_executor(
|
|
None,
|
|
self._send_pushover,
|
|
title,
|
|
message,
|
|
priority,
|
|
url,
|
|
url_title,
|
|
)
|
|
|
|
_LOGGER.info("Pushover notification sent successfully: %s", title)
|
|
return True
|
|
|
|
except Exception:
|
|
_LOGGER.exception("Failed to send Pushover notification: %s", title)
|
|
return False
|
|
|
|
def _send_pushover(
|
|
self,
|
|
title: str,
|
|
message: str,
|
|
priority: int,
|
|
url: str | None,
|
|
url_title: str | None,
|
|
) -> None:
|
|
"""Send notification via Pushover (blocking operation).
|
|
|
|
Args:
|
|
title: Notification title
|
|
message: Notification message
|
|
priority: Priority level
|
|
url: Optional URL
|
|
url_title: Optional URL title
|
|
|
|
Raises:
|
|
Exception: If notification sending fails
|
|
|
|
"""
|
|
kwargs = {
|
|
"user": self.config.user_key,
|
|
"title": title,
|
|
"message": message,
|
|
"priority": priority,
|
|
}
|
|
|
|
if url:
|
|
kwargs["url"] = url
|
|
if url_title:
|
|
kwargs["url_title"] = url_title
|
|
|
|
self.api.send_message(**kwargs)
|
|
|
|
async def send_alert(
|
|
self,
|
|
title: str,
|
|
message: str,
|
|
priority: int = 1,
|
|
) -> bool:
|
|
"""Send an alert notification (convenience method).
|
|
|
|
Args:
|
|
title: Alert title
|
|
message: Alert message
|
|
priority: Priority level (default 1 for high priority)
|
|
|
|
Returns:
|
|
True if notification was sent successfully, False otherwise
|
|
|
|
"""
|
|
return await self.send_notification(title, message, priority=priority)
|
|
|
|
async def send_daily_report(
|
|
self,
|
|
stats: dict[str, Any],
|
|
errors: list[dict[str, Any]] | None = None,
|
|
priority: int = 0,
|
|
) -> bool:
|
|
"""Send a daily report notification.
|
|
|
|
Args:
|
|
stats: Dictionary containing statistics to include in report
|
|
errors: Optional list of errors to include
|
|
priority: Priority level (default 0 for normal)
|
|
|
|
Returns:
|
|
True if notification was sent successfully, False otherwise
|
|
|
|
"""
|
|
date_str = datetime.now().strftime("%Y-%m-%d")
|
|
title = f"AlpineBits Daily Report - {date_str}"
|
|
|
|
# Build message body (Pushover has a 1024 character limit)
|
|
message = self._build_daily_report_message(date_str, stats, errors)
|
|
|
|
return await self.send_notification(title, message, priority=priority)
|
|
|
|
def _build_daily_report_message(
|
|
self,
|
|
date_str: str,
|
|
stats: dict[str, Any],
|
|
errors: list[dict[str, Any]] | None,
|
|
) -> str:
|
|
"""Build daily report message for Pushover.
|
|
|
|
Args:
|
|
date_str: Date string for the report
|
|
stats: Statistics dictionary
|
|
errors: Optional list of errors
|
|
|
|
Returns:
|
|
Formatted message string (max 1024 chars for Pushover)
|
|
|
|
"""
|
|
lines = [f"Report for {date_str}", ""]
|
|
|
|
# Add statistics (simplified for push notification)
|
|
if stats:
|
|
# Handle reporting period
|
|
period = stats.get("reporting_period", {})
|
|
if period:
|
|
start = period.get("start", "")
|
|
end = period.get("end", "")
|
|
if start and end:
|
|
# Parse the datetime strings to check if they're on different days
|
|
if " " in start and " " in end:
|
|
start_date, start_time = start.split(" ")
|
|
end_date, end_time = end.split(" ")
|
|
|
|
# If same day, just show times
|
|
if start_date == end_date:
|
|
lines.append(f"Period: {start_time} - {end_time}")
|
|
else:
|
|
# Different days, show date + time in compact format
|
|
# Format: "MM-DD HH:MM - MM-DD HH:MM"
|
|
start_compact = f"{start_date[5:]} {start_time[:5]}"
|
|
end_compact = f"{end_date[5:]} {end_time[:5]}"
|
|
lines.append(f"Period: {start_compact} - {end_compact}")
|
|
else:
|
|
# Fallback if format is unexpected
|
|
lines.append(f"Period: {start} - {end}")
|
|
|
|
# Total reservations
|
|
total = stats.get("total_reservations", 0)
|
|
lines.append(f"Total Reservations: {total}")
|
|
|
|
# Per-hotel breakdown (top 5 only to save space)
|
|
hotels = stats.get("hotels", [])
|
|
if hotels:
|
|
lines.append("")
|
|
lines.append("By Hotel:")
|
|
for hotel in hotels[:5]: # Top 5 hotels
|
|
hotel_name = hotel.get("hotel_name", "Unknown")
|
|
count = hotel.get("reservations", 0)
|
|
# Truncate long hotel names
|
|
if len(hotel_name) > 20:
|
|
hotel_name = hotel_name[:17] + "..."
|
|
lines.append(f" • {hotel_name}: {count}")
|
|
|
|
if len(hotels) > 5:
|
|
lines.append(f" • ... and {len(hotels) - 5} more")
|
|
|
|
# Add error summary if present
|
|
if errors:
|
|
lines.append("")
|
|
lines.append(f"Errors: {len(errors)} (see logs)")
|
|
|
|
message = "\n".join(lines)
|
|
|
|
# Truncate if too long (Pushover limit is 1024 chars)
|
|
if len(message) > 1020:
|
|
message = message[:1017] + "..."
|
|
|
|
return message
|
|
|
|
|
|
def create_pushover_service(config: dict[str, Any]) -> PushoverService | None:
|
|
"""Create a Pushover service from configuration.
|
|
|
|
Args:
|
|
config: Full application configuration dictionary
|
|
|
|
Returns:
|
|
PushoverService instance if Pushover is configured, None otherwise
|
|
|
|
"""
|
|
pushover_config = config.get("pushover")
|
|
if not pushover_config:
|
|
_LOGGER.info("Pushover not configured, push notification service disabled")
|
|
return None
|
|
|
|
try:
|
|
pushover_cfg = PushoverConfig(pushover_config)
|
|
service = PushoverService(pushover_cfg)
|
|
_LOGGER.info("Pushover service initialized successfully")
|
|
return service
|
|
except Exception:
|
|
_LOGGER.exception("Failed to initialize Pushover service")
|
|
return None
|