Immich: rescan libraries after changes, open-in-Immich links for duplicate groups

README.md

@@ -204,6 +204,19 @@ import of the Immich trash (`POST /api/integrations/immich/import-trash`). Offli
 disk) are skipped. Exclude `duplicates_target_directory` from the Immich library, so moved files are not imported
 again.
 
+The group detail page links every file Immich knows to its asset in Immich ("open in Immich", looked up via
+`GET /api/groups/{id}/immich`). Local paths are translated with the inverse of `immich.path_mappings`.
+Set `immich.public_url` if the browser reaches Immich under another URL than `immich.url` (f.ex. a cluster
+internal service URL). Finding assets by path uses the structured search filter of Immich v3.2.0 and newer,
+older versions fall back to the deprecated `originalPath` search parameter.
+
+With `immich.rescan_after_changes: true`, py-image-dedup queues a scan of the Immich external libraries
+containing the affected files (of all external libraries, if none matches) after it really moved or deleted
+files: at most once per deduplication run and once per batch of executed web UI decisions, never in dry run.
+The scan runs in the background; if Immich is unreachable, only a warning is logged. Library scans are admin
+endpoints in Immich: the API key must belong to an Immich admin and needs the `library.read` and `library.update`
+permissions in addition to `asset.read`.
+
 With `deduplication.hold_ambiguous_groups` enabled, groups that need a review are not acted on automatically.
 
 The state shown in the web UI (runs, groups, decisions, action log) is stored in a small SQLite database

py_image_dedup/config.py

@@ -68,6 +68,8 @@
 NODE_URL = "url"
 NODE_API_KEY = "api_key"
 NODE_PATH_MAPPINGS = "path_mappings"
+NODE_PUBLIC_URL = "public_url"
+NODE_RESCAN_AFTER_CHANGES = "rescan_after_changes"
 NODE_STATIC_DIRECTORY = "static_directory"
 
 DEFAULT_STATE_DATABASE_PATH = "~/.local/share/py-image-dedup/state.db"
@@ -470,6 +472,32 @@ def __new__(cls, *args, **kwargs):
         example=[{"immich": "/mnt/photos", "local": "/data/photos"}]
     )
 
+    IMMICH_PUBLIC_URL = StringConfigEntry(
+        description="URL of the Immich server as reachable from the browser, used for the \"open in Immich\" "
+                    "links in the web UI. Falls back to immich.url.",
+        key_path=[
+            NODE_MAIN,
+            NODE_IMMICH,
+            NODE_PUBLIC_URL
+        ],
+        required=False,
+        default=None,
+        example="https://immich.example.com"
+    )
+
+    IMMICH_RESCAN_AFTER_CHANGES = BoolConfigEntry(
+        description="Queue a scan of the Immich external libraries after files were really moved or deleted "
+                    "(at most once per run / batch of executed decisions), so Immich notices the change. "
+                    "Needs an API key of an Immich admin with the library.read and library.update permissions.",
+        key_path=[
+            NODE_MAIN,
+            NODE_IMMICH,
+            NODE_RESCAN_AFTER_CHANGES
+        ],
+        required=False,
+        default=False
+    )
+
     @property
     def user_decisions_dry_run(self) -> bool:
         """

py_image_dedup/integrations/immich.py

@@ -1,5 +1,5 @@
 import logging
-from typing import List
+from typing import Dict, List
 
 import requests
 
@@ -10,7 +10,10 @@
 
 
 class ImmichError(Exception):
-    pass
+
+    def __init__(self, message: str, status_code: int | None = None):
+        super().__init__(message)
+        self.status_code = status_code
 
 
 class ImmichClient:
@@ -53,31 +56,104 @@ def trashed_original_paths(self) -> List[str]:
         :return: original file paths (as reported by Immich) of all assets in the trash
                  of the API key's user, excluding offline assets (files missing on disk)
         """
