Add an immich client that can upload as well as annotate

Carry over immich-sync's client, which lists assets by checksum and drives
albums and tags, and add a hand-rolled multipart upload with the checksum
header so immich can answer duplicate without receiving the bytes twice,
plus a per-asset update for description, date and coordinates.

Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>

rgl-photo-backup

@@ -919,5 +919,215 @@ def fetchphase(conf, flickrkey):
     return failures
 
 
+# immich identifies a file by the base64 of its SHA1
+def checksum(path):
+    digest = hashlib.sha1()
+    with open(path, "rb") as fh:
+        while True:
+            block = fh.read(CHUNK)
+            if not block:
+                break
+            digest.update(block)
+    return base64.b64encode(digest.digest()).decode("ascii")
+
+#
+# Hashing the whole backup means reading every byte of it. Remember the
+# answers, keyed on size and modification time so an edited file is spotted.
+#
+class Hashes:
+    def __init__(self, path):
+        self.path = path
+        self.dirty = False
+        try:
+            with open(path, encoding="utf-8") as fh:
+                self.cache = json.load(fh)
+        except (OSError, ValueError):
+            self.cache = {}
+
+    def of(self, path, key, info):
+        stamp = [info.st_size, info.st_mtime_ns]
+        held = self.cache.get(key)
+        if held and held[:2] == stamp:
+            return held[2]
+        value = checksum(path)
+        self.cache[key] = stamp + [value]
+        self.dirty = True
+        return value
+
+    def save(self):
+        if not self.dirty:
+            return
+        writejson(self.path, self.cache)
+        self.dirty = False
+
+
+class Immich:
+    def __init__(self, url, key):
+        self.url = url
+        self.key = key
+        self.calls = 0
+
+    def call(self, method, path, body=None, raw=None, contenttype=None):
+        headers = {"x-api-key": self.key, "Accept": "application/json"}
+        if raw is not None:
+            data = raw
+            headers["Content-Type"] = contenttype
+        elif body is not None:
+            data = json.dumps(body).encode("utf-8")
+            headers["Content-Type"] = "application/json"
+        else:
+            data = None
+        self.calls += 1
+        try:
+            answer = fetch(self.url + path, "immich %s %s" % (method, path), headers, data, method)
+        except HTTPFailure as e:
+            raise BackupError(str(e))
+        if not answer:
+            return None
+        # A proxy in front of immich can answer 200 with an HTML page
+        try:
+            return json.loads(answer)
+        except ValueError as e:
+            raise BackupError("immich %s %s: response was not json: %s" % (method, path, e))
+
+    def me(self):
+        body = self.call("GET", "/users/me") or {}
+        if "id" not in body:
+            raise BackupError("immich /users/me: no user id in the response")
+        return body
+
+    #
+    # Every asset this key's user owns, keyed by checksum, with the fields
+    # the sync phase compares against. One pass of a thousand at a time is far
+    # cheaper than asking after each photo.
+    #
+    # Archived and trashed assets are left out of a metadata search by
+    # default, but not out of immich's duplicate detection, so without this an
+    # archived asset would be uploaded again and annotated as if it were new.
+    def assets(self, owner, progress=None):
+        found = {}
+        page = 1
+        while page:
+            body = self.call("POST", "/search/metadata",
+                             {"size": PAGE, "page": page, "withExif": True,
+                              "withArchived": True})
+            result = body.get("assets", {})
+            for asset in result.get("items", []):
+                if asset.get("ownerId") != owner:
+                    continue
+                found[asset["checksum"]] = self.state(asset)
+            if progress:
+                progress(len(found))
+            page = result.get("nextPage")
+            page = int(page) if page else None
+        return found
+
+    @staticmethod
+    def state(asset):
+        exif = asset.get("exifInfo") or {}
+        return {
+            "id":          asset.get("id"),
+            "name":        asset.get("originalFileName", ""),
+            "description": exif.get("description") or "",
+            "latitude":    exif.get("latitude"),
+            "longitude":   exif.get("longitude"),
+            "taken":       exif.get("dateTimeOriginal") or "",
+        }
+
+    # One asset as it stands now, for planning against real state rather than
+    # against what an upload response implies.
+    def asset(self, assetid):
+        return self.state(self.call("GET", "/assets/" + assetid) or {})
+
+    def upload(self, path, digest, key, created, modified):
+        boundary = "----rgl-photo-backup-" + uuid.uuid4().hex
+        fields = [("deviceAssetId", key), ("deviceId", DEVICE_ID),
+                  ("fileCreatedAt", created), ("fileModifiedAt", modified),
+                  ("isFavorite", "false")]
+        parts = []
+        for name, value in fields:
+            parts.append(("--%s\r\nContent-Disposition: form-data; name=\"%s\"\r\n\r\n%s\r\n"
+                          % (boundary, name, value)).encode("utf-8"))
+        with open(path, "rb") as fh:
+            content = fh.read()
+        # strip characters that could break out of the quoted filename or add a header line
+        safename = os.path.basename(path).replace('"', "").replace("\r", "").replace("\n", "")
+        parts.append(("--%s\r\nContent-Disposition: form-data; name=\"assetData\"; "
+                      "filename=\"%s\"\r\nContent-Type: application/octet-stream\r\n\r\n"
+                      % (boundary, safename)).encode("utf-8"))
+        parts.append(content)
+        parts.append(("\r\n--%s--\r\n" % boundary).encode("utf-8"))
+        body = b"".join(parts)
+        headers = {"x-api-key": self.key, "Accept": "application/json",
+                   "Content-Type": "multipart/form-data; boundary=" + boundary,
+                   "x-immich-checksum": digest}
+        self.calls += 1
+        try:
+            answer = fetch(self.url + "/assets", "immich upload %s" % os.path.basename(path),
+                           headers, body, "POST")
+        except HTTPFailure as e:
+            raise BackupError(str(e))
+        try:
+            result = json.loads(answer) if answer else {}
+        except ValueError:
+            result = {}
+        if "id" not in result:
+            raise BackupError("immich upload %s: bad response" % os.path.basename(path))
+        return (result["id"], result.get("status", "created"))
+
+    def update(self, assetid, **fields):
+        self.call("PUT", "/assets/" + assetid, fields)
+
+    def createalbum(self, name, description=""):
+        body = self.call("POST", "/albums",
+                         {"albumName": name, "description": description}) or {}
+        if "id" not in body:
+            raise BackupError("immich: creating album %s returned no id" % name)
+        return body["id"]
+
+    def albumassets(self, albumid):
+        album = self.call("GET", "/albums/%s" % albumid) or {}
+        return {a["id"] for a in album.get("assets", [])}
+
+    def addtoalbum(self, albumid, ids):
+        added = 0
+        for start in range(0, len(ids), BATCH):
+            results = self.call("PUT", "/albums/%s/assets" % albumid,
+                                {"ids": ids[start:start + BATCH]}) or []
+            added += sum(1 for r in results if r.get("success"))
+        return added
+
+    # Albums shared with this user are listed too, and one of those could
+    # carry the RGL name, so only the ones this user owns count.
+    def albums(self, owner):
+        found = {}
+        for album in self.call("GET", "/albums") or []:
+            name, albumid = album.get("albumName"), album.get("id")
+            if name and albumid and album.get("ownerId") == owner:
+                found[name] = albumid
+        return found
+
+    # Upsert tag names and get their ids back; a slash makes a hierarchy.
+    # Skip any entry missing a value or id rather than raise on a malformed one.
+    def tagids(self, names):
+        ids = {}
+        for t in self.call("PUT", "/tags", {"tags": names}) or []:
+            value, tagid = t.get("value"), t.get("id")
+            if value and tagid:
+                ids[value] = tagid
+        return ids
+
+    # immich reports an asset already carrying the tag as a duplicate, which
+    # is what a second run should see and not a failure
+    def tagassets(self, tagid, ids):
+        tagged = already = 0
+        for start in range(0, len(ids), BATCH):
+            results = self.call("PUT", "/tags/%s/assets" % tagid,
+                                {"ids": ids[start:start + BATCH]}) or []
+            tagged += sum(1 for r in results if r.get("success"))
+            already += sum(1 for r in results if r.get("error") == "duplicate")
+        return (tagged, already)
+
+
 if __name__ == "__main__":
     sys.exit(0)

