"""SQLModel database table definitions for Notify Bridge.""" from __future__ import annotations from datetime import datetime, timezone from typing import Any from sqlmodel import JSON, Column, Field, SQLModel def _utcnow() -> datetime: return datetime.now(timezone.utc) class User(SQLModel, table=True): id: int | None = Field(default=None, primary_key=True) username: str = Field(index=True, unique=True) hashed_password: str role: str = Field(default="user") created_at: datetime = Field(default_factory=_utcnow) class ServiceProvider(SQLModel, table=True): """A service provider instance (e.g., an Immich server).""" __tablename__ = "service_provider" id: int | None = Field(default=None, primary_key=True) user_id: int = Field(foreign_key="user.id") type: str # ServiceProviderType value ("immich") name: str icon: str = Field(default="") config: dict[str, Any] = Field(default_factory=dict, sa_column=Column(JSON)) created_at: datetime = Field(default_factory=_utcnow) class TelegramBot(SQLModel, table=True): __tablename__ = "telegram_bot" id: int | None = Field(default=None, primary_key=True) user_id: int = Field(foreign_key="user.id") name: str token: str icon: str = Field(default="") bot_username: str = Field(default="") bot_id: int = Field(default=0) created_at: datetime = Field(default_factory=_utcnow) class TelegramChat(SQLModel, table=True): __tablename__ = "telegram_chat" id: int | None = Field(default=None, primary_key=True) bot_id: int = Field(foreign_key="telegram_bot.id") chat_id: str title: str = Field(default="") chat_type: str = Field(default="private") username: str = Field(default="") discovered_at: datetime = Field(default_factory=_utcnow) class TrackingConfig(SQLModel, table=True): """What events to track + scheduling rules. Tied to a provider type.""" __tablename__ = "tracking_config" id: int | None = Field(default=None, primary_key=True) user_id: int = Field(foreign_key="user.id") provider_type: str # Must match provider's type name: str icon: str = Field(default="") # Event-driven tracking track_assets_added: bool = Field(default=True) track_assets_removed: bool = Field(default=False) track_collection_renamed: bool = Field(default=True) track_collection_deleted: bool = Field(default=True) track_sharing_changed: bool = Field(default=False) track_images: bool = Field(default=True) track_videos: bool = Field(default=True) notify_favorites_only: bool = Field(default=False) # Asset display include_tags: bool = Field(default=True) include_asset_details: bool = Field(default=False) max_assets_to_show: int = Field(default=5) assets_order_by: str = Field(default="none") assets_order: str = Field(default="descending") # Periodic summary periodic_enabled: bool = Field(default=False) periodic_interval_days: int = Field(default=1) periodic_start_date: str = Field(default="2025-01-01") periodic_times: str = Field(default="12:00") # Scheduled assets scheduled_enabled: bool = Field(default=False) scheduled_times: str = Field(default="09:00") scheduled_collection_mode: str = Field(default="per_collection") scheduled_limit: int = Field(default=10) scheduled_favorite_only: bool = Field(default=False) scheduled_asset_type: str = Field(default="all") scheduled_min_rating: int = Field(default=0) scheduled_order_by: str = Field(default="random") scheduled_order: str = Field(default="descending") # Memory mode memory_enabled: bool = Field(default=False) memory_times: str = Field(default="09:00") memory_collection_mode: str = Field(default="combined") memory_limit: int = Field(default=10) memory_favorite_only: bool = Field(default=False) memory_asset_type: str = Field(default="all") memory_min_rating: int = Field(default=0) created_at: datetime = Field(default_factory=_utcnow) class TemplateConfig(SQLModel, table=True): """Jinja2 message templates. Tied to a provider type.""" __tablename__ = "template_config" id: int | None = Field(default=None, primary_key=True) user_id: int = Field(default=0) # 0 = system-owned (no FK to allow sentinel) provider_type: str # Must match provider's type name: str description: str = Field(default="") icon: str = Field(default="") # Event-driven notification templates message_assets_added: str = Field(default="") message_assets_removed: str = Field(default="") message_collection_renamed: str = Field(default="") message_collection_deleted: str = Field(default="") message_sharing_changed: str = Field(default="") # Scheduled notification templates periodic_summary_message: str = Field(default="") scheduled_assets_message: str = Field(default="") memory_mode_message: str = Field(default="") date_format: str = Field(default="%d.%m.%Y, %H:%M UTC") date_only_format: str = Field(default="%d.%m.%Y") created_at: datetime = Field(default_factory=_utcnow) class NotificationTarget(SQLModel, table=True): """Where to send notifications. Pure delivery endpoint.""" __tablename__ = "notification_target" id: int | None = Field(default=None, primary_key=True) user_id: int = Field(foreign_key="user.id") type: str # "telegram" or "webhook" name: str icon: str = Field(default="") config: dict[str, Any] = Field(default_factory=dict, sa_column=Column(JSON)) created_at: datetime = Field(default_factory=_utcnow) class Tracker(SQLModel, table=True): """Watches a provider's collections for changes.""" __tablename__ = "tracker" id: int | None = Field(default=None, primary_key=True) user_id: int = Field(foreign_key="user.id") provider_id: int = Field(foreign_key="service_provider.id") name: str icon: str = Field(default="") collection_ids: list[str] = Field(default_factory=list, sa_column=Column(JSON)) scan_interval: int = Field(default=60) batch_duration: int = Field(default=0) # seconds to accumulate events before dispatch (0=immediate) enabled: bool = Field(default=True) created_at: datetime = Field(default_factory=_utcnow) class TrackerTarget(SQLModel, table=True): """Junction between Tracker and NotificationTarget with per-link config.""" __tablename__ = "tracker_target" id: int | None = Field(default=None, primary_key=True) tracker_id: int = Field(foreign_key="tracker.id", index=True) target_id: int = Field(foreign_key="notification_target.id", index=True) tracking_config_id: int | None = Field( default=None, foreign_key="tracking_config.id" ) template_config_id: int | None = Field( default=None, foreign_key="template_config.id" ) enabled: bool = Field(default=True) quiet_hours_start: str | None = None quiet_hours_end: str | None = None commands_config: dict[str, Any] | None = Field( default=None, sa_column=Column(JSON) ) created_at: datetime = Field(default_factory=_utcnow) class TrackerState(SQLModel, table=True): """Persisted state for change detection.""" __tablename__ = "tracker_state" id: int | None = Field(default=None, primary_key=True) tracker_id: int = Field(foreign_key="tracker.id") collection_id: str collection_name: str = Field(default="") shared: bool = Field(default=False) asset_ids: list[str] = Field(default_factory=list, sa_column=Column(JSON)) pending_asset_ids: list[str] = Field(default_factory=list, sa_column=Column(JSON)) last_updated: datetime = Field(default_factory=_utcnow) class EventLog(SQLModel, table=True): """Log of detected events.""" __tablename__ = "event_log" id: int | None = Field(default=None, primary_key=True) tracker_id: int | None = Field(default=None, foreign_key="tracker.id") tracker_name: str = Field(default="") provider_id: int | None = Field(default=None) provider_name: str = Field(default="") event_type: str collection_id: str collection_name: str assets_count: int = Field(default=0) details: dict[str, Any] = Field(default_factory=dict, sa_column=Column(JSON)) created_at: datetime = Field(default_factory=_utcnow)