notify-bridge/packages/server/src/notify_bridge_server/api/tracking_configs.py

"""Tracking configuration CRUD API routes."""

import logging

from fastapi import APIRouter, Depends, HTTPException, status
from pydantic import BaseModel
from sqlmodel import select
from sqlmodel.ext.asyncio.session import AsyncSession

from ..auth.dependencies import get_current_user
from ..database.engine import get_session
from ..database.models import TrackingConfig, User
from ..services.scheduler import reschedule_immich_dispatch_jobs

_LOGGER = logging.getLogger(__name__)

router = APIRouter(prefix="/api/tracking-configs", tags=["tracking-configs"])


class TrackingConfigCreate(BaseModel):
    provider_type: str
    name: str
    icon: str = ""
    track_assets_added: bool = True
    track_assets_removed: bool = False
    track_collection_renamed: bool = True
    track_collection_deleted: bool = True
    track_sharing_changed: bool = False
    track_images: bool = True
    track_videos: bool = True
    notify_favorites_only: bool = False
    include_tags: bool = True
    include_asset_details: bool = False
    max_assets_to_show: int = 10
    assets_order_by: str = "none"
    assets_order: str = "descending"
    periodic_enabled: bool = False
    periodic_interval_days: int = 1
    periodic_start_date: str = "2025-01-01"
    periodic_times: str = "12:00"
    scheduled_enabled: bool = False
    scheduled_times: str = "09:00"
    scheduled_collection_mode: str = "per_collection"
    scheduled_limit: int = 10
    scheduled_favorite_only: bool = False
    scheduled_asset_type: str = "all"
    scheduled_min_rating: int = 0
    scheduled_order_by: str = "random"
    scheduled_order: str = "descending"
    memory_enabled: bool = False
    memory_source: str = "albums"
    memory_times: str = "09:00"
    memory_collection_mode: str = "combined"
    memory_limit: int = 10
    memory_favorite_only: bool = False
    memory_asset_type: str = "all"
    memory_min_rating: int = 0
    quiet_hours_enabled: bool = False
    quiet_hours_start: str | None = None
    quiet_hours_end: str | None = None


class TrackingConfigUpdate(BaseModel):
    name: str | None = None
    icon: str | None = None
    track_assets_added: bool | None = None
    track_assets_removed: bool | None = None
    track_collection_renamed: bool | None = None
    track_collection_deleted: bool | None = None
    track_sharing_changed: bool | None = None
    track_images: bool | None = None
    track_videos: bool | None = None
    notify_favorites_only: bool | None = None
    include_tags: bool | None = None
    include_asset_details: bool | None = None
    max_assets_to_show: int | None = None
    assets_order_by: str | None = None
    assets_order: str | None = None
    periodic_enabled: bool | None = None
    periodic_interval_days: int | None = None
    periodic_start_date: str | None = None
    periodic_times: str | None = None
    scheduled_enabled: bool | None = None
    scheduled_times: str | None = None
    scheduled_collection_mode: str | None = None
    scheduled_limit: int | None = None
    scheduled_favorite_only: bool | None = None
    scheduled_asset_type: str | None = None
    scheduled_min_rating: int | None = None
    scheduled_order_by: str | None = None
    scheduled_order: str | None = None
    memory_enabled: bool | None = None
    memory_source: str | None = None
    memory_times: str | None = None
    memory_collection_mode: str | None = None
    memory_limit: int | None = None
    memory_favorite_only: bool | None = None
    memory_asset_type: str | None = None
    memory_min_rating: int | None = None
    quiet_hours_enabled: bool | None = None
    quiet_hours_start: str | None = None
    quiet_hours_end: str | None = None


@router.get("")
async def list_configs(
    provider_type: str | None = None,
    user: User = Depends(get_current_user),
    session: AsyncSession = Depends(get_session),
):
    from sqlmodel import or_
    query = select(TrackingConfig).where(
        or_(TrackingConfig.user_id == user.id, TrackingConfig.user_id == 0)
    )
    if provider_type:
        query = query.where(TrackingConfig.provider_type == provider_type)
    result = await session.exec(query)
    return [_response(c) for c in result.all()]


@router.post("", status_code=status.HTTP_201_CREATED)
async def create_config(
    body: TrackingConfigCreate,
    user: User = Depends(get_current_user),
    session: AsyncSession = Depends(get_session),
):
    config = TrackingConfig(user_id=user.id, **body.model_dump())
    session.add(config)
    await session.commit()
    await session.refresh(config)
    if config.provider_type == "immich":
        await reschedule_immich_dispatch_jobs()
    return _response(config)


@router.get("/{config_id}")
async def get_config(
    config_id: int,
    user: User = Depends(get_current_user),
    session: AsyncSession = Depends(get_session),
):
    return _response(await _get(session, config_id, user.id))


@router.put("/{config_id}")
async def update_config(
    config_id: int,
    body: TrackingConfigUpdate,
    user: User = Depends(get_current_user),
    session: AsyncSession = Depends(get_session),
):
    config = await _get(session, config_id, user.id)
    for field, value in body.model_dump(exclude_unset=True).items():
        setattr(config, field, value)
    session.add(config)
    await session.commit()
    await session.refresh(config)
    if config.provider_type == "immich":
        await reschedule_immich_dispatch_jobs()
    return _response(config)


@router.delete("/{config_id}", status_code=status.HTTP_204_NO_CONTENT)
async def delete_config(
    config_id: int,
    user: User = Depends(get_current_user),
    session: AsyncSession = Depends(get_session),
):
    from .delete_protection import check_tracking_config, raise_if_used
    config = await _get(session, config_id, user.id)
    raise_if_used(await check_tracking_config(session, config.id), config.name)
    provider_type = config.provider_type
    await session.delete(config)
    await session.commit()
    if provider_type == "immich":
        await reschedule_immich_dispatch_jobs()


def _response(c: TrackingConfig) -> dict:
    return {k: getattr(c, k) for k in TrackingConfig.model_fields if k not in ("user_id", "created_at")} | {
        "created_at": c.created_at.isoformat(),
    }


async def _get(session: AsyncSession, config_id: int, user_id: int) -> TrackingConfig:
    config = await session.get(TrackingConfig, config_id)
    if not config or (config.user_id != user_id and config.user_id != 0):
        raise HTTPException(status_code=404, detail="Tracking config not found")
    return config