feat(tracking): per-config quiet hours with app-level IANA timezone
Add quiet_hours_enabled/start/end to TrackingConfig (HH:MM strings interpreted in the app-level timezone AppSetting). The dispatch path loads the app timezone once per run and passes it through event_allowed_by_config -> in_quiet_hours, so overnight windows like 22:00-07:00 work correctly in any IANA tz. Frontend exposes a Timezone field under Settings and a Quiet Hours section on the Immich tracking-config form with time-picker inputs.
This commit is contained in:
@@ -5,6 +5,7 @@ from __future__ import annotations
|
||||
import logging
|
||||
from datetime import datetime, time, timezone
|
||||
from typing import Any
|
||||
from zoneinfo import ZoneInfo, ZoneInfoNotFoundError
|
||||
|
||||
from sqlmodel import select
|
||||
from sqlmodel.ext.asyncio.session import AsyncSession
|
||||
@@ -29,12 +30,32 @@ from ..database.models import (
|
||||
_LOGGER = logging.getLogger(__name__)
|
||||
|
||||
|
||||
def in_quiet_hours(start: str | None, end: str | None) -> bool:
|
||||
"""Check if the current UTC time is within the quiet hours window."""
|
||||
def _resolve_zoneinfo(tz_name: str | None) -> ZoneInfo:
|
||||
"""Resolve an IANA tz string to a ZoneInfo, falling back to UTC on any error."""
|
||||
if not tz_name:
|
||||
return ZoneInfo("UTC")
|
||||
try:
|
||||
return ZoneInfo(tz_name)
|
||||
except (ZoneInfoNotFoundError, ValueError):
|
||||
_LOGGER.warning("Unknown timezone %r; falling back to UTC", tz_name)
|
||||
return ZoneInfo("UTC")
|
||||
|
||||
|
||||
def in_quiet_hours(
|
||||
start: str | None,
|
||||
end: str | None,
|
||||
tz_name: str | None = "UTC",
|
||||
) -> bool:
|
||||
"""Check if the current time (in the given timezone) is within the quiet window.
|
||||
|
||||
HH:MM strings are interpreted in the supplied timezone. If either bound is
|
||||
missing, quiet hours are disabled.
|
||||
"""
|
||||
if not start or not end:
|
||||
return False
|
||||
try:
|
||||
now = datetime.now(timezone.utc).time()
|
||||
tz = _resolve_zoneinfo(tz_name)
|
||||
now = datetime.now(timezone.utc).astimezone(tz).time()
|
||||
t_start = time.fromisoformat(start)
|
||||
t_end = time.fromisoformat(end)
|
||||
if t_start <= t_end:
|
||||
@@ -46,8 +67,25 @@ def in_quiet_hours(start: str | None, end: str | None) -> bool:
|
||||
return False
|
||||
|
||||
|
||||
def event_allowed_by_config(event: ServiceEvent, tc: TrackingConfig) -> bool:
|
||||
"""Check if an event type is allowed by the tracking config's flags."""
|
||||
async def get_app_timezone(session: AsyncSession) -> str:
|
||||
"""Load the app-level timezone from AppSetting (falls back to UTC)."""
|
||||
from ..api.app_settings import get_setting
|
||||
value = await get_setting(session, "timezone")
|
||||
return value or "UTC"
|
||||
|
||||
|
||||
def event_allowed_by_config(
|
||||
event: ServiceEvent,
|
||||
tc: TrackingConfig,
|
||||
tz_name: str | None = "UTC",
|
||||
) -> bool:
|
||||
"""Check if an event is allowed by the tracking config's flags + quiet hours."""
|
||||
# Quiet hours gate every event type when enabled.
|
||||
if tc.quiet_hours_enabled and in_quiet_hours(
|
||||
tc.quiet_hours_start, tc.quiet_hours_end, tz_name
|
||||
):
|
||||
return False
|
||||
|
||||
event_type = event.event_type.value
|
||||
flag_map = {
|
||||
# Immich events
|
||||
|
||||
@@ -21,7 +21,11 @@ from ..database.models import (
|
||||
NotificationTrackerState,
|
||||
ServiceProvider,
|
||||
)
|
||||
from .dispatch_helpers import event_allowed_by_config, load_link_data
|
||||
from .dispatch_helpers import (
|
||||
event_allowed_by_config,
|
||||
get_app_timezone,
|
||||
load_link_data,
|
||||
)
|
||||
|
||||
_LOGGER = logging.getLogger(__name__)
|
||||
|
||||
@@ -85,7 +89,10 @@ async def check_tracker(tracker_id: int) -> dict[str, Any]:
|
||||
}
|
||||
|
||||
# Load tracker-target links
|
||||
link_data = await load_link_data(session, tracker_id, check_quiet_hours=True)
|
||||
link_data = await load_link_data(session, tracker_id)
|
||||
|
||||
# Load app-level timezone for quiet-hours evaluation.
|
||||
app_tz = await get_app_timezone(session)
|
||||
|
||||
# Snapshot the data we need
|
||||
provider_type = provider.type
|
||||
@@ -236,7 +243,7 @@ async def check_tracker(tracker_id: int) -> dict[str, Any]:
|
||||
for ld in link_data:
|
||||
# Apply per-link event filtering from tracking config
|
||||
tc = ld["tracking_config"]
|
||||
if tc and not event_allowed_by_config(event, tc):
|
||||
if tc and not event_allowed_by_config(event, tc, app_tz):
|
||||
_LOGGER.info(" Skipped by tracking config filter")
|
||||
continue
|
||||
|
||||
|
||||
Reference in New Issue
Block a user