86 lines
2.6 KiB
Python
86 lines
2.6 KiB
Python
"""photo-pipeline: Immich import flow (v4).
|
|
|
|
Uploads sorted/staged photos to Immich (on .35) using the official @immich/cli.
|
|
|
|
Security: the API key comes from env IMMICH_API_KEY or ~/photo-pipeline/.immich-key
|
|
(chmod 600). Never committed to git.
|
|
|
|
The CLI does checksum dedup natively (skips already-uploaded assets), so
|
|
re-running is safe.
|
|
"""
|
|
|
|
import os
|
|
import shutil
|
|
import subprocess
|
|
from pathlib import Path
|
|
|
|
from prefect import flow, task
|
|
|
|
IMMICH_URL = os.environ.get("IMMICH_URL", "http://192.168.20.35:2283")
|
|
KEY_FILE = Path(__file__).parent / ".immich-key"
|
|
CLI = Path(__file__).parent / "node_modules/.bin/immich"
|
|
|
|
|
|
def _api_key() -> str:
|
|
key = os.environ.get("IMMICH_API_KEY")
|
|
if key:
|
|
return key
|
|
if KEY_FILE.exists():
|
|
return KEY_FILE.read_text().strip()
|
|
raise RuntimeError("IMMICH_API_KEY env or ~/photo-pipeline/.immich-key required")
|
|
|
|
|
|
@task
|
|
def immich_login():
|
|
"""Ensure CLI is authenticated (login-key is idempotent-ish; stores auth.yml)."""
|
|
key = _api_key()
|
|
r = subprocess.run(
|
|
[str(CLI), "login-key", IMMICH_URL, key],
|
|
capture_output=True, text=True,
|
|
)
|
|
print(f"immich login: rc={r.returncode} {r.stdout.strip()[:200]}")
|
|
if r.returncode != 0:
|
|
raise RuntimeError(f"immich login failed: {r.stderr[-500:]}")
|
|
return True
|
|
|
|
|
|
@task
|
|
def immich_upload(src_dir: str, album: str | None = None, dry_run: bool = False) -> str:
|
|
"""Upload assets from a directory to Immich.
|
|
|
|
Recursive, album-per-folder optional, dry-run supported. Immich dedups by checksum.
|
|
"""
|
|
cmd = [str(CLI), "upload", src_dir, "--recursive", "--skip-hash"]
|
|
# --skip-hash: with a least-privilege API key the CLI's dup-check is broken
|
|
# (immich-app issue #21456); our fingerprint DB does dedup pre-upload.
|
|
if album:
|
|
cmd += ["--album-name", album]
|
|
if dry_run:
|
|
cmd += ["--dry-run"]
|
|
r = subprocess.run(cmd, capture_output=True, text=True)
|
|
out = r.stdout.strip()
|
|
print(f"immich upload rc={r.returncode}")
|
|
print(out[-2000:])
|
|
if r.returncode != 0:
|
|
raise RuntimeError(f"immich upload failed: {r.stderr[-500:]}")
|
|
return out
|
|
|
|
|
|
@flow(name="immich-import")
|
|
def immich_import(
|
|
src_dir: str,
|
|
album: str | None = None,
|
|
dry_run: bool = True,
|
|
):
|
|
"""Login to Immich and upload a staged folder. Dry-run by default — safe."""
|
|
immich_login()
|
|
result = immich_upload(src_dir, album=album, dry_run=dry_run)
|
|
return {"uploaded": result}
|
|
|
|
|
|
if __name__ == "__main__":
|
|
import sys
|
|
src = sys.argv[1] if len(sys.argv) > 1 else "/mnt/data/01_keep"
|
|
dry = "--dry-run" in sys.argv or "--no-move" not in sys.argv
|
|
immich_import(src, dry_run=dry)
|