diff --git a/immich_album_watcher/README.md b/immich_album_watcher/README.md index f0db2f8..5641666 100644 --- a/immich_album_watcher/README.md +++ b/immich_album_watcher/README.md @@ -43,9 +43,12 @@ A Home Assistant custom integration that monitors [Immich](https://immich.app/) | Sensor | People Count | Number of unique people detected | | Sensor | Last Updated | When the album was last modified | | Sensor | Created | When the album was created | -| Sensor | Public URL | Public share link URL (if album is shared) | +| Sensor | Public URL | Public share link URL (accessible links without password) | +| Sensor | Protected URL | Password-protected share link URL (if any exist) | +| Sensor | Protected Password | Password for the protected share link (read-only) | | Binary Sensor | New Assets | On when new assets were recently added | | Camera | Thumbnail | Album cover image | +| Text | Share Password | Editable password for the protected share link | ## Installation @@ -153,7 +156,20 @@ automation: - Home Assistant 2024.1.0 or newer - Immich server with API access -- Valid Immich API key with `album.read` and `asset.read` permissions +- Valid Immich API key with the following permissions: + +### Required API Permissions + +| Permission | Required | Description | +| ---------- | -------- | ----------- | +| `album.read` | Yes | Read album data and asset lists | +| `asset.read` | Yes | Read asset details (type, filename, creation date) | +| `user.read` | Yes | Resolve asset owner names | +| `person.read` | Yes | Read face recognition / people data | +| `sharedLink.read` | Yes | Read shared links for public/protected URL sensors | +| `sharedLink.edit` | Optional | Edit shared link passwords via the Text entity | + +> **Note:** If you don't grant `sharedLink.edit` permission, the "Share Password" text entity will not be able to update passwords but will still display the current password. ## License diff --git a/immich_album_watcher/const.py b/immich_album_watcher/const.py index ecdfbfb..76c2576 100644 --- a/immich_album_watcher/const.py +++ b/immich_album_watcher/const.py @@ -24,6 +24,9 @@ EVENT_ASSETS_REMOVED: Final = f"{DOMAIN}_assets_removed" ATTR_ALBUM_ID: Final = "album_id" ATTR_ALBUM_NAME: Final = "album_name" ATTR_ALBUM_URL: Final = "album_url" +ATTR_ALBUM_URLS: Final = "album_urls" +ATTR_ALBUM_PROTECTED_URL: Final = "album_protected_url" +ATTR_ALBUM_PROTECTED_PASSWORD: Final = "album_protected_password" ATTR_ASSET_COUNT: Final = "asset_count" ATTR_PHOTO_COUNT: Final = "photo_count" ATTR_VIDEO_COUNT: Final = "video_count" @@ -50,7 +53,7 @@ ASSET_TYPE_IMAGE: Final = "IMAGE" ASSET_TYPE_VIDEO: Final = "VIDEO" # Platforms -PLATFORMS: Final = ["sensor", "binary_sensor", "camera"] +PLATFORMS: Final = ["sensor", "binary_sensor", "camera", "text"] # Services SERVICE_REFRESH: Final = "refresh" diff --git a/immich_album_watcher/coordinator.py b/immich_album_watcher/coordinator.py index 16c29b6..7ab6a8c 100644 --- a/immich_album_watcher/coordinator.py +++ b/immich_album_watcher/coordinator.py @@ -40,6 +40,54 @@ from .const import ( _LOGGER = logging.getLogger(__name__) +@dataclass +class SharedLinkInfo: + """Data class for shared link information.""" + + id: str + key: str + has_password: bool = False + password: str | None = None + expires_at: datetime | None = None + allow_download: bool = True + show_metadata: bool = True + + @property + def is_expired(self) -> bool: + """Check if the link has expired.""" + if self.expires_at is None: + return False + return datetime.now(self.expires_at.tzinfo) > self.expires_at + + @property + def is_accessible(self) -> bool: + """Check if the link is accessible without password and not expired.""" + return not self.has_password and not self.is_expired + + @classmethod + def from_api_response(cls, data: dict[str, Any]) -> SharedLinkInfo: + """Create SharedLinkInfo from API response.""" + expires_at = None + if data.get("expiresAt"): + try: + expires_at = datetime.fromisoformat( + data["expiresAt"].replace("Z", "+00:00") + ) + except ValueError: + pass + + password = data.get("password") + return cls( + id=data["id"], + key=data["key"], + has_password=bool(password), + password=password if password else None, + expires_at=expires_at, + allow_download=data.get("allowDownload", True), + show_metadata=data.get("showMetadata", True), + ) + + @dataclass class AssetInfo: """Data class for asset information.""" @@ -174,7 +222,7 @@ class ImmichAlbumWatcherCoordinator(DataUpdateCoordinator[dict[str, AlbumData]]) self._session: aiohttp.ClientSession | None = None self._people_cache: dict[str, str] = {} # person_id -> name self._users_cache: dict[str, str] = {} # user_id -> name - self._shared_links_cache: dict[str, str] = {} # album_id -> share_key + self._shared_links_cache: dict[str, list[SharedLinkInfo]] = {} # album_id -> list of SharedLinkInfo @property def immich_url(self) -> str: @@ -269,8 +317,8 @@ class ImmichAlbumWatcherCoordinator(DataUpdateCoordinator[dict[str, AlbumData]]) return self._users_cache - async def _async_fetch_shared_links(self) -> dict[str, str]: - """Fetch shared links from Immich and cache album_id -> share_key mapping.""" + async def _async_fetch_shared_links(self) -> dict[str, list[SharedLinkInfo]]: + """Fetch shared links from Immich and cache album_id -> SharedLinkInfo mapping.""" if self._session is None: self._session = async_get_clientsession(self.hass) @@ -285,22 +333,26 @@ class ImmichAlbumWatcherCoordinator(DataUpdateCoordinator[dict[str, AlbumData]]) _LOGGER.debug("Fetched %d shared links from Immich", len(data)) self._shared_links_cache.clear() for link in data: - # Only process album-type shared links - link_type = link.get("type", "") album = link.get("album") key = link.get("key") - _LOGGER.debug( - "Shared link: type=%s, key=%s, album_id=%s", - link_type, - key[:8] if key else None, - album.get("id") if album else None, - ) if album and key: album_id = album.get("id") if album_id: - self._shared_links_cache[album_id] = key + link_info = SharedLinkInfo.from_api_response(link) + _LOGGER.debug( + "Shared link: key=%s, album_id=%s, " + "has_password=%s, expired=%s, accessible=%s", + key[:8], + album_id[:8], + link_info.has_password, + link_info.is_expired, + link_info.is_accessible, + ) + if album_id not in self._shared_links_cache: + self._shared_links_cache[album_id] = [] + self._shared_links_cache[album_id].append(link_info) _LOGGER.debug( - "Cached %d album shared links", len(self._shared_links_cache) + "Cached shared links for %d albums", len(self._shared_links_cache) ) else: _LOGGER.warning( @@ -311,18 +363,88 @@ class ImmichAlbumWatcherCoordinator(DataUpdateCoordinator[dict[str, AlbumData]]) return self._shared_links_cache + def _get_accessible_links(self, album_id: str) -> list[SharedLinkInfo]: + """Get all accessible (no password, not expired) shared links for an album.""" + all_links = self._shared_links_cache.get(album_id, []) + return [link for link in all_links if link.is_accessible] + + def _get_non_expired_links(self, album_id: str) -> list[SharedLinkInfo]: + """Get all non-expired shared links for an album (including password-protected).""" + all_links = self._shared_links_cache.get(album_id, []) + return [link for link in all_links if not link.is_expired] + + def _get_protected_only_links(self, album_id: str) -> list[SharedLinkInfo]: + """Get password-protected but not expired shared links for an album.""" + all_links = self._shared_links_cache.get(album_id, []) + return [link for link in all_links if link.has_password and not link.is_expired] + def get_album_public_url(self, album_id: str) -> str | None: - """Get the public URL for an album if it has a shared link.""" - share_key = self._shared_links_cache.get(album_id) - if share_key: - return f"{self._url}/share/{share_key}" + """Get the public URL for an album if it has an accessible shared link.""" + accessible_links = self._get_accessible_links(album_id) + if accessible_links: + return f"{self._url}/share/{accessible_links[0].key}" return None + def get_album_any_url(self, album_id: str) -> str | None: + """Get any non-expired URL for an album (prefers accessible, falls back to protected).""" + # First try accessible links + accessible_links = self._get_accessible_links(album_id) + if accessible_links: + return f"{self._url}/share/{accessible_links[0].key}" + # Fall back to any non-expired link (including password-protected) + non_expired = self._get_non_expired_links(album_id) + if non_expired: + return f"{self._url}/share/{non_expired[0].key}" + return None + + def get_album_protected_url(self, album_id: str) -> str | None: + """Get a protected URL for an album if any password-protected link exists.""" + protected_links = self._get_protected_only_links(album_id) + if protected_links: + return f"{self._url}/share/{protected_links[0].key}" + return None + + def get_album_protected_urls(self, album_id: str) -> list[str]: + """Get all password-protected (but not expired) URLs for an album.""" + protected_links = self._get_protected_only_links(album_id) + return [f"{self._url}/share/{link.key}" for link in protected_links] + + def get_album_protected_password(self, album_id: str) -> str | None: + """Get the password for the first protected link (matches get_album_protected_url).""" + protected_links = self._get_protected_only_links(album_id) + if protected_links and protected_links[0].password: + return protected_links[0].password + return None + + def get_album_public_urls(self, album_id: str) -> list[str]: + """Get all accessible public URLs for an album.""" + accessible_links = self._get_accessible_links(album_id) + return [f"{self._url}/share/{link.key}" for link in accessible_links] + + def get_album_shared_links_info(self, album_id: str) -> list[dict[str, Any]]: + """Get detailed info about all shared links for an album.""" + all_links = self._shared_links_cache.get(album_id, []) + return [ + { + "url": f"{self._url}/share/{link.key}", + "has_password": link.has_password, + "is_expired": link.is_expired, + "expires_at": link.expires_at.isoformat() if link.expires_at else None, + "is_accessible": link.is_accessible, + } + for link in all_links + ] + def _get_asset_public_url(self, album_id: str, asset_id: str) -> str | None: - """Get the public URL for an asset if the album has a shared link.""" - share_key = self._shared_links_cache.get(album_id) - if share_key: - return f"{self._url}/share/{share_key}/photos/{asset_id}" + """Get the public URL for an asset (prefers accessible, falls back to protected).""" + # First try accessible links + accessible_links = self._get_accessible_links(album_id) + if accessible_links: + return f"{self._url}/share/{accessible_links[0].key}/photos/{asset_id}" + # Fall back to any non-expired link + non_expired = self._get_non_expired_links(album_id) + if non_expired: + return f"{self._url}/share/{non_expired[0].key}/photos/{asset_id}" return None async def _async_update_data(self) -> dict[str, AlbumData]: @@ -451,8 +573,8 @@ class ImmichAlbumWatcherCoordinator(DataUpdateCoordinator[dict[str, AlbumData]]) ATTR_PEOPLE: list(album.people), } - # Add album public URL if it has a shared link - album_url = self.get_album_public_url(change.album_id) + # Add album URL if it has a shared link (prefers accessible, falls back to protected) + album_url = self.get_album_any_url(change.album_id) if album_url: event_data[ATTR_ALBUM_URL] = album_url @@ -473,6 +595,57 @@ class ImmichAlbumWatcherCoordinator(DataUpdateCoordinator[dict[str, AlbumData]]) if change.removed_count > 0: self.hass.bus.async_fire(EVENT_ASSETS_REMOVED, event_data) + def get_album_protected_link_id(self, album_id: str) -> str | None: + """Get the ID of the first protected link (matches get_album_protected_url).""" + protected_links = self._get_protected_only_links(album_id) + if protected_links: + return protected_links[0].id + return None + + async def async_set_shared_link_password( + self, link_id: str, password: str | None + ) -> bool: + """Update the password for a shared link via Immich API. + + Args: + link_id: The ID of the shared link to update. + password: The new password, or None/empty string to remove the password. + + Returns: + True if successful, False otherwise. + """ + if self._session is None: + self._session = async_get_clientsession(self.hass) + + headers = { + "x-api-key": self._api_key, + "Content-Type": "application/json", + } + + # Immich API expects null to remove password, or a string to set it + payload = {"password": password if password else None} + + try: + async with self._session.patch( + f"{self._url}/api/shared-links/{link_id}", + headers=headers, + json=payload, + ) as response: + if response.status == 200: + _LOGGER.info("Successfully updated shared link password") + # Refresh shared links cache to reflect the change + await self._async_fetch_shared_links() + return True + else: + _LOGGER.error( + "Failed to update shared link password: HTTP %s", + response.status, + ) + return False + except aiohttp.ClientError as err: + _LOGGER.error("Error updating shared link password: %s", err) + return False + def clear_new_assets_flag(self, album_id: str) -> None: """Clear the new assets flag for an album.""" if self.data and album_id in self.data: diff --git a/immich_album_watcher/sensor.py b/immich_album_watcher/sensor.py index 5a91dba..7e7aa42 100644 --- a/immich_album_watcher/sensor.py +++ b/immich_album_watcher/sensor.py @@ -19,7 +19,10 @@ from homeassistant.helpers.update_coordinator import CoordinatorEntity from .const import ( ATTR_ALBUM_ID, + ATTR_ALBUM_PROTECTED_PASSWORD, + ATTR_ALBUM_PROTECTED_URL, ATTR_ALBUM_URL, + ATTR_ALBUM_URLS, ATTR_ASSET_COUNT, ATTR_CREATED_AT, ATTR_LAST_UPDATED, @@ -55,6 +58,8 @@ async def async_setup_entry( entities.append(ImmichAlbumCreatedSensor(coordinator, entry, album_id)) entities.append(ImmichAlbumPeopleSensor(coordinator, entry, album_id)) entities.append(ImmichAlbumPublicUrlSensor(coordinator, entry, album_id)) + entities.append(ImmichAlbumProtectedUrlSensor(coordinator, entry, album_id)) + entities.append(ImmichAlbumProtectedPasswordSensor(coordinator, entry, album_id)) async_add_entities(entities) @@ -336,7 +341,97 @@ class ImmichAlbumPublicUrlSensor(ImmichAlbumBaseSensor): if not self._album_data: return {} - return { + attrs = { ATTR_ALBUM_ID: self._album_data.id, ATTR_SHARED: self._album_data.shared, } + + # Include all accessible URLs if there are multiple + all_urls = self.coordinator.get_album_public_urls(self._album_id) + if len(all_urls) > 1: + attrs[ATTR_ALBUM_URLS] = all_urls + + # Include detailed info about all shared links (including protected/expired) + links_info = self.coordinator.get_album_shared_links_info(self._album_id) + if links_info: + attrs["shared_links"] = links_info + + return attrs + + +class ImmichAlbumProtectedUrlSensor(ImmichAlbumBaseSensor): + """Sensor representing an Immich album password-protected URL.""" + + _attr_icon = "mdi:link-lock" + _attr_translation_key = "album_protected_url" + + def __init__( + self, + coordinator: ImmichAlbumWatcherCoordinator, + entry: ConfigEntry, + album_id: str, + ) -> None: + """Initialize the sensor.""" + super().__init__(coordinator, entry, album_id) + self._attr_unique_id = f"{entry.entry_id}_{album_id}_protected_url" + + @property + def native_value(self) -> str | None: + """Return the state of the sensor (protected URL).""" + if self._album_data: + return self.coordinator.get_album_protected_url(self._album_id) + return None + + @property + def extra_state_attributes(self) -> dict[str, Any]: + """Return extra state attributes.""" + if not self._album_data: + return {} + + attrs = { + ATTR_ALBUM_ID: self._album_data.id, + } + + # Include all protected URLs if there are multiple + all_urls = self.coordinator.get_album_protected_urls(self._album_id) + if len(all_urls) > 1: + attrs["protected_urls"] = all_urls + + return attrs + + +class ImmichAlbumProtectedPasswordSensor(ImmichAlbumBaseSensor): + """Sensor representing an Immich album protected link password.""" + + _attr_icon = "mdi:key" + _attr_translation_key = "album_protected_password" + + def __init__( + self, + coordinator: ImmichAlbumWatcherCoordinator, + entry: ConfigEntry, + album_id: str, + ) -> None: + """Initialize the sensor.""" + super().__init__(coordinator, entry, album_id) + self._attr_unique_id = f"{entry.entry_id}_{album_id}_protected_password" + + @property + def native_value(self) -> str | None: + """Return the state of the sensor (protected link password).""" + if self._album_data: + return self.coordinator.get_album_protected_password(self._album_id) + return None + + @property + def extra_state_attributes(self) -> dict[str, Any]: + """Return extra state attributes.""" + if not self._album_data: + return {} + + return { + ATTR_ALBUM_ID: self._album_data.id, + ATTR_ALBUM_PROTECTED_URL: self.coordinator.get_album_protected_url( + self._album_id + ), + } diff --git a/immich_album_watcher/text.py b/immich_album_watcher/text.py new file mode 100644 index 0000000..9e3d726 --- /dev/null +++ b/immich_album_watcher/text.py @@ -0,0 +1,148 @@ +"""Text platform for Immich Album Watcher.""" + +from __future__ import annotations + +import logging + +from homeassistant.components.text import TextEntity, TextMode +from homeassistant.config_entries import ConfigEntry +from homeassistant.core import HomeAssistant, callback +from homeassistant.helpers.entity import DeviceInfo +from homeassistant.helpers.entity_platform import AddEntitiesCallback +from homeassistant.helpers.update_coordinator import CoordinatorEntity + +from .const import ( + ATTR_ALBUM_ID, + ATTR_ALBUM_PROTECTED_URL, + CONF_ALBUMS, + DOMAIN, +) +from .coordinator import AlbumData, ImmichAlbumWatcherCoordinator + +_LOGGER = logging.getLogger(__name__) + + +async def async_setup_entry( + hass: HomeAssistant, + entry: ConfigEntry, + async_add_entities: AddEntitiesCallback, +) -> None: + """Set up Immich Album Watcher text entities from a config entry.""" + coordinator: ImmichAlbumWatcherCoordinator = hass.data[DOMAIN][entry.entry_id] + album_ids = entry.options.get(CONF_ALBUMS, []) + + entities: list[TextEntity] = [] + for album_id in album_ids: + entities.append(ImmichAlbumProtectedPasswordText(coordinator, entry, album_id)) + + async_add_entities(entities) + + +class ImmichAlbumProtectedPasswordText( + CoordinatorEntity[ImmichAlbumWatcherCoordinator], TextEntity +): + """Text entity for editing an Immich album's protected link password.""" + + _attr_has_entity_name = True + _attr_icon = "mdi:key-variant" + _attr_translation_key = "album_protected_password_edit" + _attr_mode = TextMode.PASSWORD + _attr_native_max = 100 + + def __init__( + self, + coordinator: ImmichAlbumWatcherCoordinator, + entry: ConfigEntry, + album_id: str, + ) -> None: + """Initialize the text entity.""" + super().__init__(coordinator) + self._album_id = album_id + self._entry = entry + self._attr_unique_id = f"{entry.entry_id}_{album_id}_protected_password_edit" + + @property + def _album_data(self) -> AlbumData | None: + """Get the album data from coordinator.""" + if self.coordinator.data is None: + return None + return self.coordinator.data.get(self._album_id) + + @property + def translation_placeholders(self) -> dict[str, str]: + """Return translation placeholders.""" + if self._album_data: + return {"album_name": self._album_data.name} + return {"album_name": f"Album {self._album_id[:8]}"} + + @property + def available(self) -> bool: + """Return if entity is available. + + Only available when the album has a protected shared link. + """ + if not self.coordinator.last_update_success or self._album_data is None: + return False + # Only available if there's a protected link to edit + return self.coordinator.get_album_protected_link_id(self._album_id) is not None + + @property + def device_info(self) -> DeviceInfo: + """Return device info.""" + return DeviceInfo( + identifiers={(DOMAIN, self._entry.entry_id)}, + name="Immich Album Watcher", + manufacturer="Immich", + entry_type="service", + ) + + @property + def native_value(self) -> str | None: + """Return the current password value.""" + if self._album_data: + return self.coordinator.get_album_protected_password(self._album_id) + return None + + @property + def extra_state_attributes(self) -> dict[str, str]: + """Return extra state attributes.""" + if not self._album_data: + return {} + + attrs = { + ATTR_ALBUM_ID: self._album_data.id, + } + + protected_url = self.coordinator.get_album_protected_url(self._album_id) + if protected_url: + attrs[ATTR_ALBUM_PROTECTED_URL] = protected_url + + return attrs + + async def async_set_value(self, value: str) -> None: + """Set the password for the protected shared link.""" + link_id = self.coordinator.get_album_protected_link_id(self._album_id) + if not link_id: + _LOGGER.error( + "Cannot set password: no protected link found for album %s", + self._album_id, + ) + return + + # Empty string means remove password + password = value if value else None + + success = await self.coordinator.async_set_shared_link_password( + link_id, password + ) + + if success: + # Trigger a coordinator update to refresh the state + await self.coordinator.async_request_refresh() + else: + _LOGGER.error("Failed to update password for album %s", self._album_id) + + @callback + def _handle_coordinator_update(self) -> None: + """Handle updated data from the coordinator.""" + self.async_write_ha_state() diff --git a/immich_album_watcher/translations/en.json b/immich_album_watcher/translations/en.json index 2740d38..791c4b5 100644 --- a/immich_album_watcher/translations/en.json +++ b/immich_album_watcher/translations/en.json @@ -21,6 +21,12 @@ }, "album_public_url": { "name": "{album_name}: Public URL" + }, + "album_protected_url": { + "name": "{album_name}: Protected URL" + }, + "album_protected_password": { + "name": "{album_name}: Protected Password" } }, "binary_sensor": { @@ -32,6 +38,11 @@ "album_thumbnail": { "name": "{album_name}: Thumbnail" } + }, + "text": { + "album_protected_password_edit": { + "name": "{album_name}: Share Password" + } } }, "config": { diff --git a/immich_album_watcher/translations/ru.json b/immich_album_watcher/translations/ru.json index d78b2a9..b3cdf35 100644 --- a/immich_album_watcher/translations/ru.json +++ b/immich_album_watcher/translations/ru.json @@ -21,6 +21,12 @@ }, "album_public_url": { "name": "{album_name}: Публичная ссылка" + }, + "album_protected_url": { + "name": "{album_name}: Защищённая ссылка" + }, + "album_protected_password": { + "name": "{album_name}: Пароль ссылки" } }, "binary_sensor": { @@ -32,6 +38,11 @@ "album_thumbnail": { "name": "{album_name}: Превью" } + }, + "text": { + "album_protected_password_edit": { + "name": "{album_name}: Пароль ссылки" + } } }, "config": {