mirror of
https://github.com/maziggy/bambuddy.git
synced 2026-10-01 19:51:16 +02:00
maziggy
1555fad539
Pushover rejects priority-2 (Emergency) messages unless they carry retry and expire. _send_pushover never sent them, so setting priority 2 always failed with Pushover's "retry and expire are required" error. Now at priority 2 we send retry/expire (default 60s/3600s, clamped to Pushover's 30-10800s range), surfaced as two provider fields shown only when priority is 2. Added PushoverConfig schema fields, i18n labels across all locales, and unit tests.
290 lines
11 KiB
Python
290 lines
11 KiB
Python
"""Pydantic schemas for notification providers."""
|
|
|
|
from datetime import datetime
|
|
from typing import Any
|
|
|
|
from pydantic import BaseModel, Field, field_validator
|
|
|
|
from backend.app.core.compat import StrEnum
|
|
|
|
|
|
class ProviderType(StrEnum):
|
|
"""Supported notification provider types."""
|
|
|
|
CALLMEBOT = "callmebot"
|
|
NTFY = "ntfy"
|
|
PUSHOVER = "pushover"
|
|
TELEGRAM = "telegram"
|
|
EMAIL = "email"
|
|
DISCORD = "discord"
|
|
WEBHOOK = "webhook"
|
|
HOMEASSISTANT = "homeassistant"
|
|
|
|
|
|
class NotificationProviderBase(BaseModel):
|
|
"""Base schema for notification providers."""
|
|
|
|
name: str = Field(..., min_length=1, max_length=100, description="User-defined name")
|
|
provider_type: ProviderType = Field(..., description="Type of notification provider")
|
|
enabled: bool = Field(default=True, description="Whether notifications are enabled")
|
|
config: dict[str, Any] = Field(..., description="Provider-specific configuration")
|
|
|
|
# Event triggers - print lifecycle
|
|
on_print_start: bool = Field(default=False, description="Notify on print start")
|
|
on_print_complete: bool = Field(default=True, description="Notify on print complete")
|
|
on_print_failed: bool = Field(default=True, description="Notify on print failed")
|
|
on_print_stopped: bool = Field(default=True, description="Notify when print is stopped/cancelled")
|
|
on_print_progress: bool = Field(default=False, description="Notify at 25%, 50%, 75% progress")
|
|
on_print_missing_spool_assignment: bool = Field(
|
|
default=False,
|
|
description="Notify when a print starts with required trays missing spool assignments",
|
|
)
|
|
|
|
# Event triggers - printer status
|
|
on_printer_offline: bool = Field(default=False, description="Notify when printer goes offline")
|
|
on_printer_error: bool = Field(default=False, description="Notify on printer errors (AMS, etc.)")
|
|
on_ai_failure_detection: bool = Field(
|
|
default=False,
|
|
description="Notify when Obico AI detects a possible print failure (spaghetti)",
|
|
)
|
|
on_filament_low: bool = Field(default=False, description="Notify when filament is running low")
|
|
on_maintenance_due: bool = Field(default=False, description="Notify when maintenance is due")
|
|
|
|
# Event triggers - AMS environmental alarms (regular AMS)
|
|
on_ams_humidity_high: bool = Field(default=False, description="Notify when AMS humidity exceeds threshold")
|
|
on_ams_temperature_high: bool = Field(default=False, description="Notify when AMS temperature exceeds threshold")
|
|
|
|
# Event triggers - AMS-HT environmental alarms
|
|
on_ams_ht_humidity_high: bool = Field(default=False, description="Notify when AMS-HT humidity exceeds threshold")
|
|
on_ams_ht_temperature_high: bool = Field(
|
|
default=False, description="Notify when AMS-HT temperature exceeds threshold"
|
|
)
|
|
|
|
# Event triggers - Build plate detection
|
|
on_plate_not_empty: bool = Field(default=True, description="Notify when objects detected on plate before print")
|
|
|
|
# Event triggers - Bed cooled
|
|
on_bed_cooled: bool = Field(default=False, description="Notify when bed cools after print")
|
|
|
|
# Event triggers - First layer complete
|
|
on_first_layer_complete: bool = Field(default=False, description="Notify when first layer completes")
|
|
|
|
# Event triggers - Print queue
|
|
on_queue_job_added: bool = Field(default=False, description="Notify when job is added to queue")
|
|
on_queue_job_assigned: bool = Field(default=False, description="Notify when model-based job is assigned to printer")
|
|
on_queue_job_started: bool = Field(default=False, description="Notify when queue job starts printing")
|
|
on_queue_job_waiting: bool = Field(default=True, description="Notify when job is waiting for filament or printer")
|
|
on_queue_job_skipped: bool = Field(default=True, description="Notify when job is skipped")
|
|
on_queue_job_failed: bool = Field(default=True, description="Notify when job fails to start")
|
|
on_queue_completed: bool = Field(default=False, description="Notify when all queue jobs finish")
|
|
|
|
# Quiet hours
|
|
quiet_hours_enabled: bool = Field(default=False, description="Enable quiet hours")
|
|
quiet_hours_start: str | None = Field(default=None, description="Start time in HH:MM format")
|
|
quiet_hours_end: str | None = Field(default=None, description="End time in HH:MM format")
|
|
|
|
# Daily digest
|
|
daily_digest_enabled: bool = Field(default=False, description="Batch notifications into daily digest")
|
|
daily_digest_time: str | None = Field(default=None, description="Time to send digest in HH:MM format")
|
|
|
|
# Printer filter
|
|
printer_id: int | None = Field(default=None, description="Specific printer ID or null for all")
|
|
|
|
@field_validator("quiet_hours_start", "quiet_hours_end", "daily_digest_time")
|
|
@classmethod
|
|
def validate_time_format(cls, v: str | None) -> str | None:
|
|
if v is None:
|
|
return v
|
|
try:
|
|
parts = v.split(":")
|
|
if len(parts) != 2:
|
|
raise ValueError("Invalid time format")
|
|
hour, minute = int(parts[0]), int(parts[1])
|
|
if not (0 <= hour <= 23 and 0 <= minute <= 59):
|
|
raise ValueError("Invalid time range")
|
|
return f"{hour:02d}:{minute:02d}"
|
|
except (ValueError, TypeError):
|
|
raise ValueError("Time must be in HH:MM format (e.g., 22:00)")
|
|
|
|
|
|
class NotificationProviderCreate(NotificationProviderBase):
|
|
"""Schema for creating a notification provider."""
|
|
|
|
pass
|
|
|
|
|
|
class NotificationProviderUpdate(BaseModel):
|
|
"""Schema for updating a notification provider (all fields optional)."""
|
|
|
|
name: str | None = Field(default=None, min_length=1, max_length=100)
|
|
provider_type: ProviderType | None = None
|
|
enabled: bool | None = None
|
|
config: dict[str, Any] | None = None
|
|
|
|
# Event triggers - print lifecycle
|
|
on_print_start: bool | None = None
|
|
on_print_complete: bool | None = None
|
|
on_print_failed: bool | None = None
|
|
on_print_stopped: bool | None = None
|
|
on_print_progress: bool | None = None
|
|
on_print_missing_spool_assignment: bool | None = None
|
|
|
|
# Event triggers - printer status
|
|
on_printer_offline: bool | None = None
|
|
on_printer_error: bool | None = None
|
|
on_ai_failure_detection: bool | None = None
|
|
on_filament_low: bool | None = None
|
|
on_maintenance_due: bool | None = None
|
|
|
|
# Event triggers - AMS environmental alarms (regular AMS)
|
|
on_ams_humidity_high: bool | None = None
|
|
on_ams_temperature_high: bool | None = None
|
|
|
|
# Event triggers - AMS-HT environmental alarms
|
|
on_ams_ht_humidity_high: bool | None = None
|
|
on_ams_ht_temperature_high: bool | None = None
|
|
|
|
# Event triggers - Build plate detection
|
|
on_plate_not_empty: bool | None = None
|
|
|
|
# Event triggers - Bed cooled
|
|
on_bed_cooled: bool | None = None
|
|
|
|
# Event triggers - First layer complete
|
|
on_first_layer_complete: bool | None = None
|
|
|
|
# Event triggers - Print queue
|
|
on_queue_job_added: bool | None = None
|
|
on_queue_job_assigned: bool | None = None
|
|
on_queue_job_started: bool | None = None
|
|
on_queue_job_waiting: bool | None = None
|
|
on_queue_job_skipped: bool | None = None
|
|
on_queue_job_failed: bool | None = None
|
|
on_queue_completed: bool | None = None
|
|
|
|
# Quiet hours
|
|
quiet_hours_enabled: bool | None = None
|
|
quiet_hours_start: str | None = None
|
|
quiet_hours_end: str | None = None
|
|
|
|
# Daily digest
|
|
daily_digest_enabled: bool | None = None
|
|
daily_digest_time: str | None = None
|
|
|
|
# Printer filter
|
|
printer_id: int | None = None
|
|
|
|
|
|
class NotificationProviderResponse(NotificationProviderBase):
|
|
"""Schema for notification provider API responses."""
|
|
|
|
id: int
|
|
last_success: datetime | None = None
|
|
last_error: str | None = None
|
|
last_error_at: datetime | None = None
|
|
created_at: datetime
|
|
updated_at: datetime
|
|
|
|
class Config:
|
|
from_attributes = True
|
|
|
|
|
|
class NotificationTestRequest(BaseModel):
|
|
"""Schema for testing notification configuration."""
|
|
|
|
provider_type: ProviderType
|
|
config: dict[str, Any]
|
|
|
|
|
|
class NotificationTestResponse(BaseModel):
|
|
"""Schema for test notification response."""
|
|
|
|
success: bool
|
|
message: str
|
|
|
|
|
|
# Provider-specific config schemas for documentation/validation reference
|
|
class CallMeBotConfig(BaseModel):
|
|
"""CallMeBot/WhatsApp configuration."""
|
|
|
|
phone: str = Field(..., description="Phone number with country code (e.g., +1234567890)")
|
|
apikey: str = Field(..., description="API key from CallMeBot")
|
|
|
|
|
|
class NtfyConfig(BaseModel):
|
|
"""ntfy configuration."""
|
|
|
|
server: str = Field(default="https://ntfy.sh", description="ntfy server URL")
|
|
topic: str = Field(..., description="Topic name to publish to")
|
|
auth_token: str | None = Field(default=None, description="Optional authentication token")
|
|
event_priorities: dict[str, int] | None = Field(
|
|
default=None,
|
|
description=(
|
|
"Per-event priority override. Keys are event names (e.g. 'on_print_failed'); "
|
|
"values are ntfy priorities 1-5 (1=min, 2=low, 3=default, 4=high, 5=urgent). "
|
|
"Events without an entry use ntfy's server-side default."
|
|
),
|
|
)
|
|
|
|
|
|
class PushoverConfig(BaseModel):
|
|
"""Pushover configuration."""
|
|
|
|
user_key: str = Field(..., description="Your Pushover user key")
|
|
app_token: str = Field(..., description="Your Pushover application token")
|
|
priority: int = Field(default=0, ge=-2, le=2, description="Message priority (-2 to 2)")
|
|
# Emergency priority (2) only: how often to re-alert and when to stop.
|
|
# Pushover requires retry >= 30s and expire <= 10800s (3h).
|
|
retry: int = Field(default=60, ge=30, le=10800, description="Emergency re-alert interval in seconds (priority 2)")
|
|
expire: int = Field(default=3600, ge=30, le=10800, description="Emergency alert expiry in seconds (priority 2)")
|
|
|
|
|
|
class TelegramConfig(BaseModel):
|
|
"""Telegram bot configuration."""
|
|
|
|
bot_token: str = Field(..., description="Bot token from @BotFather")
|
|
chat_id: str = Field(..., description="Chat ID to send messages to")
|
|
|
|
|
|
class EmailConfig(BaseModel):
|
|
"""Email/SMTP configuration."""
|
|
|
|
smtp_server: str = Field(..., description="SMTP server hostname")
|
|
smtp_port: int = Field(default=587, description="SMTP port (587 for TLS, 465 for SSL)")
|
|
username: str = Field(..., description="SMTP username/email")
|
|
password: str = Field(..., description="SMTP password or app password")
|
|
from_email: str = Field(..., description="From email address")
|
|
to_email: str = Field(..., description="Recipient email address")
|
|
use_tls: bool = Field(default=True, description="Use TLS encryption")
|
|
|
|
|
|
# Notification Log schemas
|
|
class NotificationLogResponse(BaseModel):
|
|
"""Schema for notification log API responses."""
|
|
|
|
id: int
|
|
provider_id: int
|
|
provider_name: str | None = None
|
|
provider_type: str | None = None
|
|
event_type: str
|
|
title: str
|
|
message: str
|
|
success: bool
|
|
error_message: str | None = None
|
|
printer_id: int | None = None
|
|
printer_name: str | None = None
|
|
created_at: datetime
|
|
|
|
class Config:
|
|
from_attributes = True
|
|
|
|
|
|
class NotificationLogStats(BaseModel):
|
|
"""Statistics for notification logs."""
|
|
|
|
total: int
|
|
success_count: int
|
|
failure_count: int
|
|
by_event_type: dict[str, int]
|
|
by_provider: dict[str, int]
|