108 lines
3.6 KiB
Python
108 lines
3.6 KiB
Python
"""immich-go adapter: binary discovery, version, and redacted command preview.
|
|
|
|
Upload is the only stage that needs credentials, so this module is also the single
|
|
place that knows an API key exists. It never returns, logs, or renders the secret:
|
|
:func:`build_command` produces the real argument list for the uploader, and
|
|
:func:`redact` produces the copy that is safe for the API, the browser, and the
|
|
activity log. Both come from the same builder so the preview can never drift from
|
|
the command that would actually run.
|
|
|
|
Server reachability uses ``/api/server/ping`` through stdlib ``urllib`` — the app
|
|
has no HTTP client dependency and this is one request.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
import shutil
|
|
import subprocess
|
|
import urllib.error
|
|
import urllib.request
|
|
from pathlib import Path
|
|
|
|
REDACTED = "***"
|
|
PING_PATH = "/api/server/ping"
|
|
PING_TIMEOUT_SECONDS = 5.0
|
|
|
|
|
|
def find_binary(binary: str = "immich-go") -> str | None:
|
|
"""Absolute path of the uploader, or ``None`` when it is not installed."""
|
|
return shutil.which(binary)
|
|
|
|
|
|
def version(binary: str = "immich-go") -> str | None:
|
|
"""Reported uploader version, or ``None`` when it is missing or unusable.
|
|
|
|
The version is persisted with every batch (concept §8) because immich-go's
|
|
flags and report text change between releases.
|
|
"""
|
|
path = find_binary(binary)
|
|
if path is None:
|
|
return None
|
|
try:
|
|
result = subprocess.run([path, "--version"], capture_output=True, text=True, timeout=30)
|
|
except (OSError, subprocess.SubprocessError):
|
|
return None
|
|
output = (result.stdout or result.stderr or "").strip()
|
|
return output.splitlines()[0].strip() if output else None
|
|
|
|
|
|
def ping(server_url: str, *, timeout: float = PING_TIMEOUT_SECONDS) -> tuple[bool, str | None]:
|
|
"""``(reachable, detail)`` for the configured Immich server.
|
|
|
|
A reachable Immich answers ``{"res": "pong"}``. Anything else — wrong host, no
|
|
Immich, HTTP error — is a blocker with a short human detail. The detail never
|
|
carries the URL's credentials because the key travels in a header, not the URL.
|
|
"""
|
|
if not server_url:
|
|
return False, "no server URL configured"
|
|
url = server_url.rstrip("/") + PING_PATH
|
|
try:
|
|
with urllib.request.urlopen(url, timeout=timeout) as response: # noqa: S310
|
|
payload = json.loads(response.read().decode("utf-8") or "{}")
|
|
except (urllib.error.URLError, OSError, ValueError, TimeoutError) as error:
|
|
return False, f"{type(error).__name__}: {error}"
|
|
if payload.get("res") == "pong":
|
|
return True, None
|
|
return False, "server did not answer with pong"
|
|
|
|
|
|
def build_command(
|
|
*,
|
|
binary: str,
|
|
server_url: str,
|
|
api_key: str,
|
|
album_name: str,
|
|
folder: Path | str,
|
|
) -> list[str]:
|
|
"""The exact upload invocation for one folder-as-album batch."""
|
|
return [
|
|
binary,
|
|
"upload",
|
|
"from-folder",
|
|
f"--server={server_url}",
|
|
f"--api-key={api_key}",
|
|
f"--album-name={album_name}",
|
|
str(folder),
|
|
]
|
|
|
|
|
|
def redact(command: list[str]) -> list[str]:
|
|
"""The same command with every secret-bearing argument masked."""
|
|
return [f"--api-key={REDACTED}" if arg.startswith("--api-key=") else arg for arg in command]
|
|
|
|
|
|
def preview_command(
|
|
*, binary: str, server_url: str, album_name: str, folder: Path | str
|
|
) -> list[str]:
|
|
"""Redacted preview built without ever handling the real key."""
|
|
return redact(
|
|
build_command(
|
|
binary=binary,
|
|
server_url=server_url,
|
|
api_key=REDACTED,
|
|
album_name=album_name,
|
|
folder=folder,
|
|
)
|
|
)
|