Save photos locally if Immich API not configured

app/app.py

@@ -344,6 +344,8 @@ def immich_ping() -> bool:
 
 def immich_bulk_check(checks: List[dict]) -> Dict[str, dict]:
     """Try Immich bulk upload check; return map id->result (or empty on failure)."""
+    if SETTINGS.local_save_only:
+        return {}
     try:
         url = f"{SETTINGS.normalized_base_url}/assets/bulk-upload-check"
         r = requests.post(url, headers=immich_headers(), json={"assets": checks}, timeout=10)
@@ -397,6 +399,8 @@ async def favicon() -> Response:
 @app.post("/api/ping")
 async def api_ping() -> dict:
     """Connectivity test endpoint used by the UI to display a temporary banner."""
+    if SETTINGS.local_save_only:
+        return { "ok": True, "base_url": "Local Save Mode", "album_name": None }
     return {
         "ok": immich_ping(), 
         "base_url": SETTINGS.normalized_base_url,
@@ -473,6 +477,29 @@ async def api_upload(
         await send_progress(session_id, item_id, "duplicate", 100, "Already uploaded from this device (local cache)")
         return JSONResponse({"status": "duplicate", "id": None}, status_code=200)
 
+    if SETTINGS.local_save_only:
+        try:
+            save_dir = "./data/uploads"
+            os.makedirs(save_dir, exist_ok=True)
+            safe_name = sanitize_filename(file.filename)
+            save_path = os.path.join(save_dir, safe_name)
+            # Avoid overwriting when filenames collide (not same as duplicate)
+            if os.path.exists(save_path):
+                base, ext = os.path.splitext(safe_name)
+                i = 1
+                while os.path.exists(save_path):
+                    save_path = os.path.join(save_dir, f"{base}_{i}{ext}")
+                    i += 1
+            with open(save_path, "wb") as f:
+                f.write(raw)
+            db_insert_upload(checksum, file.filename, size, device_asset_id, None, created_iso)
+            await send_progress(session_id, item_id, "done", 100, f"Saved locally to {os.path.basename(save_path)}")
+            return JSONResponse({"status": "done", "id": None}, status_code=200)
+        except Exception as e:
+            logger.exception("Local save failed: %s", e)
+            await send_progress(session_id, item_id, "error", 100, "Failed to save file locally")
+            return JSONResponse({"error": "local_save_failed"}, status_code=500)
+
     await send_progress(session_id, item_id, "checking", 2, "Checking duplicates…")
     bulk = immich_bulk_check([{"id": item_id, "checksum": checksum}])
     if bulk.get(item_id, {}).get("action") == "reject" and bulk[item_id].get("reason") == "duplicate":
@@ -853,6 +880,29 @@ async def api_upload_chunk_complete(request: Request) -> JSONResponse:
         await send_progress(session_id_local, item_id_local, "duplicate", 100, "Already uploaded from this device (local cache)")
         return JSONResponse({"status": "duplicate", "id": None}, status_code=200)
 
+    if SETTINGS.local_save_only:
+        try:
+            save_dir = "./data/uploads"
+            os.makedirs(save_dir, exist_ok=True)
+            safe_name = sanitize_filename(file_like_name)
+            save_path = os.path.join(save_dir, safe_name)
+            # Avoid overwriting when filenames collide (not same as duplicate)
+            if os.path.exists(save_path):
+                base, ext = os.path.splitext(safe_name)
+                i = 1
+                while os.path.exists(save_path):
+                    save_path = os.path.join(save_dir, f"{base}_{i}{ext}")
+                    i += 1
+            with open(save_path, "wb") as f:
+                f.write(raw)
+            db_insert_upload(checksum, file_like_name, file_size, device_asset_id, None, created_iso)
+            await send_progress(session_id_local, item_id_local, "done", 100, f"Saved locally to {os.path.basename(save_path)}")
+            return JSONResponse({"status": "done", "id": None}, status_code=200)
+        except Exception as e:
+            logger.exception("Local save failed: %s", e)
+            await send_progress(session_id_local, item_id_local, "error", 100, "Failed to save file locally")
+            return JSONResponse({"error": "local_save_failed"}, status_code=500)
+
     await send_progress(session_id_local, item_id_local, "checking", 2, "Checking duplicates…")
     bulk = immich_bulk_check([{ "id": item_id_local, "checksum": checksum }])
     if bulk.get(item_id_local, {}).get("action") == "reject" and bulk[item_id_local].get("reason") == "duplicate":

app/config.py

@@ -30,6 +30,11 @@ class Settings:
         """Return the base URL without a trailing slash for clean joining and display."""
         return self.immich_base_url.rstrip("/")
 
+    @property
+    def local_save_only(self) -> bool:
+        """True if configured to save locally instead of uploading to Immich."""
+        return str(self.immich_base_url).lower() == "false"
+
 def load_settings() -> Settings:
     """Load settings from .env, applying defaults when absent."""
     # Load environment variables from .env once here so importers don’t have to