Files
notify-bridge/packages/server/src/notify_bridge_server/services/backup_schema.py
T
alexei.dolgolyov ab621b6abc feat: wire tracking-config display filters + per-tracker adaptive polling
Display filters (Immich tracking config):
- favorites_only drops events with no favorited new assets, or filters
  added_assets to favorites only
- assets_order_by/assets_order sort the rendered list
  (date / name / rating / random / none)
- max_assets_to_show caps rendered+attached media (default 5 -> 10)
- include_tags strips people from event extras and tags from each asset
- include_asset_details strips city/country/state/lat/lon/is_favorite/
  rating/description; load-bearing fields (thumbhash, file_size,
  playback_size, cache keys) preserved
- New apply_tracking_display_filters helper in dispatch_helpers; wired
  into watcher, webhooks, scheduled/periodic/memory, and manual
  test-dispatch
- Targets sharing a TrackingConfig dispatch together; targets with
  different TCs each see their own shaped event

Adaptive polling:
- Replace NotificationTracker.batch_duration with adaptive_max_skip
- Per-tracker opt-in: NULL/0 disables back-off (every tick runs);
  positive N caps the skip factor at (N-1)-in-N after long idle
- Scheduler caches the cap in module state for the tick fast-path
- Migration adds the new column; API schemas/responses, frontend types,
  i18n, and the tracker form updated to match
2026-04-24 21:12:10 +03:00

269 lines
6.1 KiB
Python

"""Pydantic models for the configuration backup/restore file format."""
from __future__ import annotations
from enum import Enum
from typing import Any
from pydantic import BaseModel, Field
class SecretsMode(str, Enum):
EXCLUDE = "exclude"
MASKED = "masked"
INCLUDE = "include"
class ConflictMode(str, Enum):
SKIP = "skip"
RENAME = "rename"
OVERWRITE = "overwrite"
class BackupCategory(str, Enum):
PROVIDERS = "providers"
TELEGRAM_BOTS = "telegram_bots"
MATRIX_BOTS = "matrix_bots"
EMAIL_BOTS = "email_bots"
TARGETS = "targets"
TRACKING_CONFIGS = "tracking_configs"
TEMPLATE_CONFIGS = "template_configs"
COMMAND_CONFIGS = "command_configs"
COMMAND_TEMPLATE_CONFIGS = "command_template_configs"
NOTIFICATION_TRACKERS = "notification_trackers"
COMMAND_TRACKERS = "command_trackers"
ACTIONS = "actions"
APP_SETTINGS = "app_settings"
ALL_CATEGORIES = list(BackupCategory)
# Secret fields in provider config dicts
PROVIDER_SECRET_FIELDS = frozenset(
("api_key", "api_token", "webhook_secret", "password",
"client_secret", "refresh_token")
)
# ---------- nested child models ----------
class ReceiverData(BaseModel):
name: str = ""
config: dict[str, Any] = {}
receiver_key: str = ""
locale: str = ""
enabled: bool = True
class TargetData(BaseModel):
id: int
type: str
name: str
icon: str = ""
config: dict[str, Any] = {}
chat_action: str | None = "typing"
receivers: list[ReceiverData] = []
class TemplateSlotData(BaseModel):
slot_name: str
locale: str = "en"
template: str = ""
class TemplateConfigData(BaseModel):
id: int
provider_type: str
name: str
description: str = ""
icon: str = ""
locale: str = ""
date_format: str = "%d.%m.%Y, %H:%M UTC"
date_only_format: str = "%d.%m.%Y"
slots: list[TemplateSlotData] = []
class CommandTemplateSlotData(BaseModel):
slot_name: str
locale: str = "en"
template: str = ""
class CommandTemplateConfigData(BaseModel):
id: int
provider_type: str
name: str
description: str = ""
icon: str = ""
locale: str = ""
slots: list[CommandTemplateSlotData] = []
class TrackerTargetData(BaseModel):
target_id: int
tracking_config_id: int | None = None
template_config_id: int | None = None
enabled: bool = True
quiet_hours_start: str | None = None
quiet_hours_end: str | None = None
class NotificationTrackerData(BaseModel):
id: int
provider_id: int
name: str
icon: str = ""
collection_ids: list[str] = []
filters: dict[str, Any] = {}
scan_interval: int = 60
default_tracking_config_id: int | None = None
default_template_config_id: int | None = None
enabled: bool = True
targets: list[TrackerTargetData] = []
class CommandTrackerListenerData(BaseModel):
listener_type: str
listener_id: int
class CommandTrackerData(BaseModel):
id: int
provider_id: int
command_config_id: int
name: str
icon: str = ""
enabled: bool = True
listeners: list[CommandTrackerListenerData] = []
class ActionRuleData(BaseModel):
name: str = ""
rule_config: dict[str, Any] = {}
enabled: bool = True
order: int = 0
class ActionData(BaseModel):
id: int
provider_id: int
name: str
icon: str = ""
action_type: str
config: dict[str, Any] = {}
schedule_type: str = "interval"
schedule_interval: int = 3600
schedule_cron: str = ""
enabled: bool = False
rules: list[ActionRuleData] = []
class ProviderData(BaseModel):
id: int
type: str
name: str
icon: str = ""
config: dict[str, Any] = {}
class TelegramBotData(BaseModel):
id: int
name: str
token: str = ""
icon: str = ""
bot_username: str = ""
update_mode: str = "none"
class MatrixBotData(BaseModel):
id: int
name: str
icon: str = ""
homeserver_url: str = ""
access_token: str = ""
display_name: str = ""
class EmailBotData(BaseModel):
id: int
name: str
icon: str = ""
email: str = ""
smtp_host: str = ""
smtp_port: int = 587
smtp_username: str = ""
smtp_password: str = ""
smtp_use_tls: bool = True
class TrackingConfigData(BaseModel):
id: int
provider_type: str
name: str
icon: str = ""
# All the boolean / int / str tracking fields are captured generically
fields: dict[str, Any] = {}
class CommandConfigData(BaseModel):
id: int
provider_type: str
name: str
icon: str = ""
enabled_commands: list[str] = []
response_mode: str = "media"
default_count: int = 5
rate_limits: dict[str, Any] = {}
command_template_config_id: int | None = None
class AppSettingData(BaseModel):
key: str
value: str = ""
# ---------- top-level backup envelope ----------
class BackupData(BaseModel):
providers: list[ProviderData] = []
telegram_bots: list[TelegramBotData] = []
matrix_bots: list[MatrixBotData] = []
email_bots: list[EmailBotData] = []
targets: list[TargetData] = []
tracking_configs: list[TrackingConfigData] = []
template_configs: list[TemplateConfigData] = []
command_configs: list[CommandConfigData] = []
command_template_configs: list[CommandTemplateConfigData] = []
notification_trackers: list[NotificationTrackerData] = []
command_trackers: list[CommandTrackerData] = []
actions: list[ActionData] = []
app_settings: list[AppSettingData] = []
class BackupFile(BaseModel):
format: str = "notify-bridge-backup"
version: int = 1
created_at: str = ""
app_version: str = ""
secrets_mode: SecretsMode = SecretsMode.EXCLUDE
categories: list[str] = []
data: BackupData = Field(default_factory=BackupData)
# ---------- import result ----------
class ImportResult(BaseModel):
created: int = 0
skipped: int = 0
overwritten: int = 0
errors: list[str] = []
warnings: list[str] = []
class ValidateResult(BaseModel):
valid: bool = True
version: int = 0
entity_counts: dict[str, int] = {}
warnings: list[str] = []
errors: list[str] = []