-        paths = []
+        items = self._search_items({"withDeleted": True, "trashedAfter": _EPOCH})
+        return [item["originalPath"] for item in items if item.get("isTrashed") and not item.get("isOffline")]
+
+    def to_immich_path(self, path: str) -> str | None:
+        """
+        Maps a local path to the path Immich knows the file under (inverse of map_path).
+        :return: the Immich path, None if there are mappings and none of them matches
+        """
+        if not self._path_mappings:
+            return path
+        for immich_prefix, local_prefix in self._path_mappings:
+            if path == local_prefix or path.startswith(local_prefix + "/"):
+                return immich_prefix + path[len(local_prefix):]
+        return None
+
+    def find_asset_ids(self, immich_paths: List[str]) -> Dict[str, str]:
+        """
+        Looks up the (not trashed) assets of the API key's user with the given original paths.
+        Needs the asset.read permission.
+
+        :param immich_paths: original paths as Immich knows them
+        :return: original path -> asset id, paths without asset are missing
+        """
+        paths = sorted(set(immich_paths))
+        if not paths:
+            return {}
+        try:
+            # structured filter, Immich >= v3.2.0
+            items = self._search_items({"filter": {"originalPath": {"in": paths}}})
+        except ImmichError as ex:
+            if ex.status_code not in (400, 422):
+                raise
+            # older Immich: the deprecated originalPath parameter matches substrings, one request per path
+            items = []
+            for path in paths:
+                items.extend(self._search_items({"originalPath": path}))
+        wanted = set(paths)
+        return {item["originalPath"]: item["id"] for item in items if item.get("originalPath") in wanted}
+
+    def _search_items(self, body: dict) -> List[dict]:
+        items = []
         page = 1
         while page is not None:
-            assets = self._post("/api/search/metadata", {
-                "withDeleted": True,
-                "trashedAfter": _EPOCH,
-                "page": page,
-                "size": self._page_size,
-            })["assets"]
-            for asset in assets["items"]:
-                if asset.get("isTrashed") and not asset.get("isOffline"):
-                    paths.append(asset["originalPath"])
+            assets = self._post("/api/search/metadata", {**body, "page": page, "size": self._page_size},
+                                permission="asset.read")["assets"]
+            items.extend(assets["items"])
             next_page = assets.get("nextPage")
             page = int(next_page) if next_page else None
-        return paths
+        return items
+
+    def libraries(self) -> List[dict]:
+        """
+        :return: all external libraries (needs an admin API key with the library.read permission)
+        """
+        return self._request("GET", "/api/libraries", permission="library.read (admin)")
+
+    def scan_library(self, library_id: str):
+        """
+        Queues a scan of an external library (needs an admin API key with the library.update permission).
+        """
+        self._request("POST", f"/api/libraries/{library_id}/scan", permission="library.update (admin)")
+
+    def rescan_libraries(self, changed_immich_paths: List[str]) -> List[str]:
+        """
+        Queues a scan of the external libraries containing one of the given paths,
+        of all external libraries if none of them matches (f.ex. because the path mappings are incomplete).
+
+        :param changed_immich_paths: paths (as Immich knows them) of files that were moved or deleted
+        :return: ids of the libraries a scan was queued for
+        """
+        libraries = self.libraries()
+
+        def contains(library: dict, path: str) -> bool:
+            return any(path == p.rstrip("/") or path.startswith(p.rstrip("/") + "/")
+                       for p in library.get("importPaths") or [])
+
+        affected = [lib for lib in libraries if any(contains(lib, p) for p in changed_immich_paths)]
+        if not affected:
+            affected = libraries
+        for library in affected:
+            self.scan_library(library["id"])
+        return [lib["id"] for lib in affected]
+
+    def _post(self, path: str, body: dict, permission: str = "asset.read") -> dict:
+        return self._request("POST", path, body=body, permission=permission)
 
-    def _post(self, path: str, body: dict) -> dict:
+    def _request(self, method: str, path: str, body: dict | None = None, permission: str = "asset.read"):
         try:
-            response = self._session.post(self._url + path, json=body, timeout=self._timeout,
-                                          headers={"x-api-key": self._api_key, "Accept": "application/json"})
+            response = self._session.request(method, self._url + path, json=body, timeout=self._timeout,
+                                             headers={"x-api-key": self._api_key, "Accept": "application/json"})
         except requests.RequestException as ex:
             raise ImmichError(f"Immich is not reachable at {self._url}: {ex}") from ex
         if response.status_code in (401, 403):
             raise ImmichError(f"Immich rejected the API key (HTTP {response.status_code}), "
-                              f"it needs the asset.read permission")
+                              f"it needs the {permission} permission", status_code=response.status_code)
         if not response.ok:
-            raise ImmichError(f"Immich request {path} failed: HTTP {response.status_code} {response.text[:200]}")
+            raise ImmichError(f"Immich request {path} failed: HTTP {response.status_code} {response.text[:200]}",
+                              status_code=response.status_code)
+        if response.status_code == 204 or not response.content:
+            return None
         return response.json()

