"""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, StrictUndefined 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 asset matching what build_asset_detail() actually returns _SAMPLE_ASSET = { "id": "a1b2c3d4-e5f6-7890-abcd-ef1234567890", "filename": "IMG_001.jpg", "type": "IMAGE", "created_at": "2026-03-19T10:30:00", "owner": "Alice", "owner_id": "user-uuid-1", "description": "Family picnic", "people": ["Alice", "Bob"], "is_favorite": True, "rating": 5, "latitude": 48.8566, "longitude": 2.3522, "city": "Paris", "state": "Île-de-France", "country": "France", "url": "https://immich.example.com/photos/abc123", "download_url": "https://immich.example.com/api/assets/abc123/original", "photo_url": "https://immich.example.com/api/assets/abc123/thumbnail", } _SAMPLE_VIDEO_ASSET = { **_SAMPLE_ASSET, "id": "d4e5f6a7-b8c9-0123-defg-456789abcdef", "filename": "VID_002.mp4", "type": "VIDEO", "is_favorite": False, "rating": None, "photo_url": None, "playback_url": "https://immich.example.com/api/assets/def456/video", } _SAMPLE_ALBUM = { "name": "Family Photos", "url": "https://immich.example.com/share/abc123", "asset_count": 42, "shared": True, } # Full context covering ALL possible template variables from _build_event_data() _SAMPLE_CONTEXT = { # Core event fields (always present) "album_id": "b2eeeaa4-bba0-477a-a06f-5cb9e21818e8", "album_name": "Family Photos", "album_url": "https://immich.example.com/share/abc123", "change_type": "assets_added", "added_count": 3, "removed_count": 1, "added_assets": [_SAMPLE_ASSET, _SAMPLE_VIDEO_ASSET], "removed_assets": ["asset-id-1", "asset-id-2"], "people": ["Alice", "Bob"], "shared": True, "target_type": "telegram", "has_videos": True, "has_photos": True, # Rename fields (always present, empty for non-rename events) "old_name": "Old Album", "new_name": "New Album", "old_shared": False, "new_shared": True, # Scheduled/periodic variables (for those templates) "albums": [_SAMPLE_ALBUM, {**_SAMPLE_ALBUM, "name": "Vacation 2025", "asset_count": 120}], "assets": [_SAMPLE_ASSET, {**_SAMPLE_ASSET, "filename": "IMG_002.jpg", "city": "London", "country": "UK"}], "date": "2026-03-19", } class TemplateConfigCreate(BaseModel): name: str description: str | None = None icon: str | None = None message_assets_added: str | None = None message_assets_removed: str | None = None message_album_renamed: str | None = None message_album_deleted: str | None = None periodic_summary_message: str | None = None scheduled_assets_message: str | None = None memory_mode_message: str | None = None date_format: 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 target_type: str = "telegram" # "telegram" or "webhook" @router.post("/preview-raw") async def preview_raw( body: PreviewRequest, user: User = Depends(get_current_user), ): """Render arbitrary Jinja2 template text with sample data. Two-pass validation: 1. Parse with default Undefined (catches syntax errors) 2. Render with StrictUndefined (catches unknown variables like {{ asset.a }}) """ # Pass 1: syntax check try: env = SandboxedEnvironment(autoescape=False) env.from_string(body.template) except TemplateSyntaxError as e: return { "rendered": None, "error": e.message, "error_line": e.lineno, } # Pass 2: render with strict undefined to catch unknown variables try: ctx = {**_SAMPLE_CONTEXT, "target_type": body.target_type} strict_env = SandboxedEnvironment(autoescape=False, undefined=StrictUndefined) tmpl = strict_env.from_string(body.template) rendered = tmpl.render(**ctx) return {"rendered": rendered} except UndefinedError as e: # Still a valid template syntactically, but references unknown variable return {"rendered": None, "error": str(e), "error_line": None, "error_type": "undefined"} 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