alexei.dolgolyov/haos-hacs-immich-album-watcher
1
0
Fork 0
Code Issues Pull Requests Actions 1 Packages Projects Releases Wiki Activity

Add external domain support for URLs
All checks were successful
Validate / Hassfest (push) Successful in 4s
Details

Browse Source 
- Fetch externalDomain from Immich server config on startup
- Use external domain for user-facing URLs (share links, asset URLs)
- Keep internal connection URL for API calls
- Add get_internal_download_url() to convert external URLs back to
  internal for faster local network downloads (Telegram notifications)

Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>
This commit is contained in:
Admin
2026-02-01 21:53:02 +03:00
parent 0bb7e71a1e
commit 38a2a6ad7a
2 changed files with 93 additions and 25 deletions
Download Patch File Download Diff File Expand all files Collapse all files

98
custom_components/immich_album_watcher/coordinator.py
View File

@@ -330,12 +330,42 @@ class ImmichAlbumWatcherCoordinator(DataUpdateCoordinator[AlbumData | None]):
self._storage = storage self._storage = storage
self._telegram_cache = telegram_cache self._telegram_cache = telegram_cache
self._persisted_asset_ids: set[str] | None = None self._persisted_asset_ids: set[str] | None = None
self._external_domain: str | None = None # Fetched from server config
@property @property
def immich_url(self) -> str: def immich_url(self) -> str:
"""Return the Immich URL.""" """Return the Immich URL (for API calls)."""
return self._url return self._url
@property
def external_url(self) -> str:
"""Return the external URL for links.
Uses externalDomain from Immich server config if set,
otherwise falls back to the connection URL.
"""
if self._external_domain:
return self._external_domain.rstrip("/")
return self._url
def get_internal_download_url(self, url: str) -> str:
"""Convert an external URL to internal URL for faster downloads.
If the URL starts with the external domain, replace it with the
internal connection URL to download via local network.
Args:
url: The URL to convert (may be external or internal)
Returns:
URL using internal connection for downloads
"""
if self._external_domain:
external = self._external_domain.rstrip("/")
if url.startswith(external):
return url.replace(external, self._url, 1)
return url
@property @property
def api_key(self) -> str: def api_key(self) -> str:
"""Return the API key.""" """Return the API key."""
@@ -514,6 +544,36 @@ class ImmichAlbumWatcherCoordinator(DataUpdateCoordinator[AlbumData | None]):
return self._users_cache return self._users_cache
async def _async_fetch_server_config(self) -> None:
"""Fetch server config from Immich to get external domain."""
if self._session is None:
self._session = async_get_clientsession(self.hass)
headers = {"x-api-key": self._api_key}
try:
async with self._session.get(
f"{self._url}/api/server/config",
headers=headers,
) as response:
if response.status == 200:
data = await response.json()
external_domain = data.get("externalDomain", "") or ""
self._external_domain = external_domain
if external_domain:
_LOGGER.debug(
"Using external domain from Immich: %s", external_domain
)
else:
_LOGGER.debug(
"No external domain configured in Immich, using connection URL"
)
else:
_LOGGER.warning(
"Failed to fetch server config: HTTP %s", response.status
)
except aiohttp.ClientError as err:
_LOGGER.warning("Failed to fetch server config: %s", err)
async def _async_fetch_shared_links(self) -> list[SharedLinkInfo]: async def _async_fetch_shared_links(self) -> list[SharedLinkInfo]:
"""Fetch shared links for this album from Immich.""" """Fetch shared links for this album from Immich."""
if self._session is None: if self._session is None:
@@ -557,29 +617,29 @@ class ImmichAlbumWatcherCoordinator(DataUpdateCoordinator[AlbumData | None]):
"""Get the public URL if album has an accessible shared link.""" """Get the public URL if album has an accessible shared link."""
accessible_links = self._get_accessible_links() accessible_links = self._get_accessible_links()
if accessible_links: if accessible_links:
return f"{self._url}/share/{accessible_links[0].key}" return f"{self.external_url}/share/{accessible_links[0].key}"
return None return None
def get_any_url(self) -> str | None: def get_any_url(self) -> str | None:
"""Get any non-expired URL (prefers accessible, falls back to protected).""" """Get any non-expired URL (prefers accessible, falls back to protected)."""
accessible_links = self._get_accessible_links() accessible_links = self._get_accessible_links()
if accessible_links: if accessible_links:
return f"{self._url}/share/{accessible_links[0].key}" return f"{self.external_url}/share/{accessible_links[0].key}"
non_expired = [link for link in self._shared_links if not link.is_expired] non_expired = [link for link in self._shared_links if not link.is_expired]
if non_expired: if non_expired:
return f"{self._url}/share/{non_expired[0].key}" return f"{self.external_url}/share/{non_expired[0].key}"
return None return None
def get_protected_url(self) -> str | None: def get_protected_url(self) -> str | None:
"""Get a protected URL if any password-protected link exists.""" """Get a protected URL if any password-protected link exists."""
protected_links = self._get_protected_links() protected_links = self._get_protected_links()
if protected_links: if protected_links:
return f"{self._url}/share/{protected_links[0].key}" return f"{self.external_url}/share/{protected_links[0].key}"
return None return None
def get_protected_urls(self) -> list[str]: def get_protected_urls(self) -> list[str]:
"""Get all password-protected URLs.""" """Get all password-protected URLs."""
return [f"{self._url}/share/{link.key}" for link in self._get_protected_links()] return [f"{self.external_url}/share/{link.key}" for link in self._get_protected_links()]
def get_protected_password(self) -> str | None: def get_protected_password(self) -> str | None:
"""Get the password for the first protected link.""" """Get the password for the first protected link."""
@@ -590,13 +650,13 @@ class ImmichAlbumWatcherCoordinator(DataUpdateCoordinator[AlbumData | None]):
def get_public_urls(self) -> list[str]: def get_public_urls(self) -> list[str]:
"""Get all accessible public URLs.""" """Get all accessible public URLs."""
return [f"{self._url}/share/{link.key}" for link in self._get_accessible_links()] return [f"{self.external_url}/share/{link.key}" for link in self._get_accessible_links()]
def get_shared_links_info(self) -> list[dict[str, Any]]: def get_shared_links_info(self) -> list[dict[str, Any]]:
"""Get detailed info about all shared links.""" """Get detailed info about all shared links."""
return [ return [
{ {
"url": f"{self._url}/share/{link.key}", "url": f"{self.external_url}/share/{link.key}",
"has_password": link.has_password, "has_password": link.has_password,
"is_expired": link.is_expired, "is_expired": link.is_expired,
"expires_at": link.expires_at.isoformat() if link.expires_at else None, "expires_at": link.expires_at.isoformat() if link.expires_at else None,
@@ -609,40 +669,40 @@ class ImmichAlbumWatcherCoordinator(DataUpdateCoordinator[AlbumData | None]):
"""Get the public viewer URL for an asset (web page).""" """Get the public viewer URL for an asset (web page)."""
accessible_links = self._get_accessible_links() accessible_links = self._get_accessible_links()
if accessible_links: if accessible_links:
return f"{self._url}/share/{accessible_links[0].key}/photos/{asset_id}" return f"{self.external_url}/share/{accessible_links[0].key}/photos/{asset_id}"
non_expired = [link for link in self._shared_links if not link.is_expired] non_expired = [link for link in self._shared_links if not link.is_expired]
if non_expired: if non_expired:
return f"{self._url}/share/{non_expired[0].key}/photos/{asset_id}" return f"{self.external_url}/share/{non_expired[0].key}/photos/{asset_id}"
return None return None
def _get_asset_download_url(self, asset_id: str) -> str | None: def _get_asset_download_url(self, asset_id: str) -> str | None:
"""Get the direct download URL for an asset (media file).""" """Get the direct download URL for an asset (media file)."""
accessible_links = self._get_accessible_links() accessible_links = self._get_accessible_links()
if accessible_links: if accessible_links:
return f"{self._url}/api/assets/{asset_id}/original?key={accessible_links[0].key}" return f"{self.external_url}/api/assets/{asset_id}/original?key={accessible_links[0].key}"
non_expired = [link for link in self._shared_links if not link.is_expired] non_expired = [link for link in self._shared_links if not link.is_expired]
if non_expired: if non_expired:
return f"{self._url}/api/assets/{asset_id}/original?key={non_expired[0].key}" return f"{self.external_url}/api/assets/{asset_id}/original?key={non_expired[0].key}"
return None return None
def _get_asset_video_url(self, asset_id: str) -> str | None: def _get_asset_video_url(self, asset_id: str) -> str | None:
"""Get the transcoded video playback URL for a video asset.""" """Get the transcoded video playback URL for a video asset."""
accessible_links = self._get_accessible_links() accessible_links = self._get_accessible_links()
if accessible_links: if accessible_links:
return f"{self._url}/api/assets/{asset_id}/video/playback?key={accessible_links[0].key}" return f"{self.external_url}/api/assets/{asset_id}/video/playback?key={accessible_links[0].key}"
non_expired = [link for link in self._shared_links if not link.is_expired] non_expired = [link for link in self._shared_links if not link.is_expired]
if non_expired: if non_expired:
return f"{self._url}/api/assets/{asset_id}/video/playback?key={non_expired[0].key}" return f"{self.external_url}/api/assets/{asset_id}/video/playback?key={non_expired[0].key}"
return None return None
def _get_asset_photo_url(self, asset_id: str) -> str | None: def _get_asset_photo_url(self, asset_id: str) -> str | None:
"""Get the preview-sized thumbnail URL for a photo asset.""" """Get the preview-sized thumbnail URL for a photo asset."""
accessible_links = self._get_accessible_links() accessible_links = self._get_accessible_links()
if accessible_links: if accessible_links:
return f"{self._url}/api/assets/{asset_id}/thumbnail?size=preview&key={accessible_links[0].key}" return f"{self.external_url}/api/assets/{asset_id}/thumbnail?size=preview&key={accessible_links[0].key}"
non_expired = [link for link in self._shared_links if not link.is_expired] non_expired = [link for link in self._shared_links if not link.is_expired]
if non_expired: if non_expired:
return f"{self._url}/api/assets/{asset_id}/thumbnail?size=preview&key={non_expired[0].key}" return f"{self.external_url}/api/assets/{asset_id}/thumbnail?size=preview&key={non_expired[0].key}"
return None return None
def _build_asset_detail( def _build_asset_detail(
@@ -678,7 +738,7 @@ class ImmichAlbumWatcherCoordinator(DataUpdateCoordinator[AlbumData | None]):
# Add thumbnail URL if requested # Add thumbnail URL if requested
if include_thumbnail: if include_thumbnail:
asset_detail[ATTR_THUMBNAIL_URL] = f"{self._url}/api/assets/{asset.id}/thumbnail" asset_detail[ATTR_THUMBNAIL_URL] = f"{self.external_url}/api/assets/{asset.id}/thumbnail"
# Add public viewer URL (web page) # Add public viewer URL (web page)
asset_url = self._get_asset_public_url(asset.id) asset_url = self._get_asset_public_url(asset.id)
@@ -707,6 +767,10 @@ class ImmichAlbumWatcherCoordinator(DataUpdateCoordinator[AlbumData | None]):
if self._session is None: if self._session is None:
self._session = async_get_clientsession(self.hass) self._session = async_get_clientsession(self.hass)
# Fetch server config to get external domain (once)
if self._external_domain is None:
await self._async_fetch_server_config()
# Fetch users to resolve owner names # Fetch users to resolve owner names
if not self._users_cache: if not self._users_cache:
await self._async_fetch_users() await self._async_fetch_users()

20
custom_components/immich_album_watcher/sensor.py
View File

@@ -535,9 +535,10 @@ class ImmichAlbumBaseSensor(CoordinatorEntity[ImmichAlbumWatcherCoordinator], Se
_LOGGER.debug("Cached file_id request failed: %s", err) _LOGGER.debug("Cached file_id request failed: %s", err)
try: try:
# Download the photo # Download the photo using internal URL for faster local network transfer
_LOGGER.debug("Downloading photo from %s", url[:80]) download_url = self.coordinator.get_internal_download_url(url)
async with session.get(url) as resp: _LOGGER.debug("Downloading photo from %s", download_url[:80])
async with session.get(download_url) as resp:
if resp.status != 200: if resp.status != 200:
return { return {
"success": False, "success": False,
@@ -680,9 +681,10 @@ class ImmichAlbumBaseSensor(CoordinatorEntity[ImmichAlbumWatcherCoordinator], Se
_LOGGER.debug("Cached file_id request failed: %s", err) _LOGGER.debug("Cached file_id request failed: %s", err)
try: try:
# Download the video # Download the video using internal URL for faster local network transfer
_LOGGER.debug("Downloading video from %s", url[:80]) download_url = self.coordinator.get_internal_download_url(url)
async with session.get(url) as resp: _LOGGER.debug("Downloading video from %s", download_url[:80])
async with session.get(download_url) as resp:
if resp.status != 200: if resp.status != 200:
return { return {
"success": False, "success": False,
@@ -958,8 +960,10 @@ class ImmichAlbumBaseSensor(CoordinatorEntity[ImmichAlbumWatcherCoordinator], Se
continue continue
try: try:
_LOGGER.debug("Downloading media %d from %s", chunk_idx * max_group_size + i, url[:80]) # Download using internal URL for faster local network transfer
async with session.get(url) as resp: download_url = self.coordinator.get_internal_download_url(url)
_LOGGER.debug("Downloading media %d from %s", chunk_idx * max_group_size + i, download_url[:80])
async with session.get(download_url) as resp:
if resp.status != 200: if resp.status != 200:
return { return {
"success": False, "success": False,