feat(immich): wire cron-fired scheduled/periodic/memory dispatch
The scheduled_enabled / scheduled_times (and the periodic / memory counterparts) on TrackingConfig had been wired into the model, the API, and the test-dispatch path — but no production scheduler ever read them, so users saw the slot in the UI and only ever got fires through "Test". This adds the missing cron jobs and the dispatch fan-out, both keyed off the app-level IANA timezone. * services/scheduled_dispatch.py — production fan-out reusing the test-path event builders, picking the slot template per kind, and writing an EventLog row per fire so the dashboard reflects it. * services/scheduler.py — _load_immich_dispatch_jobs builds one CronTrigger per (tracker, kind, HH:MM) from the tracker's default TrackingConfig; reschedule_immich_dispatch_jobs rebuilds them all on any relevant CRUD or timezone change. * tracker / link / tracking-config CRUD endpoints now invalidate. Also: skip dispatch when scheduled/memory yield zero matching assets (prevents header-only "On this day:" spam), and update the EN/RU default scheduled_assets templates to surface that the delivery is a scheduled random selection.
This commit is contained in:
@@ -0,0 +1,242 @@
|
||||
"""Cron-fired scheduled / periodic / memory dispatch for Immich trackers.
|
||||
|
||||
The Immich provider exposes three notification slots that fire on a wall-clock
|
||||
schedule rather than in response to album changes:
|
||||
|
||||
* ``scheduled_assets_message`` — random asset selection at fixed times of day
|
||||
* ``periodic_summary_message`` — album stats summary at fixed times of day
|
||||
* ``memory_mode_message`` — "On This Day" memories at fixed times of day
|
||||
|
||||
The fire times live on the tracker's default ``TrackingConfig`` as comma-
|
||||
separated ``HH:MM`` strings (``scheduled_times`` / ``periodic_times`` /
|
||||
``memory_times``) interpreted in the app-level IANA timezone
|
||||
(``AppSetting.timezone``). The scheduler module wires the cron jobs; this
|
||||
module owns the dispatch flow once a job fires.
|
||||
|
||||
Note on per-link tracking-config overrides: schedule *times* come from the
|
||||
tracker's default config — a per-link override may disable the slot for that
|
||||
link (via ``{kind}_enabled``) but cannot shift its fire time. Consistent with
|
||||
the test-dispatch path in ``manual_dispatch``.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import logging
|
||||
from typing import Literal
|
||||
|
||||
from sqlmodel import select
|
||||
from sqlmodel.ext.asyncio.session import AsyncSession
|
||||
|
||||
from notify_bridge_core.models.events import EventType
|
||||
from notify_bridge_core.notifications.dispatcher import (
|
||||
NotificationDispatcher,
|
||||
TargetConfig,
|
||||
)
|
||||
|
||||
from ..database.engine import get_engine
|
||||
from ..database.models import (
|
||||
EventLog,
|
||||
NotificationTracker,
|
||||
ServiceProvider,
|
||||
TemplateSlot,
|
||||
TrackingConfig,
|
||||
)
|
||||
from .dispatch_helpers import (
|
||||
event_allowed_by_config,
|
||||
get_app_timezone,
|
||||
load_link_data,
|
||||
)
|
||||
from .manual_dispatch import _build_immich_event, _build_immich_periodic_event
|
||||
|
||||
_LOGGER = logging.getLogger(__name__)
|
||||
|
||||
ScheduledKind = Literal["scheduled", "periodic", "memory"]
|
||||
|
||||
# Maps the dispatch kind to the DB slot name that holds its template.
|
||||
# The dispatcher keys templates by ``event.event_type.value`` (always
|
||||
# ``scheduled_message`` here), so we read the right ``TemplateSlot`` row and
|
||||
# inject it under that single event-type key — same pattern as the test path.
|
||||
_SLOT_MAP: dict[ScheduledKind, str] = {
|
||||
"scheduled": "scheduled_assets_message",
|
||||
"periodic": "periodic_summary_message",
|
||||
"memory": "memory_mode_message",
|
||||
}
|
||||
|
||||
|
||||
async def dispatch_scheduled_for_tracker(
|
||||
tracker_id: int, kind: ScheduledKind
|
||||
) -> None:
|
||||
"""Build the slot's event for ``tracker_id`` and fan out to its links.
|
||||
|
||||
Skips silently when the tracker is disabled, the provider is not Immich,
|
||||
the slot is disabled on the tracker's default tracking config, or no link
|
||||
has a ``TemplateConfig`` with the corresponding slot row.
|
||||
"""
|
||||
engine = get_engine()
|
||||
async with AsyncSession(engine) as session:
|
||||
tracker = await session.get(NotificationTracker, tracker_id)
|
||||
if not tracker or not tracker.enabled:
|
||||
return
|
||||
provider = await session.get(ServiceProvider, tracker.provider_id)
|
||||
if not provider or provider.type != "immich":
|
||||
return
|
||||
|
||||
default_tc: TrackingConfig | None = None
|
||||
if tracker.default_tracking_config_id:
|
||||
default_tc = await session.get(
|
||||
TrackingConfig, tracker.default_tracking_config_id
|
||||
)
|
||||
# If the default config disables this kind, nothing to do — schedule
|
||||
# rebuild only adds jobs when the flag is set, but a stale job from
|
||||
# a previous DB state could still fire one tick before invalidation.
|
||||
if default_tc is None or not getattr(default_tc, f"{kind}_enabled", False):
|
||||
_LOGGER.debug(
|
||||
"Scheduled %s skipped for tracker %d: kind disabled on default config",
|
||||
kind, tracker_id,
|
||||
)
|
||||
return
|
||||
|
||||
# Snapshot every field we need outside the session — after the
|
||||
# ``async with`` exits the instances are detached and lazy-load
|
||||
# would fail. Cheaper than re-fetching, safer than touching
|
||||
# attributes through a closed session.
|
||||
provider_id = provider.id
|
||||
provider_config = dict(provider.config)
|
||||
provider_name = provider.name or provider.type
|
||||
tracker_user_id = tracker.user_id
|
||||
tracker_name = tracker.name or ""
|
||||
collection_ids = list(tracker.collection_ids or [])
|
||||
|
||||
app_tz = await get_app_timezone(session)
|
||||
link_data = await load_link_data(session, tracker_id)
|
||||
|
||||
if not link_data:
|
||||
_LOGGER.info(
|
||||
"Scheduled %s for tracker %d: no enabled links, skipping",
|
||||
kind, tracker_id,
|
||||
)
|
||||
return
|
||||
|
||||
if kind == "periodic":
|
||||
event = await _build_immich_periodic_event(
|
||||
provider_config=provider_config,
|
||||
provider_name=provider_name,
|
||||
tracker_name=tracker_name,
|
||||
collection_ids=collection_ids,
|
||||
)
|
||||
else:
|
||||
event = await _build_immich_event(
|
||||
provider_config=provider_config,
|
||||
provider_name=provider_name,
|
||||
tracker_name=tracker_name,
|
||||
collection_ids=collection_ids,
|
||||
test_type=kind,
|
||||
tracking_config=default_tc,
|
||||
)
|
||||
if event is None:
|
||||
_LOGGER.warning(
|
||||
"Scheduled %s for tracker %d: provider returned no event",
|
||||
kind, tracker_id,
|
||||
)
|
||||
return
|
||||
|
||||
# Skip empty payloads for asset-bearing kinds — sending the bare
|
||||
# "On this day:" / "Scheduled delivery —" header with no items below
|
||||
# spams chats with title-only messages every day. ``periodic`` is
|
||||
# different: it's a stats summary that's still meaningful with zero
|
||||
# assets, so we let it through.
|
||||
if kind in ("scheduled", "memory") and not event.added_assets:
|
||||
_LOGGER.info(
|
||||
"Scheduled %s for tracker %d: 0 assets matched, skipping dispatch",
|
||||
kind, tracker_id,
|
||||
)
|
||||
return
|
||||
|
||||
slot_name = _SLOT_MAP[kind]
|
||||
target_configs: list[TargetConfig] = []
|
||||
async with AsyncSession(engine) as session:
|
||||
for ld in link_data:
|
||||
tc = ld["tracking_config"] or default_tc
|
||||
tmpl = ld["template_config"]
|
||||
if tc is not None:
|
||||
# Per-link override may disable this kind even when the
|
||||
# default has it on — honour that here.
|
||||
if not getattr(tc, f"{kind}_enabled", True):
|
||||
continue
|
||||
if not event_allowed_by_config(event, tc, app_tz):
|
||||
continue
|
||||
if tmpl is None:
|
||||
continue
|
||||
|
||||
slot_rows = (await session.exec(
|
||||
select(TemplateSlot).where(
|
||||
TemplateSlot.config_id == tmpl.id,
|
||||
TemplateSlot.slot_name == slot_name,
|
||||
)
|
||||
)).all()
|
||||
if not slot_rows:
|
||||
continue
|
||||
locale_map = {s.locale: s.template for s in slot_rows}
|
||||
template_slots = {EventType.SCHEDULED_MESSAGE.value: locale_map}
|
||||
|
||||
target_configs.append(TargetConfig(
|
||||
type=ld["target_type"],
|
||||
config=ld["target_config"],
|
||||
template_slots=template_slots,
|
||||
date_format=tmpl.date_format,
|
||||
date_only_format=(
|
||||
tmpl.date_only_format or "%d.%m.%Y"
|
||||
),
|
||||
provider_api_key=provider_config.get("api_key"),
|
||||
provider_internal_url=provider_config.get("url", ""),
|
||||
provider_external_url=provider_config.get("external_domain", ""),
|
||||
receivers=ld["receivers"],
|
||||
))
|
||||
|
||||
if not target_configs:
|
||||
_LOGGER.info(
|
||||
"Scheduled %s for tracker %d: no targets after filtering",
|
||||
kind, tracker_id,
|
||||
)
|
||||
return
|
||||
|
||||
# Lazy import to break the watcher↔scheduler↔scheduled_dispatch cycle.
|
||||
from .watcher import _get_telegram_caches
|
||||
from .http_session import get_http_session
|
||||
|
||||
url_cache, asset_cache = await _get_telegram_caches()
|
||||
http_session = await get_http_session()
|
||||
dispatcher = NotificationDispatcher(
|
||||
url_cache=url_cache, asset_cache=asset_cache, session=http_session,
|
||||
)
|
||||
_LOGGER.info(
|
||||
"Dispatching scheduled %s for tracker %d to %d link(s)",
|
||||
kind, tracker_id, len(target_configs),
|
||||
)
|
||||
results = await dispatcher.dispatch(event, target_configs)
|
||||
|
||||
# Mirror the watcher's audit trail: surface scheduled fires in EventLog so
|
||||
# the dashboard shows *why* a notification arrived (otherwise these would
|
||||
# be invisible to the activity feed).
|
||||
successes = sum(1 for r in results if isinstance(r, dict) and r.get("success"))
|
||||
async with AsyncSession(engine) as session:
|
||||
session.add(EventLog(
|
||||
user_id=tracker_user_id,
|
||||
tracker_id=tracker_id,
|
||||
tracker_name=tracker_name,
|
||||
provider_id=provider_id,
|
||||
provider_name=provider_name,
|
||||
event_type=event.event_type.value,
|
||||
collection_id=event.collection_id,
|
||||
collection_name=event.collection_name,
|
||||
assets_count=event.added_count or 0,
|
||||
details={
|
||||
"kind": kind,
|
||||
"slot": slot_name,
|
||||
"trigger": "cron",
|
||||
"timezone": app_tz,
|
||||
"targets_dispatched": len(target_configs),
|
||||
"targets_succeeded": successes,
|
||||
},
|
||||
))
|
||||
await session.commit()
|
||||
Reference in New Issue
Block a user