feat(notify-bridge): phase 4 - template system
Implement hybrid template system: - Jinja2 SandboxedEnvironment renderer with error fallback - Context builder: transforms ServiceEvent into flat template variables - Template validator: checks variable references against provider type - Default templates in EN/RU for all 8 event slots (Immich provider) - Template loader reads .jinja2 files, returns slot->content dict - Slots: assets_added/removed, collection_renamed/deleted, sharing_changed, periodic_summary, scheduled_assets, memory_mode Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
This commit is contained in:
@@ -1,5 +1,8 @@
|
||||
"""Template system — rendering, variables, validation."""
|
||||
"""Template system — rendering, variables, validation, defaults."""
|
||||
|
||||
from notify_bridge_core.templates.context import build_template_context
|
||||
from notify_bridge_core.templates.renderer import render_template
|
||||
from notify_bridge_core.templates.validator import validate_template
|
||||
from notify_bridge_core.templates.variables import (
|
||||
BASE_VARIABLES,
|
||||
TemplateVariableDefinition,
|
||||
@@ -11,5 +14,8 @@ __all__ = [
|
||||
"BASE_VARIABLES",
|
||||
"TemplateVariableDefinition",
|
||||
"VariableRegistry",
|
||||
"build_template_context",
|
||||
"registry",
|
||||
"render_template",
|
||||
"validate_template",
|
||||
]
|
||||
|
||||
@@ -0,0 +1,78 @@
|
||||
"""Template context building — transforms ServiceEvent into template variables."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from typing import Any
|
||||
|
||||
from notify_bridge_core.models.events import ServiceEvent
|
||||
|
||||
|
||||
def build_template_context(
|
||||
event: ServiceEvent,
|
||||
target_type: str = "webhook",
|
||||
) -> dict[str, Any]:
|
||||
"""Build a flat template context dict from a ServiceEvent.
|
||||
|
||||
Merges base variables with provider-specific extras.
|
||||
The context is passed directly to Jinja2 templates.
|
||||
|
||||
Args:
|
||||
event: The service event to render.
|
||||
target_type: "telegram" or "webhook" — for conditional formatting.
|
||||
|
||||
Returns:
|
||||
Dict of template variables.
|
||||
"""
|
||||
# Base variables (available to all providers)
|
||||
ctx: dict[str, Any] = {
|
||||
"event_type": event.event_type.value,
|
||||
"timestamp": event.timestamp,
|
||||
"service_name": event.provider_name,
|
||||
"service_type": event.provider_type.value,
|
||||
"collection_name": event.collection_name,
|
||||
"collection_id": event.collection_id,
|
||||
"added_count": event.added_count,
|
||||
"removed_count": event.removed_count,
|
||||
"old_name": event.old_name,
|
||||
"new_name": event.new_name,
|
||||
"target_type": target_type,
|
||||
}
|
||||
|
||||
# Convert MediaAsset list to dicts for template access
|
||||
assets = []
|
||||
for asset in event.added_assets:
|
||||
asset_dict: dict[str, Any] = {
|
||||
"id": asset.id,
|
||||
"type": asset.type.value.upper(), # "IMAGE" or "VIDEO" for template compat
|
||||
"filename": asset.filename,
|
||||
"created_at": asset.created_at.isoformat() if asset.created_at else "",
|
||||
"owner": asset.owner_name or "",
|
||||
"description": asset.description or "",
|
||||
"tags": asset.tags,
|
||||
"thumbnail_url": asset.thumbnail_url or "",
|
||||
"full_url": asset.full_url or "",
|
||||
}
|
||||
# Flatten extras into asset dict for template access
|
||||
asset_dict.update(asset.extra)
|
||||
assets.append(asset_dict)
|
||||
|
||||
ctx["assets"] = assets
|
||||
ctx["added_assets"] = assets # alias for backward compat
|
||||
|
||||
# Asset type flags
|
||||
ctx["has_videos"] = any(a.get("type") == "VIDEO" for a in assets)
|
||||
ctx["has_photos"] = any(a.get("type") == "IMAGE" for a in assets)
|
||||
|
||||
# Provider-specific extras merged at top level
|
||||
ctx.update(event.extra)
|
||||
|
||||
# Provider-specific aliases for Immich
|
||||
if event.provider_type.value == "immich":
|
||||
ctx.setdefault("album_name", event.collection_name)
|
||||
ctx.setdefault("album_id", event.collection_id)
|
||||
if event.old_name:
|
||||
ctx.setdefault("old_album_name", event.old_name)
|
||||
if event.new_name:
|
||||
ctx.setdefault("new_album_name", event.new_name)
|
||||
|
||||
return ctx
|
||||
@@ -0,0 +1,5 @@
|
||||
"""Default template files and loader."""
|
||||
|
||||
from .loader import load_default_templates, get_available_locales
|
||||
|
||||
__all__ = ["load_default_templates", "get_available_locales"]
|
||||
@@ -0,0 +1,15 @@
|
||||
📷 {{ added_count }} new photo(s) added to album "{{ album_name }}".
|
||||
{%- if people %}
|
||||
👤 {{ people | join(", ") }}
|
||||
{%- endif %}
|
||||
{%- if added_assets %}
|
||||
{%- for asset in added_assets %}
|
||||
• {%- if asset.type == "VIDEO" %} 🎬{% else %} 🖼️{% endif %} {{ asset.filename }}
|
||||
{%- if asset.city %} 📍 {{ asset.city }}{% if asset.country %}, {{ asset.country }}{% endif %}{% endif %}
|
||||
{%- if asset.is_favorite %} ❤️{% endif %}
|
||||
{%- endfor %}
|
||||
{%- endif %}
|
||||
{%- if target_type == "telegram" and has_videos %}
|
||||
|
||||
⚠️ Videos may not be sent due to Telegram's 50 MB file size limit.
|
||||
{%- endif %}
|
||||
@@ -0,0 +1 @@
|
||||
🗑️ {{ removed_count }} photo(s) removed from album "{{ album_name }}".
|
||||
@@ -0,0 +1 @@
|
||||
🗑️ Album "{{ collection_name }}" was deleted.
|
||||
@@ -0,0 +1 @@
|
||||
✏️ Album "{{ old_name }}" renamed to "{{ new_name }}".
|
||||
@@ -0,0 +1,4 @@
|
||||
📅 On this day:
|
||||
{%- for asset in assets %}
|
||||
• {%- if asset.type == "VIDEO" %} 🎬{% else %} 🖼️{% endif %} {{ asset.filename }} ({{ asset.created_at[:4] }})
|
||||
{%- endfor %}
|
||||
@@ -0,0 +1,5 @@
|
||||
📋 Tracked Albums Summary ({{ albums | length }} albums):
|
||||
{%- for album in albums %}
|
||||
• {{ album.name }}: {{ album.asset_count }} assets
|
||||
{%- if album.url %} — {{ album.url }}{% endif %}
|
||||
{%- endfor %}
|
||||
@@ -0,0 +1,4 @@
|
||||
📸 Photos from "{{ album_name }}":
|
||||
{%- for asset in assets %}
|
||||
• {%- if asset.type == "VIDEO" %} 🎬{% else %} 🖼️{% endif %} {{ asset.filename }}
|
||||
{%- endfor %}
|
||||
@@ -0,0 +1 @@
|
||||
🔗 Sharing changed for album "{{ album_name }}".
|
||||
@@ -0,0 +1,57 @@
|
||||
"""Default template loader — reads .jinja2 files for seeding into DB."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import logging
|
||||
from pathlib import Path
|
||||
from typing import Any
|
||||
|
||||
_LOGGER = logging.getLogger(__name__)
|
||||
|
||||
_DEFAULTS_DIR = Path(__file__).parent
|
||||
|
||||
# Mapping of template slot names to file names
|
||||
SLOT_FILE_MAP: dict[str, str] = {
|
||||
"message_assets_added": "assets_added.jinja2",
|
||||
"message_assets_removed": "assets_removed.jinja2",
|
||||
"message_collection_renamed": "collection_renamed.jinja2",
|
||||
"message_collection_deleted": "collection_deleted.jinja2",
|
||||
"message_sharing_changed": "sharing_changed.jinja2",
|
||||
"periodic_summary_message": "periodic_summary.jinja2",
|
||||
"scheduled_assets_message": "scheduled_assets.jinja2",
|
||||
"memory_mode_message": "memory_mode.jinja2",
|
||||
}
|
||||
|
||||
|
||||
def load_default_templates(locale: str = "en") -> dict[str, str]:
|
||||
"""Load default template strings for a locale.
|
||||
|
||||
Args:
|
||||
locale: "en" or "ru"
|
||||
|
||||
Returns:
|
||||
Dict mapping slot name -> template string content.
|
||||
"""
|
||||
locale_dir = _DEFAULTS_DIR / locale
|
||||
if not locale_dir.is_dir():
|
||||
_LOGGER.warning("No default templates for locale '%s'", locale)
|
||||
return {}
|
||||
|
||||
templates: dict[str, str] = {}
|
||||
for slot_name, filename in SLOT_FILE_MAP.items():
|
||||
filepath = locale_dir / filename
|
||||
if filepath.exists():
|
||||
templates[slot_name] = filepath.read_text(encoding="utf-8").strip()
|
||||
else:
|
||||
_LOGGER.debug("Missing default template: %s/%s", locale, filename)
|
||||
|
||||
return templates
|
||||
|
||||
|
||||
def get_available_locales() -> list[str]:
|
||||
"""Return list of available locale codes."""
|
||||
locales = []
|
||||
for path in _DEFAULTS_DIR.iterdir():
|
||||
if path.is_dir() and not path.name.startswith("_"):
|
||||
locales.append(path.name)
|
||||
return sorted(locales)
|
||||
@@ -0,0 +1,15 @@
|
||||
📷 {{ added_count }} новых фото добавлено в альбом "{{ album_name }}".
|
||||
{%- if people %}
|
||||
👤 {{ people | join(", ") }}
|
||||
{%- endif %}
|
||||
{%- if added_assets %}
|
||||
{%- for asset in added_assets %}
|
||||
• {%- if asset.type == "VIDEO" %} 🎬{% else %} 🖼️{% endif %} {{ asset.filename }}
|
||||
{%- if asset.city %} 📍 {{ asset.city }}{% if asset.country %}, {{ asset.country }}{% endif %}{% endif %}
|
||||
{%- if asset.is_favorite %} ❤️{% endif %}
|
||||
{%- endfor %}
|
||||
{%- endif %}
|
||||
{%- if target_type == "telegram" and has_videos %}
|
||||
|
||||
⚠️ Видео может не отправиться из-за ограничения Telegram в 50 МБ.
|
||||
{%- endif %}
|
||||
@@ -0,0 +1 @@
|
||||
🗑️ {{ removed_count }} фото удалено из альбома "{{ album_name }}".
|
||||
@@ -0,0 +1 @@
|
||||
🗑️ Альбом "{{ collection_name }}" был удалён.
|
||||
@@ -0,0 +1 @@
|
||||
✏️ Альбом "{{ old_name }}" переименован в "{{ new_name }}".
|
||||
@@ -0,0 +1,4 @@
|
||||
📅 В этот день:
|
||||
{%- for asset in assets %}
|
||||
• {%- if asset.type == "VIDEO" %} 🎬{% else %} 🖼️{% endif %} {{ asset.filename }} ({{ asset.created_at[:4] }})
|
||||
{%- endfor %}
|
||||
@@ -0,0 +1,5 @@
|
||||
📋 Сводка альбомов ({{ albums | length }}):
|
||||
{%- for album in albums %}
|
||||
• {{ album.name }}: {{ album.asset_count }} файлов
|
||||
{%- if album.url %} — {{ album.url }}{% endif %}
|
||||
{%- endfor %}
|
||||
@@ -0,0 +1,4 @@
|
||||
📸 Фото из "{{ album_name }}":
|
||||
{%- for asset in assets %}
|
||||
• {%- if asset.type == "VIDEO" %} 🎬{% else %} 🖼️{% endif %} {{ asset.filename }}
|
||||
{%- endfor %}
|
||||
@@ -0,0 +1 @@
|
||||
🔗 Изменён доступ к альбому "{{ album_name }}".
|
||||
@@ -0,0 +1,25 @@
|
||||
"""Template rendering engine using Jinja2 SandboxedEnvironment."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import logging
|
||||
from typing import Any
|
||||
|
||||
import jinja2
|
||||
from jinja2.sandbox import SandboxedEnvironment
|
||||
|
||||
_LOGGER = logging.getLogger(__name__)
|
||||
|
||||
_env = SandboxedEnvironment(autoescape=False)
|
||||
|
||||
|
||||
def render_template(template_str: str, context: dict[str, Any]) -> str:
|
||||
"""Render a Jinja2 template string with the given context.
|
||||
|
||||
Falls back to returning the raw template on error.
|
||||
"""
|
||||
try:
|
||||
return _env.from_string(template_str).render(**context)
|
||||
except jinja2.TemplateError as e:
|
||||
_LOGGER.error("Template render error: %s", e)
|
||||
return template_str
|
||||
@@ -0,0 +1,57 @@
|
||||
"""Template validation — check variable references against provider type."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import re
|
||||
|
||||
from notify_bridge_core.providers.base import ServiceProviderType
|
||||
from notify_bridge_core.templates.variables import registry
|
||||
|
||||
|
||||
def validate_template(
|
||||
template_str: str,
|
||||
provider_type: ServiceProviderType,
|
||||
) -> list[str]:
|
||||
"""Validate a template string against available variables for a provider type.
|
||||
|
||||
Returns a list of warning messages (empty if valid).
|
||||
This is a best-effort check — it looks for {{ var }} references
|
||||
and warns about unknown top-level variables.
|
||||
"""
|
||||
warnings: list[str] = []
|
||||
available = registry.get_variable_names(provider_type)
|
||||
|
||||
# Also allow common runtime variables not in the registry
|
||||
runtime_vars = {
|
||||
"target_type", "has_videos", "has_photos",
|
||||
"added_assets", "assets", "albums",
|
||||
}
|
||||
allowed = available | runtime_vars
|
||||
|
||||
# Find all {{ var }} and {{ var.attr }} references
|
||||
pattern = re.compile(r"\{\{-?\s*(\w+)")
|
||||
referenced = set()
|
||||
for match in pattern.finditer(template_str):
|
||||
referenced.add(match.group(1))
|
||||
|
||||
# Also find {% for x in var %} loop variables
|
||||
for_pattern = re.compile(r"\{%-?\s*for\s+\w+\s+in\s+(\w+)")
|
||||
for match in for_pattern.finditer(template_str):
|
||||
referenced.add(match.group(1))
|
||||
|
||||
# Also find {% if var %} conditionals
|
||||
if_pattern = re.compile(r"\{%-?\s*if\s+(\w+)")
|
||||
for match in if_pattern.finditer(template_str):
|
||||
referenced.add(match.group(1))
|
||||
|
||||
# Filter out loop variables (from {% for x in ... %})
|
||||
loop_vars = set()
|
||||
loop_var_pattern = re.compile(r"\{%-?\s*for\s+(\w+)\s+in")
|
||||
for match in loop_var_pattern.finditer(template_str):
|
||||
loop_vars.add(match.group(1))
|
||||
|
||||
# Check for unknown variables
|
||||
for var in referenced - allowed - loop_vars:
|
||||
warnings.append(f"Unknown variable '{var}' for provider type '{provider_type.value}'")
|
||||
|
||||
return warnings
|
||||
Reference in New Issue
Block a user