haos-hacs-immich-album-watcher/packages/server/src/immich_watcher_server/api/template_configs.py

"""Template configuration CRUD API routes."""

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

from jinja2.sandbox import SandboxedEnvironment
from jinja2 import TemplateSyntaxError, UndefinedError

from ..auth.dependencies import get_current_user
from ..database.engine import get_session
from ..database.models import TemplateConfig, User

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

# Sample data for template preview
_SAMPLE_CONTEXT = {
    "album_name": "Family Photos",
    "album_url": "https://immich.example.com/share/abc123",
    "album_created": "01.01.2024",
    "album_updated": "19.03.2026",
    "added_count": 3,
    "removed_count": 0,
    "change_type": "assets_added",
    "people": "Alice, Bob",
    "assets": "\n  • 🖼️ IMG_001.jpg\n  • 🖼️ IMG_002.jpg\n  • 🎬 VID_003.mp4",
    "common_date": " from 19.03.2026",
    "common_location": " in Paris, France",
    "video_warning": "",
    "old_name": "Old Album",
    "new_name": "New Album",
    "album_count": 5,
    "albums": "\n  • Family Photos: https://example.com/share/abc",
    "asset_count": 10,
    "more_count": 7,
}


class TemplateConfigCreate(BaseModel):
    name: str
    message_assets_added: str | None = None
    message_assets_removed: str | None = None
    message_album_renamed: str | None = None
    message_album_deleted: str | None = None
    message_asset_image: str | None = None
    message_asset_video: str | None = None
    message_assets_format: str | None = None
    message_assets_more: str | None = None
    message_people_format: str | None = None
    date_format: str | None = None
    common_date_template: str | None = None
    date_if_unique_template: str | None = None
    location_format: str | None = None
    common_location_template: str | None = None
    location_if_unique_template: str | None = None
    favorite_indicator: str | None = None
    periodic_summary_message: str | None = None
    periodic_album_template: str | None = None
    scheduled_assets_message: str | None = None
    memory_mode_message: str | None = None
    video_warning: str | None = None


TemplateConfigUpdate = TemplateConfigCreate  # Same shape, all optional


@router.get("")
async def list_configs(
    user: User = Depends(get_current_user),
    session: AsyncSession = Depends(get_session),
):
    from sqlalchemy import or_
    result = await session.exec(
        select(TemplateConfig).where(
            or_(TemplateConfig.user_id == user.id, TemplateConfig.user_id == 0)
        )
    )
    return [_response(c) for c in result.all()]


@router.get("/variables")
async def get_template_variables():
    """Get the variable reference for all template slots."""
    from .template_vars import TEMPLATE_VARIABLES
    return TEMPLATE_VARIABLES


@router.post("", status_code=status.HTTP_201_CREATED)
async def create_config(
    body: TemplateConfigCreate,
    user: User = Depends(get_current_user),
    session: AsyncSession = Depends(get_session),
):
    data = {k: v for k, v in body.model_dump().items() if v is not None}
    config = TemplateConfig(user_id=user.id, **data)
    session.add(config)
    await session.commit()
    await session.refresh(config)
    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: TemplateConfigUpdate,
    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():
        if value is not None:
            setattr(config, field, value)
    session.add(config)
    await session.commit()
    await session.refresh(config)
    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),
):
    config = await _get(session, config_id, user.id)
    await session.delete(config)
    await session.commit()


@router.post("/{config_id}/preview")
async def preview_config(
    config_id: int,
    slot: str = "message_assets_added",
    user: User = Depends(get_current_user),
    session: AsyncSession = Depends(get_session),
):
    """Render a specific template slot with sample data."""
    config = await _get(session, config_id, user.id)
    template_body = getattr(config, slot, None)
    if template_body is None:
        raise HTTPException(status_code=400, detail=f"Unknown slot: {slot}")
    try:
        env = SandboxedEnvironment(autoescape=False)
        tmpl = env.from_string(template_body)
        rendered = tmpl.render(**_SAMPLE_CONTEXT)
        return {"slot": slot, "rendered": rendered}
    except Exception as e:
        raise HTTPException(status_code=400, detail=f"Template error: {e}")


class PreviewRequest(BaseModel):
    template: str


@router.post("/preview-raw")
async def preview_raw(
    body: PreviewRequest,
    user: User = Depends(get_current_user),
):
    """Render arbitrary Jinja2 template text with sample data. For live preview while editing."""
    try:
        env = SandboxedEnvironment(autoescape=False)
        tmpl = env.from_string(body.template)
        rendered = tmpl.render(**_SAMPLE_CONTEXT)
        return {"rendered": rendered}
    except TemplateSyntaxError as e:
        return {
            "rendered": None,
            "error": e.message,
            "error_line": e.lineno,
        }
    except UndefinedError as e:
        return {"rendered": None, "error": str(e), "error_line": None}
    except Exception as e:
        return {"rendered": None, "error": str(e), "error_line": None}


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


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