py_image_dedup/library/deduplicator.py

@@ -15,6 +15,7 @@
 
 from py_image_dedup import util
 from py_image_dedup.config import DeduplicatorConfig
+from py_image_dedup.integrations.immich import ImmichClient
 from py_image_dedup.library import ActionEnum
 from py_image_dedup.library.deduplication_result import DeduplicationResult
 from py_image_dedup.library.progress_manager import ProgressManager
@@ -114,6 +115,8 @@ def __init__(self, interactive: bool, state_store=None):
         self._run_id = None
         # held while a run is in progress, so user decisions from the web UI are executed in between runs
         self.work_lock = threading.RLock()
+        # files really moved/deleted since the last Immich library rescan was triggered
+        self._changed_paths = []
         self.reset_result()
 
         listener = StateProgressListener(state_store, lambda: self._run_id) if state_store is not None else None
@@ -904,6 +907,42 @@ def process_duplicates(self):
             except Exception as ex:
                 LOGGER.exception(ex)
 
+        self._trigger_immich_rescan()
+
+    def _trigger_immich_rescan(self):
+        """
+        Queues a scan of the Immich external libraries (if enabled) when files were really moved or deleted
+        since the last call. Runs in the background and never raises: Immich being unreachable must not
+        affect deduplication.
+        """
+        changed_paths, self._changed_paths = self._changed_paths, []
+        if not changed_paths or not self._config.IMMICH_RESCAN_AFTER_CHANGES.value:
+            return
+        url = self._config.IMMICH_URL.value
+        api_key = self._config.IMMICH_API_KEY.value
+        if not (url and api_key):
+            LOGGER.warning("immich.rescan_after_changes is enabled, but immich.url or immich.api_key is missing")
+            return
+        try:
+            client = ImmichClient(url, api_key, path_mappings=self._config.IMMICH_PATH_MAPPINGS.value, timeout=10)
+        except Exception as ex:
+            LOGGER.warning(f"Could not trigger an Immich library rescan: {ex}")
+            return
+        immich_paths = [p for p in (client.to_immich_path(p) for p in changed_paths) if p is not None]
+        self._run_in_background(lambda: self._rescan_immich_libraries(client, immich_paths))
+
+    @staticmethod
+    def _rescan_immich_libraries(client: ImmichClient, immich_paths: List[str]):
+        try:
+            library_ids = client.rescan_libraries(immich_paths)
+            LOGGER.info(f"Queued a scan of {len(library_ids)} Immich external library(ies)")
+        except Exception as ex:
+            LOGGER.warning(f"Could not trigger an Immich library rescan: {ex}")
+
+    @staticmethod
+    def _run_in_background(function):
+        threading.Thread(target=function, name="immich-rescan").start()
+
     def execute_user_decisions(self) -> int:
         """
         Executes the decisions users made in the web UI (chosen keeper of a duplicate group).
@@ -926,6 +965,7 @@ def execute_user_decisions(self) -> int:
                                               status=ActionStatus.FAILED, message=f"{type(ex).__name__}: {ex}",
                                               group_id=group["id"])
                     self._state.mark_executed(group["id"], resolved=False)
+            self._trigger_immich_rescan()
             return len(groups)
 
     def _execute_user_decision(self, group: dict):
@@ -970,6 +1010,8 @@ def _record_action(self, path: Path, action: str, status: str, dry_run: bool, so
         """
         Writes a file action to the action log of the state store (if any). Never raises.
         """
+        if status == ActionStatus.DONE and not dry_run and action in ("move", "delete"):
+            self._changed_paths.append(str(path))
         group_id = self._group_of_path.get(str(path))
         if status == ActionStatus.FAILED and group_id is not None:
             self._failed_groups.add(group_id)

py_image_dedup/web/app.py

@@ -21,6 +21,10 @@
 
 LOGGER = logging.getLogger(__name__)
 
+# how long asset lookups in Immich are cached (also failed ones, so a down Immich is not asked on every page load)
+IMMICH_ASSET_CACHE_SECONDS = 60
+IMMICH_ASSET_CACHE_SIZE = 500
+
 # keep in sync with ImageMatchDeduplicator._quality_criteria and WEAK_PRIORITIZATION_RULES
 KNOWN_PRIORITIZATION_RULES = {
     "more-exif-data", "less-exif-data",
@@ -238,10 +242,60 @@
                                 detail="duplicates_target_directory is not configured. User decisions "
                                        "only move duplicates, they never delete files.")
 
+    def immich_configured() -> bool:
+        return bool(config.IMMICH_URL.value and config.IMMICH_API_KEY.value)
+
+    def immich_public_url() -> str | None:
+        url = config.IMMICH_PUBLIC_URL.value or config.IMMICH_URL.value
+        return url.rstrip("/") if url else None
+
     @app.get("/api/integrations/immich")
     def immich_status():
-        return {"configured": bool(config.IMMICH_URL.value and config.IMMICH_API_KEY.value),
-                "url": config.IMMICH_URL.value}
+        return {"configured": immich_configured(),
+                "url": config.IMMICH_URL.value,
+                "public_url": immich_public_url(),
+                "rescan_after_changes": config.IMMICH_RESCAN_AFTER_CHANGES.value}
+
+    immich_asset_cache = {}
+
+    @app.get("/api/groups/{group_id}/immich")
+    def group_immich_assets(group_id: int):
+        """
+        Finds the Immich assets of the files of a group, for "open in Immich" links.
+        Never fails because of Immich: errors are reported in "error". Cached briefly.
+        """
+        group = group_or_404(group_id)
+        if not immich_configured():
+            return {"configured": False, "url": None, "error": None, "candidates": []}
+
+        client = ImmichClient(config.IMMICH_URL.value, config.IMMICH_API_KEY.value,
+                              path_mappings=config.IMMICH_PATH_MAPPINGS.value, timeout=5)
+        immich_paths = {c["id"]: client.to_immich_path(c["path"]) for c in group["candidates"]}
+        cache_key = tuple(sorted(p for p in immich_paths.values() if p is not None))
+        cached = immich_asset_cache.get(cache_key)
+        if cached is not None and time.monotonic() - cached[0] < IMMICH_ASSET_CACHE_SECONDS:
+            asset_ids, error = cached[1], cached[2]
+        else:
+            asset_ids, error = {}, None
+            try:
+                asset_ids = client.find_asset_ids(list(cache_key))
+            except Exception as ex:
+                LOGGER.warning(f"Could not look up the Immich assets of group {group_id}: {ex}")
+                error = str(ex)
+            if len(immich_asset_cache) >= IMMICH_ASSET_CACHE_SIZE:
+                immich_asset_cache.clear()
+            immich_asset_cache[cache_key] = (time.monotonic(), asset_ids, error)
+
+        base_url = immich_public_url()
+        candidates = []
+        for candidate_id, immich_path in immich_paths.items():
+            asset_id = asset_ids.get(immich_path) if immich_path is not None else None
+            candidates.append({
+                "candidate_id": candidate_id,
+                "asset_id": asset_id,
+                "url": f"{base_url}/photos/{asset_id}" if asset_id is not None else None,
+            })
+        return {"configured": True, "url": base_url, "error": error, "candidates": candidates}
 
     @app.post("/api/integrations/immich/import-trash")
     def immich_import_trash(request: ImportRequest):

py_image_dedup_reference.yaml

@@ -132,13 +132,22 @@ py_image_dedup:
     # static_directory: /app/web/dist
 
   # Optional Immich integration: import the files trashed in Immich (f.ex. by
-  # its duplicate review) as decisions for the duplicate groups (web UI dashboard).
+  # its duplicate review) as decisions for the duplicate groups (web UI dashboard),
+  # "open in Immich" links in the group detail and library rescans after changes.
   immich:
     # Base URL of the Immich server.
     # url: http://immich-server:2283
     # API key with the asset.read permission of the user who owns the trashed
     # assets. Better set via env: PY_IMAGE_DEDUP_IMMICH_API_KEY
     # api_key: ...
+    # URL of Immich as reachable from the browser, for the "open in Immich" links
+    # in the web UI. Falls back to url.
+    # public_url: https://immich.example.com
+    # Queue a scan of the Immich external libraries after files were really
+    # moved or deleted (at most once per run / batch of executed decisions).
+    # Needs an API key of an Immich admin with the library.read and
+    # library.update permissions.
+    rescan_after_changes: false
     # Maps the paths Immich reports to the paths this application sees,
     # f.ex. when Immich mounts the library at a different location.
     # Only paths matching a mapping are imported (if any mapping is set).