feat: test menu dropdown, split text/media messages, target settings, provider URL links
- Replace 3 test buttons with unified dropdown menu (basic/periodic/scheduled/memory) - Send text message first, then assets as reply (not combined caption+media) - Pass all target config settings to Telegram client (disable_url_preview, max_media, chunk_delay, etc.) - Real data test notifications for periodic/scheduled/memory (fetch from Immich) - Provider card URL is now a clickable hyperlink - Localized test type labels (EN/RU) Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
This commit is contained in:
@@ -118,3 +118,210 @@ async def _test_webhook(target: NotificationTarget, locale: str = "en") -> dict:
|
||||
"message": _get_test_message(locale, "webhook"),
|
||||
"event_type": "test",
|
||||
})
|
||||
|
||||
|
||||
async def send_real_data_notification(
|
||||
target: NotificationTarget,
|
||||
template_str: str,
|
||||
test_type: str,
|
||||
provider_type: str,
|
||||
provider_config: dict,
|
||||
collection_ids: list[str],
|
||||
date_format: str = "%d.%m.%Y, %H:%M UTC",
|
||||
date_only_format: str = "%d.%m.%Y",
|
||||
) -> dict:
|
||||
"""Fetch real data from provider, render template, and send notification."""
|
||||
from datetime import datetime, timezone
|
||||
from jinja2.sandbox import SandboxedEnvironment
|
||||
|
||||
if not template_str:
|
||||
return {"success": False, "error": f"No template configured for {test_type}"}
|
||||
|
||||
# Fetch real data from provider
|
||||
ctx: dict = {}
|
||||
try:
|
||||
ctx = await _build_real_context(
|
||||
provider_type, provider_config, collection_ids,
|
||||
test_type, date_format, date_only_format,
|
||||
)
|
||||
except Exception as e:
|
||||
_LOGGER.error("Failed to fetch real data for test: %s", e)
|
||||
return {"success": False, "error": f"Failed to fetch provider data: {e}"}
|
||||
|
||||
ctx["target_type"] = target.type
|
||||
ctx["date_format"] = date_format
|
||||
ctx["date_only_format"] = date_only_format
|
||||
|
||||
# Render template
|
||||
try:
|
||||
env = SandboxedEnvironment(autoescape=False)
|
||||
tmpl = env.from_string(template_str)
|
||||
message = tmpl.render(**ctx)
|
||||
except Exception as e:
|
||||
return {"success": False, "error": f"Template render error: {e}"}
|
||||
|
||||
# Send
|
||||
try:
|
||||
if target.type == "telegram":
|
||||
return await _test_telegram_with_message(target, message)
|
||||
elif target.type == "webhook":
|
||||
return await _test_webhook_with_message(target, message)
|
||||
return {"success": False, "error": f"Unknown target type: {target.type}"}
|
||||
except Exception as e:
|
||||
_LOGGER.error("Test notification failed: %s", e)
|
||||
return {"success": False, "error": str(e)}
|
||||
|
||||
|
||||
async def _build_real_context(
|
||||
provider_type: str,
|
||||
provider_config: dict,
|
||||
collection_ids: list[str],
|
||||
test_type: str,
|
||||
date_format: str,
|
||||
date_only_format: str,
|
||||
) -> dict:
|
||||
"""Build template context from real provider data."""
|
||||
from datetime import datetime, timezone
|
||||
|
||||
if provider_type != "immich":
|
||||
return {"date": datetime.now(timezone.utc).strftime(date_only_format)}
|
||||
|
||||
from notify_bridge_core.providers.immich import ImmichServiceProvider
|
||||
|
||||
async with aiohttp.ClientSession() as http_session:
|
||||
immich = ImmichServiceProvider(
|
||||
http_session,
|
||||
provider_config.get("url", ""),
|
||||
provider_config.get("api_key", ""),
|
||||
provider_config.get("external_domain"),
|
||||
"Immich",
|
||||
)
|
||||
connected = await immich.connect()
|
||||
if not connected:
|
||||
raise RuntimeError("Failed to connect to Immich")
|
||||
|
||||
# Fetch album data for all tracked collections
|
||||
collections = []
|
||||
all_assets = []
|
||||
|
||||
for album_id in collection_ids:
|
||||
album = await immich.client.get_album(album_id)
|
||||
if not album:
|
||||
continue
|
||||
|
||||
# Get shared link for public URL
|
||||
shared_links = await immich.client.get_shared_links(album_id)
|
||||
ext_domain = provider_config.get("external_domain") or provider_config.get("url", "")
|
||||
album_public_url = ""
|
||||
for link in shared_links:
|
||||
if link.is_accessible and not link.is_expired and not link.has_password:
|
||||
album_public_url = f"{ext_domain.rstrip('/')}/share/{link.key}"
|
||||
break
|
||||
|
||||
collections.append({
|
||||
"name": album.name,
|
||||
"url": album_public_url or f"{ext_domain.rstrip('/')}/albums/{album_id}",
|
||||
"public_url": album_public_url,
|
||||
"asset_count": album.asset_count,
|
||||
"shared": album.shared,
|
||||
"photo_count": album.photo_count,
|
||||
"video_count": album.video_count,
|
||||
"owner": album.owner,
|
||||
})
|
||||
|
||||
# Collect assets (limited sample)
|
||||
for asset_id, asset in list(album.assets.items())[:10]:
|
||||
asset_public_url = f"{album_public_url}/photos/{asset_id}" if album_public_url else ""
|
||||
all_assets.append({
|
||||
"id": asset.id,
|
||||
"filename": asset.filename,
|
||||
"type": asset.type.upper(),
|
||||
"created_at": asset.created_at,
|
||||
"owner": asset.owner_name or "",
|
||||
"description": asset.description or "",
|
||||
"people": list(asset.people),
|
||||
"is_favorite": asset.is_favorite,
|
||||
"rating": asset.rating,
|
||||
"city": asset.city or "",
|
||||
"state": asset.state or "",
|
||||
"country": asset.country or "",
|
||||
"public_url": asset_public_url,
|
||||
"url": f"{ext_domain.rstrip('/')}/api/assets/{asset.id}/original",
|
||||
"photo_url": f"{ext_domain.rstrip('/')}/api/assets/{asset.id}/thumbnail",
|
||||
})
|
||||
|
||||
# Build context based on test type
|
||||
now = datetime.now(timezone.utc)
|
||||
ctx: dict = {
|
||||
"date": now.strftime(date_only_format),
|
||||
"timestamp": now.isoformat(),
|
||||
"service_name": "Immich",
|
||||
"service_type": "immich",
|
||||
"collections": collections,
|
||||
"albums": collections, # alias
|
||||
"assets": all_assets,
|
||||
}
|
||||
|
||||
# Common date/location for assets
|
||||
if len(all_assets) > 1:
|
||||
dates = set()
|
||||
for a in all_assets:
|
||||
ca = a.get("created_at", "")
|
||||
if ca:
|
||||
dates.add(ca[:10])
|
||||
if len(dates) == 1:
|
||||
try:
|
||||
ctx["common_date"] = datetime.fromisoformat(dates.pop()).strftime(date_only_format)
|
||||
except (ValueError, TypeError):
|
||||
ctx["common_date"] = ""
|
||||
else:
|
||||
ctx["common_date"] = ""
|
||||
|
||||
locations = set()
|
||||
for a in all_assets:
|
||||
city = a.get("city", "")
|
||||
country = a.get("country", "")
|
||||
if city:
|
||||
locations.add(f"{city}, {country}" if country else city)
|
||||
else:
|
||||
locations.add("")
|
||||
if len(locations) == 1 and "" not in locations:
|
||||
ctx["common_location"] = locations.pop()
|
||||
else:
|
||||
ctx["common_location"] = ""
|
||||
else:
|
||||
ctx["common_date"] = ""
|
||||
ctx["common_location"] = ""
|
||||
|
||||
# Add first collection details as top-level for periodic-style templates
|
||||
if collections:
|
||||
first = collections[0]
|
||||
ctx.update({
|
||||
"collection_name": first["name"],
|
||||
"album_name": first["name"],
|
||||
"public_url": first.get("public_url", ""),
|
||||
"album_url": first.get("url", ""),
|
||||
"shared": first.get("shared", False),
|
||||
"photo_count": first.get("photo_count", 0),
|
||||
"video_count": first.get("video_count", 0),
|
||||
"owner": first.get("owner", ""),
|
||||
})
|
||||
else:
|
||||
ctx.update({
|
||||
"collection_name": "", "album_name": "",
|
||||
"public_url": "", "album_url": "",
|
||||
"shared": False, "photo_count": 0, "video_count": 0, "owner": "",
|
||||
})
|
||||
|
||||
# People across all assets
|
||||
people = set()
|
||||
for a in all_assets:
|
||||
people.update(a.get("people", []))
|
||||
ctx["people"] = list(people)
|
||||
ctx["has_videos"] = any(a.get("type") == "VIDEO" for a in all_assets)
|
||||
ctx["has_photos"] = any(a.get("type") == "IMAGE" for a in all_assets)
|
||||
ctx["added_count"] = len(all_assets)
|
||||
ctx["added_assets"] = all_assets
|
||||
ctx["protected_url"] = ""
|
||||
|
||||
return ctx
|
||||
|
||||
Reference in New Issue
Block a user