test_rgl_photo_backup.py

@@ -429,6 +429,17 @@ class Retrying(TestCase):
         rb.OPEN = FakeOpen(FakeResponse({"stat": "ok"}))
         rb.flickrcall("flickr.test.echo", {}, "KEY")
         self.assertEqual(waits, [2.0])
+
+    def test_pace_leaves_immich_and_the_index_alone(self):
+        waits = []
+        rb.SLEEP = waits.append
+        rb.PACE = 2.0
+        rb.OPEN = FakeOpen(FakeResponse({"id": "u1"}),
+                           FakeResponse(json.dumps([index_entry("A", "")]).encode()))
+        rb.Immich("https://i/api", "k").me()
+        rb.loadindex("http://t/index")
+        self.assertEqual(waits, [])
+
     def test_download_refuses_an_oversized_body(self):
         rb.DOWNLOAD_CAP = 10
         rb.OPEN = FakeOpen(FakeResponse(b"x" * 100))
@@ -932,3 +943,174 @@ class FetchPhase(TestCase):
         rb.OPEN = FakeOpen(http_error(404))
         with self.assertRaises(rb.BackupError):
             rb.fetchphase(self.conf(), "k")
+
+
+def multipart_fields(request):
+    """Parse the multipart body FakeOpen captured into {name: (filename, bytes)}."""
+    ctype = request.get_header("Content-type")
+    boundary = ctype.split("boundary=")[1].encode()
+    fields = {}
+    for part in request.data.split(b"--" + boundary)[1:-1]:
+        head, _, body = part.partition(b"\r\n\r\n")
+        name = re.search(rb'name="([^"]+)"', head).group(1).decode()
+        fname = re.search(rb'filename="([^"]+)"', head)
+        fields[name] = (fname.group(1).decode() if fname else None, body.rstrip(b"\r\n"))
+    return fields
+
+
+class ImmichClient(TestCase):
+    def client(self, *responses):
+        rb.OPEN = FakeOpen(*responses)
+        return rb.Immich("https://immich.example/api", "KEY")
+
+    def test_call_sets_headers_and_parses(self):
+        im = self.client(FakeResponse({"id": "u1"}))
+        self.assertEqual(im.me(), {"id": "u1"})
+        req = rb.OPEN.requests[0]
+        self.assertEqual(req.full_url, "https://immich.example/api/users/me")
+        self.assertEqual(req.get_header("X-api-key"), "KEY")
+
+    def test_http_error_becomes_backuperror(self):
+        im = self.client(http_error(401, b"unauthorized"))
+        with self.assertRaises(rb.BackupError) as cm:
+            im.me()
+        self.assertIn("401", str(cm.exception))
+
+    def test_a_body_that_is_not_json_becomes_backuperror(self):
+        im = self.client(FakeResponse(b"<html>proxy says no</html>"))
+        with self.assertRaises(rb.BackupError) as cm:
+            im.me()
+        self.assertIn("/users/me", str(cm.exception))
+
+    def test_me_without_an_id_raises(self):
+        im = self.client(FakeResponse({"email": "someone@example.invalid"}))
+        with self.assertRaises(rb.BackupError):
+            im.me()
+
+    def test_createalbum_without_an_id_raises(self):
+        im = self.client(FakeResponse({"albumName": "RGL"}))
+        with self.assertRaises(rb.BackupError):
+            im.createalbum("RGL")
+
+    def test_assets_pages_and_filters_owner(self):
+        page1 = {"assets": {"items": [
+            {"id": "a1", "ownerId": "u1", "checksum": "c1", "originalFileName": "1.jpg",
+             "exifInfo": {"description": "d", "latitude": 51.5, "longitude": -0.1,
+                          "dateTimeOriginal": "2009-07-22T14:03:11.000Z"}},
+            {"id": "a2", "ownerId": "other", "checksum": "c2"}], "nextPage": "2"}}
+        page2 = {"assets": {"items": [
+            {"id": "a3", "ownerId": "u1", "checksum": "c3", "exifInfo": None}], "nextPage": None}}
+        im = self.client(FakeResponse(page1), FakeResponse(page2))
+        assets = im.assets("u1")
+        self.assertEqual(sorted(assets), ["c1", "c3"])
+        self.assertEqual(assets["c1"]["latitude"], 51.5)
+        self.assertEqual(assets["c1"]["taken"], "2009-07-22T14:03:11.000Z")
+        self.assertEqual(assets["c3"]["description"], "")
+        self.assertIsNone(assets["c3"]["latitude"])
+        body = json.loads(rb.OPEN.requests[1].data)
+        self.assertEqual(body["page"], 2)
+        self.assertTrue(body["withExif"])
+        self.assertTrue(body["withArchived"])
+
+    def test_asset_reads_one_in_the_same_shape(self):
+        im = self.client(FakeResponse({
+            "id": "a1", "originalFileName": "111.jpg",
+            "exifInfo": {"description": "Mine", "latitude": 51.5, "longitude": -0.1,
+                         "dateTimeOriginal": "2009-07-22T14:03:11.000Z"}}))
+        self.assertEqual(im.asset("a1"), {
+            "id": "a1", "name": "111.jpg", "description": "Mine",
+            "latitude": 51.5, "longitude": -0.1,
+            "taken": "2009-07-22T14:03:11.000Z"})
+        req = rb.OPEN.requests[0]
+        self.assertEqual(req.get_method(), "GET")
+        self.assertTrue(req.full_url.endswith("/assets/a1"))
+
+    def test_asset_without_exif_is_blank_not_missing(self):
+        im = self.client(FakeResponse({"id": "a1", "exifInfo": None}))
+        asset = im.asset("a1")
+        self.assertEqual((asset["description"], asset["taken"]), ("", ""))
+        self.assertIsNone(asset["latitude"])
+
+    def test_upload_is_multipart_with_checksum_header(self):
+        path = self.write("111.jpg", b"JPEGDATA")
+        im = self.client(FakeResponse({"id": "a9", "status": "created"}, status=201))
+        result = im.upload(path, "Q0hFQ0s=", "111", "2009-07-22T14:03:11Z", "2026-09-09T00:00:00Z")
+        self.assertEqual(result, ("a9", "created"))
+        req = rb.OPEN.requests[0]
+        self.assertEqual(req.get_method(), "POST")
+        self.assertTrue(req.full_url.endswith("/assets"))
+        self.assertEqual(req.get_header("X-immich-checksum"), "Q0hFQ0s=")
+        fields = multipart_fields(req)
+        self.assertEqual(fields["assetData"], ("111.jpg", b"JPEGDATA"))
+        self.assertEqual(fields["deviceAssetId"][1], b"111")
+        self.assertEqual(fields["deviceId"][1], rb.DEVICE_ID.encode())
+        self.assertEqual(fields["fileCreatedAt"][1], b"2009-07-22T14:03:11Z")
+        self.assertEqual(fields["fileModifiedAt"][1], b"2026-09-09T00:00:00Z")
+
+    def test_upload_duplicate(self):
+        path = self.write("111.jpg", b"J")
+        im = self.client(FakeResponse({"id": "a1", "status": "duplicate"}))
+        self.assertEqual(im.upload(path, "x", "111", "t", "t"), ("a1", "duplicate"))
+
+    def test_upload_empty_response_raises(self):
+        path = self.write("111.jpg", b"J")
+        im = self.client(FakeResponse(b""))
+        with self.assertRaises(rb.BackupError):
+            im.upload(path, "x", "111", "t", "t")
+
+    def test_update_sends_only_given_fields(self):
+        im = self.client(FakeResponse(b""))
+        im.update("a1", description="d", latitude=51.5, longitude=-0.1)
+        req = rb.OPEN.requests[0]
+        self.assertEqual(req.get_method(), "PUT")
+        self.assertTrue(req.full_url.endswith("/assets/a1"))
+        self.assertEqual(json.loads(req.data), {"description": "d", "latitude": 51.5, "longitude": -0.1})
+
+    def test_albums(self):
+        im = self.client(FakeResponse([{"albumName": "RGL", "id": "al1", "ownerId": "u1"},
+                                       {"albumName": "RGL", "id": "shared", "ownerId": "u2"},
+                                       {"albumName": "Nameless"}]),
+                         FakeResponse({"id": "al2"}),
+                         FakeResponse({"assets": [{"id": "a1"}, {"id": "a2"}]}),
+                         FakeResponse([{"id": "a3", "success": True}, {"id": "a1", "success": False, "error": "duplicate"}]))
+        self.assertEqual(im.albums("u1"), {"RGL": "al1"})
+        self.assertEqual(im.createalbum("X"), "al2")
+        self.assertEqual(im.albumassets("al1"), {"a1", "a2"})
+        self.assertEqual(im.addtoalbum("al1", ["a3", "a1"]), 1)
+
+    def test_tags(self):
+        im = self.client(FakeResponse([{"value": "rgl", "id": "t1"}, {"value": "rgl/licence/x", "id": "t2"}]),
+                         FakeResponse([{"success": True}, {"success": False, "error": "duplicate"}]))
+        self.assertEqual(im.tagids(["rgl", "rgl/licence/x"]), {"rgl": "t1", "rgl/licence/x": "t2"})
+        self.assertEqual(im.tagassets("t1", ["a1", "a2"]), (1, 1))
+
+    def test_tagids_skips_malformed_entries(self):
+        im = self.client(FakeResponse([{"value": "rgl", "id": "t1"}, {"value": "rgl/broken"}]))
+        self.assertEqual(im.tagids(["rgl", "rgl/broken"]), {"rgl": "t1"})
+
+    def test_batches_of_500(self):
+        ids = ["a%d" % i for i in range(1200)]
+        im = self.client(FakeResponse([{"success": True}] * 500),
+                         FakeResponse([{"success": True}] * 500),
+                         FakeResponse([{"success": True}] * 200))
+        self.assertEqual(im.addtoalbum("al", ids), 1200)
+        self.assertEqual(len(rb.OPEN.requests), 3)
+
+
+class Hashing(TestCase):
+    def test_checksum_is_base64_sha1(self):
+        path = self.write("f", b"hello")
+        self.assertEqual(rb.checksum(path), "qvTGHdzF6KLavt4PO0gs2a6pQ00=")
+
+    def test_cache_reuses_until_file_changes(self):
+        path = self.write("f", b"hello")
+        cache = rb.Hashes(os.path.join(self.dir, rb.HASHES))
+        first = cache.of(path, "f", os.stat(path))
+        cache.save()
+        again = rb.Hashes(os.path.join(self.dir, rb.HASHES))
+        self.assertEqual(again.of(path, "f", os.stat(path)), first)
+        self.assertFalse(again.dirty)
+        self.write("f", b"changed")
+        os.utime(path, (1, 1))
+        self.assertNotEqual(again.of(path, "f", os.stat(path)), first)
+        self.assertTrue(again.dirty)