alexei.dolgolyov
68ac13b452
Add full NUT support as a polling-based service provider: - Async TCP client for upsd protocol (port 3493, configurable) - 8 event types: online, on_battery, low_battery, battery_restored, comms_lost, comms_restored, replace_battery, overload - 3 bot commands: /status, /devices, /battery - 38 Jinja2 templates (EN+RU notification + command templates) - Database: tracking config fields, migration, seeds - Frontend: provider form with host/port/credentials, grid items, i18n Provider-agnostic UI improvements: - Remove hardcoded 'immich' defaults from all config forms - Dynamic collection labels per provider type (Albums/Repos/Boards/UPS Devices) - Capability-driven test types instead of provider type checks - Template variable helpers for all providers (was Immich-only) - Guard Immich-only shared link check to Immich providers - Auto-clear stale global provider filter from localStorage - EntitySelect search placeholder shows current selection - Fix noneLabel in linked target config selectors New CLAUDE.md rule #8: no provider-specific hardcoding
557 lines
24 KiB
Python
557 lines
24 KiB
Python
"""Template configuration CRUD API routes.
|
|
|
|
Template content is stored in TemplateSlot child rows (one per slot_name).
|
|
The API exposes slots as a flat dict in create/update/response payloads.
|
|
"""
|
|
|
|
import logging
|
|
from typing import Any
|
|
|
|
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, TemplateSlot, User
|
|
from ..services.sample_context import _SAMPLE_CONTEXT
|
|
|
|
_LOGGER = logging.getLogger(__name__)
|
|
|
|
router = APIRouter(prefix="/api/template-configs", tags=["template-configs"])
|
|
|
|
|
|
class TemplateConfigCreate(BaseModel):
|
|
provider_type: str
|
|
name: str
|
|
description: str | None = None
|
|
icon: str | None = None
|
|
date_format: str | None = None
|
|
date_only_format: str | None = None
|
|
slots: dict[str, dict[str, str]] = {} # slot_name -> {locale -> template text}
|
|
|
|
|
|
class TemplateConfigUpdate(BaseModel):
|
|
name: str | None = None
|
|
description: str | None = None
|
|
icon: str | None = None
|
|
date_format: str | None = None
|
|
date_only_format: str | None = None
|
|
slots: dict[str, dict[str, str]] | None = None # partial update
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Helpers
|
|
# ---------------------------------------------------------------------------
|
|
|
|
async def _load_slots(session: AsyncSession, config_id: int) -> dict[str, dict[str, str]]:
|
|
"""Load all template slots for a config as {slot_name: {locale: template}}."""
|
|
result = await session.exec(
|
|
select(TemplateSlot).where(TemplateSlot.config_id == config_id)
|
|
)
|
|
slots: dict[str, dict[str, str]] = {}
|
|
for s in result.all():
|
|
slots.setdefault(s.slot_name, {})[s.locale] = s.template
|
|
return slots
|
|
|
|
|
|
async def _save_slots(
|
|
session: AsyncSession, config_id: int, slots: dict[str, dict[str, str]]
|
|
) -> None:
|
|
"""Create or update template slots for a config (locale-aware)."""
|
|
for slot_name, locale_map in slots.items():
|
|
for locale, template_text in locale_map.items():
|
|
result = await session.exec(
|
|
select(TemplateSlot).where(
|
|
TemplateSlot.config_id == config_id,
|
|
TemplateSlot.slot_name == slot_name,
|
|
TemplateSlot.locale == locale,
|
|
)
|
|
)
|
|
existing = result.first()
|
|
if existing:
|
|
existing.template = template_text
|
|
session.add(existing)
|
|
else:
|
|
session.add(TemplateSlot(
|
|
config_id=config_id,
|
|
slot_name=slot_name,
|
|
locale=locale,
|
|
template=template_text,
|
|
))
|
|
|
|
|
|
async def _response(session: AsyncSession, c: TemplateConfig) -> dict[str, Any]:
|
|
"""Build API response dict for a TemplateConfig, including its slots."""
|
|
slots = await _load_slots(session, c.id)
|
|
return {
|
|
"id": c.id,
|
|
"user_id": c.user_id,
|
|
"provider_type": c.provider_type,
|
|
"name": c.name,
|
|
"description": c.description,
|
|
"icon": c.icon,
|
|
"date_format": c.date_format,
|
|
"date_only_format": c.date_only_format,
|
|
"slots": slots,
|
|
"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
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Routes
|
|
# ---------------------------------------------------------------------------
|
|
|
|
@router.get("")
|
|
async def list_configs(
|
|
provider_type: str | None = None,
|
|
user: User = Depends(get_current_user),
|
|
session: AsyncSession = Depends(get_session),
|
|
):
|
|
from sqlalchemy import or_
|
|
query = select(TemplateConfig).where(
|
|
or_(TemplateConfig.user_id == user.id, TemplateConfig.user_id == 0)
|
|
)
|
|
if provider_type:
|
|
query = query.where(TemplateConfig.provider_type == provider_type)
|
|
result = await session.exec(query)
|
|
return [await _response(session, c) for c in result.all()]
|
|
|
|
|
|
@router.get("/variables")
|
|
async def get_template_variables():
|
|
"""Get template variable reference grouped by slot.
|
|
|
|
Returns a dict keyed by template slot name, each containing:
|
|
- description: what the slot is for
|
|
- variables: dict of variable_name -> description
|
|
- asset_fields: dict of field_name -> description (for slots with assets)
|
|
- album_fields: dict of field_name -> description (for slots with albums)
|
|
"""
|
|
# Core event variables available in all event templates
|
|
event_vars = {
|
|
"collection_id": "Collection ID (UUID)",
|
|
"collection_name": "Collection name",
|
|
"collection_url": "Public share URL (empty if not shared)",
|
|
"public_url": "Public share link URL (empty if no link exists)",
|
|
"protected_url": "Password-protected share link URL (empty if none)",
|
|
"added_count": "Number of assets added",
|
|
"removed_count": "Number of assets removed",
|
|
"people": "Detected people names (list, use {{ people | join(', ') }})",
|
|
"shared": "Whether collection is shared (boolean)",
|
|
"photo_count": "Total photo count in album",
|
|
"video_count": "Total video count in album",
|
|
"owner": "Album owner name",
|
|
"target_type": "Target type: 'telegram' or 'webhook'",
|
|
"has_videos": "Whether added assets contain videos (boolean)",
|
|
"has_photos": "Whether added assets contain photos (boolean)",
|
|
"has_oversized_videos": "Whether any video exceeds the target's size limit (boolean)",
|
|
"max_video_size": "Target video size limit in bytes (null if no limit)",
|
|
"max_video_size_mb": "Target video size limit in MB (null if no limit)",
|
|
# Immich aliases
|
|
"album_name": "Alias for collection_name",
|
|
"album_id": "Alias for collection_id",
|
|
"album_url": "Alias for collection_url",
|
|
}
|
|
rename_vars = {
|
|
**event_vars,
|
|
"old_name": "Previous name (rename events)",
|
|
"new_name": "New name (rename events)",
|
|
}
|
|
sharing_vars = {
|
|
**event_vars,
|
|
"old_shared": "Was shared before change (boolean)",
|
|
"new_shared": "Is shared after change (boolean)",
|
|
}
|
|
asset_fields = {
|
|
"id": "Asset ID (UUID)",
|
|
"filename": "Original filename",
|
|
"type": "IMAGE or VIDEO",
|
|
"created_at": "Creation date/time (ISO 8601)",
|
|
"owner": "Owner display name",
|
|
"description": "User or EXIF description",
|
|
"people": "People detected in this asset (list)",
|
|
"is_favorite": "Whether asset is favorited (boolean)",
|
|
"rating": "Star rating (1-5 or null)",
|
|
"city": "City name",
|
|
"state": "State/region name",
|
|
"country": "Country name",
|
|
"file_size": "File size in bytes (null if unknown)",
|
|
"oversized": "Whether video exceeds the target's size limit (boolean, videos only)",
|
|
"public_url": "Per-asset public share URL (empty if no album link)",
|
|
"url": "Public viewer URL (if shared)",
|
|
"download_url": "Direct download URL (if shared)",
|
|
"photo_url": "Preview image URL (images only, if shared)",
|
|
"playback_url": "Video playback URL (videos only, if shared)",
|
|
}
|
|
album_fields = {
|
|
"name": "Collection/album name",
|
|
"url": "Share URL",
|
|
"public_url": "Public share link URL",
|
|
"asset_count": "Total assets in collection",
|
|
"shared": "Whether collection is shared",
|
|
}
|
|
scheduled_vars = {
|
|
"date": "Current date string",
|
|
"target_type": "Target type: 'telegram' or 'webhook'",
|
|
}
|
|
|
|
return {
|
|
"message_assets_added": {
|
|
"description": "Notification when new assets are added to a collection",
|
|
"variables": {
|
|
**event_vars,
|
|
"added_assets": "List of asset dicts (use {% for asset in added_assets %})",
|
|
"common_date": "Shared date if all assets have the same date (formatted via date_only_format, empty otherwise)",
|
|
"common_location": "Shared location if all assets are from the same place (e.g. 'Paris, France', empty otherwise)",
|
|
},
|
|
"asset_fields": asset_fields,
|
|
},
|
|
"message_assets_removed": {
|
|
"description": "Notification when assets are removed from a collection",
|
|
"variables": {**event_vars, "removed_assets": "List of removed asset IDs (strings)"},
|
|
},
|
|
"message_collection_renamed": {
|
|
"description": "Notification when a collection is renamed",
|
|
"variables": rename_vars,
|
|
},
|
|
"message_collection_deleted": {
|
|
"description": "Notification when a collection is deleted",
|
|
"variables": event_vars,
|
|
},
|
|
"message_sharing_changed": {
|
|
"description": "Notification when sharing status changes",
|
|
"variables": sharing_vars,
|
|
},
|
|
"periodic_summary_message": {
|
|
"description": "Periodic summary of all tracked collections",
|
|
"variables": {**scheduled_vars, "collections": "List of collection dicts (use {% for album in collections %})"},
|
|
"album_fields": album_fields,
|
|
},
|
|
"scheduled_assets_message": {
|
|
"description": "Scheduled asset delivery (daily photo picks)",
|
|
"variables": {**scheduled_vars, "assets": "List of asset dicts (use {% for asset in assets %})"},
|
|
"asset_fields": asset_fields,
|
|
},
|
|
"memory_mode_message": {
|
|
"description": "\"On This Day\" memories from previous years",
|
|
"variables": {**scheduled_vars, "assets": "List of asset dicts (use {% for asset in assets %})"},
|
|
"asset_fields": asset_fields,
|
|
},
|
|
# --- Gitea slots ---
|
|
**_gitea_variables(),
|
|
# --- Planka slots ---
|
|
**_planka_variables(),
|
|
# --- NUT (UPS) slots ---
|
|
**_nut_variables(),
|
|
# --- Scheduler slots ---
|
|
"message_scheduled_message": {
|
|
"description": "Notification for scheduled message events",
|
|
"variables": {
|
|
"tracker_name": "Name of the tracker that fired",
|
|
"fire_count": "How many times this tracker has fired",
|
|
"current_date": "Current date (formatted)",
|
|
"current_time": "Current time (formatted)",
|
|
"current_datetime": "Current date and time (formatted)",
|
|
},
|
|
},
|
|
}
|
|
|
|
|
|
def _gitea_variables() -> dict:
|
|
common = {
|
|
"sender": "Username who triggered the event",
|
|
"sender_name": "Display name of the sender",
|
|
"repo_name": "Repository full name (owner/repo)",
|
|
"repo_url": "Repository URL",
|
|
}
|
|
return {
|
|
"message_push": {
|
|
"description": "Code pushed to repository",
|
|
"variables": {**common, "branch": "Branch name", "commit_count": "Number of commits",
|
|
"compare_url": "Comparison URL", "commits": "List of commit dicts"},
|
|
},
|
|
"message_issue_opened": {
|
|
"description": "Issue opened",
|
|
"variables": {**common, "issue_number": "Issue number", "issue_title": "Issue title",
|
|
"issue_url": "Issue URL", "issue_body": "Issue body text", "issue_labels": "Labels list"},
|
|
},
|
|
"message_issue_closed": {
|
|
"description": "Issue closed",
|
|
"variables": {**common, "issue_number": "Issue number", "issue_title": "Issue title",
|
|
"issue_url": "Issue URL", "issue_state": "Issue state"},
|
|
},
|
|
"message_issue_commented": {
|
|
"description": "Comment on issue",
|
|
"variables": {**common, "issue_number": "Issue number", "issue_title": "Issue title",
|
|
"issue_url": "Issue URL", "comment_body": "Comment text",
|
|
"comment_url": "Comment URL", "comment_author": "Comment author"},
|
|
},
|
|
"message_pr_opened": {
|
|
"description": "Pull request opened",
|
|
"variables": {**common, "pr_number": "PR number", "pr_title": "PR title",
|
|
"pr_url": "PR URL", "pr_body": "PR body text",
|
|
"pr_base": "Base branch", "pr_head": "Head branch", "pr_labels": "Labels list"},
|
|
},
|
|
"message_pr_closed": {
|
|
"description": "Pull request closed",
|
|
"variables": {**common, "pr_number": "PR number", "pr_title": "PR title",
|
|
"pr_url": "PR URL", "pr_state": "PR state", "pr_merged": "Whether PR was merged"},
|
|
},
|
|
"message_pr_merged": {
|
|
"description": "Pull request merged",
|
|
"variables": {**common, "pr_number": "PR number", "pr_title": "PR title",
|
|
"pr_url": "PR URL", "pr_base": "Base branch", "pr_head": "Head branch"},
|
|
},
|
|
"message_pr_commented": {
|
|
"description": "Comment on pull request",
|
|
"variables": {**common, "pr_number": "PR number", "pr_title": "PR title",
|
|
"pr_url": "PR URL", "comment_body": "Comment text",
|
|
"comment_url": "Comment URL", "comment_author": "Comment author"},
|
|
},
|
|
"message_release_published": {
|
|
"description": "Release published",
|
|
"variables": {**common, "release_tag": "Release tag", "release_name": "Release name",
|
|
"release_url": "Release URL", "release_body": "Release notes",
|
|
"release_draft": "Is draft (boolean)", "release_prerelease": "Is prerelease (boolean)"},
|
|
},
|
|
}
|
|
|
|
|
|
def _planka_variables() -> dict:
|
|
common = {
|
|
"sender": "Username who triggered the event",
|
|
"sender_name": "Display name of the sender",
|
|
"board_name": "Board name",
|
|
"board_id": "Board ID",
|
|
"board_url": "Board URL",
|
|
}
|
|
card = {**common, "card_name": "Card name", "card_id": "Card ID", "card_url": "Card URL"}
|
|
return {
|
|
"message_card_created": {"description": "Card created", "variables": {**card, "list_name": "List name", "card_description": "Card description"}},
|
|
"message_card_updated": {"description": "Card updated", "variables": {**card, "card_description": "Card description", "card_due_date": "Due date"}},
|
|
"message_card_moved": {"description": "Card moved between lists", "variables": {**card, "old_list_name": "Previous list", "new_list_name": "New list"}},
|
|
"message_card_deleted": {"description": "Card deleted", "variables": card},
|
|
"message_card_commented": {"description": "Comment added to card", "variables": {**card, "comment_text": "Comment text"}},
|
|
"message_comment_updated": {"description": "Comment updated", "variables": {**card, "comment_text": "Updated comment text"}},
|
|
"message_board_created": {"description": "Board created", "variables": common},
|
|
"message_board_updated": {"description": "Board updated", "variables": common},
|
|
"message_board_deleted": {"description": "Board deleted", "variables": common},
|
|
"message_list_created": {"description": "List created", "variables": {**common, "list_name": "List name"}},
|
|
"message_list_updated": {"description": "List updated", "variables": {**common, "list_name": "List name"}},
|
|
"message_list_deleted": {"description": "List deleted", "variables": {**common, "list_name": "List name"}},
|
|
"message_attachment_created": {"description": "Attachment added", "variables": {**card, "attachment_name": "Attachment filename"}},
|
|
"message_card_label_added": {"description": "Label added to card", "variables": {**card, "label_name": "Label name", "label_color": "Label color"}},
|
|
"message_task_completed": {"description": "Task completed", "variables": {**card, "task_name": "Task name", "task_completed": "Completed (boolean)"}},
|
|
}
|
|
|
|
|
|
def _nut_variables() -> dict:
|
|
common = {
|
|
"ups_name": "UPS device name",
|
|
"ups_model": "UPS hardware model",
|
|
"ups_manufacturer": "UPS manufacturer",
|
|
"battery_charge": "Battery charge percentage",
|
|
"battery_runtime": "Estimated runtime (formatted)",
|
|
"battery_runtime_seconds": "Estimated runtime in seconds",
|
|
"ups_load": "UPS load percentage",
|
|
"ups_status": "Raw status flags (e.g. OL, OB, LB)",
|
|
"input_voltage": "Input voltage",
|
|
"output_voltage": "Output voltage",
|
|
"event_description": "Human-readable event description",
|
|
"previous_status": "Previous UPS status flags",
|
|
}
|
|
return {
|
|
"message_ups_online": {"description": "UPS back on mains power", "variables": common},
|
|
"message_ups_on_battery": {"description": "UPS switched to battery", "variables": common},
|
|
"message_ups_low_battery": {"description": "Battery critically low", "variables": common},
|
|
"message_ups_battery_restored": {"description": "Battery charge recovered", "variables": common},
|
|
"message_ups_comms_lost": {"description": "Communication with UPS lost", "variables": {"ups_name": common["ups_name"], "previous_status": common["previous_status"], "event_description": common["event_description"]}},
|
|
"message_ups_comms_restored": {"description": "Communication restored", "variables": common},
|
|
"message_ups_replace_battery": {"description": "Battery needs replacement", "variables": common},
|
|
"message_ups_overload": {"description": "UPS load exceeded capacity", "variables": common},
|
|
}
|
|
|
|
|
|
@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),
|
|
):
|
|
config = TemplateConfig(
|
|
user_id=user.id,
|
|
provider_type=body.provider_type,
|
|
name=body.name,
|
|
description=body.description or "",
|
|
icon=body.icon or "",
|
|
date_format=body.date_format or "%d.%m.%Y, %H:%M UTC",
|
|
date_only_format=body.date_only_format or "%d.%m.%Y",
|
|
)
|
|
session.add(config)
|
|
await session.flush() # get config.id
|
|
if body.slots:
|
|
await _save_slots(session, config.id, body.slots)
|
|
await session.commit()
|
|
await session.refresh(config)
|
|
return await _response(session, config)
|
|
|
|
|
|
@router.get("/{config_id}")
|
|
async def get_config(
|
|
config_id: int,
|
|
user: User = Depends(get_current_user),
|
|
session: AsyncSession = Depends(get_session),
|
|
):
|
|
config = await _get(session, config_id, user.id)
|
|
return await _response(session, config)
|
|
|
|
|
|
@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, exclude={"slots"}).items():
|
|
if value is not None:
|
|
setattr(config, field, value)
|
|
session.add(config)
|
|
if body.slots is not None:
|
|
await _save_slots(session, config.id, body.slots)
|
|
await session.commit()
|
|
await session.refresh(config)
|
|
return await _response(session, 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_template_config, raise_if_used
|
|
config = await _get(session, config_id, user.id)
|
|
raise_if_used(await check_template_config(session, config.id), config.name)
|
|
# Delete child slots first
|
|
slot_result = await session.exec(
|
|
select(TemplateSlot).where(TemplateSlot.config_id == config.id)
|
|
)
|
|
for slot in slot_result.all():
|
|
await session.delete(slot)
|
|
await session.delete(config)
|
|
await session.commit()
|
|
|
|
|
|
@router.post("/{config_id}/preview")
|
|
async def preview_config(
|
|
config_id: int,
|
|
slot: str = "message_assets_added",
|
|
locale: str = "en",
|
|
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)
|
|
slots = await _load_slots(session, config.id)
|
|
locale_map = slots.get(slot, {})
|
|
template_body = locale_map.get(locale) or locale_map.get("en", "")
|
|
if not template_body:
|
|
raise HTTPException(status_code=400, detail=f"Slot '{slot}' has no template")
|
|
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 DateFormatPreviewRequest(BaseModel):
|
|
date_format: str = "%d.%m.%Y, %H:%M UTC"
|
|
date_only_format: str = "%d.%m.%Y"
|
|
|
|
|
|
@router.post("/preview-date-format")
|
|
async def preview_date_format(
|
|
body: DateFormatPreviewRequest,
|
|
user: User = Depends(get_current_user),
|
|
):
|
|
"""Preview what date/datetime formats look like with sample data."""
|
|
from datetime import datetime, timezone
|
|
sample_dt = datetime(2026, 3, 19, 14, 30, 0, tzinfo=timezone.utc)
|
|
sample_date = datetime(2026, 3, 19)
|
|
result: dict[str, str | None] = {}
|
|
for key, fmt, sample in [
|
|
("date_format", body.date_format, sample_dt),
|
|
("date_only_format", body.date_only_format, sample_date),
|
|
]:
|
|
try:
|
|
result[key] = sample.strftime(fmt)
|
|
except (ValueError, TypeError):
|
|
result[key] = None
|
|
return result
|
|
|
|
|
|
class PreviewRequest(BaseModel):
|
|
template: str
|
|
target_type: str = "telegram" # "telegram" or "webhook"
|
|
date_format: str = "%d.%m.%Y, %H:%M UTC"
|
|
date_only_format: str = "%d.%m.%Y"
|
|
|
|
|
|
@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:
|
|
from datetime import datetime
|
|
ctx = {**_SAMPLE_CONTEXT, "target_type": body.target_type,
|
|
"date_format": body.date_format, "date_only_format": body.date_only_format}
|
|
# Format common_date using the provided date_only_format
|
|
try:
|
|
ctx["common_date"] = datetime(2026, 3, 19).strftime(body.date_only_format)
|
|
except (ValueError, TypeError):
|
|
ctx["common_date"] = "19.03.2026"
|
|
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}
|