Store each photo's metadata in a json sidecar beside it

The sidecar is what makes the directory a backup on its own and what the
immich phase reads, so it is written atomically and a photo only counts as
done when the sidecar, its file and no .part remain.

Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>
This commit is contained in:
bobandClaude Fable 5.1 committed 2026-09-10 12:07:44 +01:00
1 parent fc434a3d9b
commit f7884b943e
2 files changed
+115

No files matched your search

+55
View File
@@ -600,5 +600,60 @@ def extensionfor(source, contenttype=""):
return CONTENT_EXTENSIONS.get(kind, ".jpg")
def utcnow():
return datetime.datetime.now(datetime.timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ")
def sidecarpath(directory, key):
return os.path.join(directory, key + ".json")
def writejson(path, data):
partial = path + ".part"
with open(partial, "w", encoding="utf-8") as fh:
json.dump(data, fh, indent=2, ensure_ascii=False, sort_keys=True)
fh.write("\n")
os.replace(partial, path)
def readsidecar(path):
try:
with open(path, encoding="utf-8") as fh:
data = json.load(fh)
except (OSError, ValueError):
return None
return data if isinstance(data, dict) and data.get("key") else None
def mediapath(directory, key):
try:
names = os.listdir(directory)
except OSError:
return None
for name in sorted(names):
stem, ext = os.path.splitext(name)
if stem == key and ext not in (".json", ".part"):
return os.path.join(directory, name)
return None
def iscomplete(directory, key):
if os.path.exists(os.path.join(directory, key + ".part")):
return False
side = readsidecar(sidecarpath(directory, key))
if not side or not side.get("file"):
return False
return os.path.isfile(os.path.join(directory, side["file"]))
def listsidecars(directory):
found = []
try:
names = sorted(os.listdir(directory))
except OSError as e:
raise BackupError("%s: cannot be listed: %s" % (directory, e.strerror or e))
for name in names:
if not name.endswith(".json") or name == HASHES:
continue
side = readsidecar(os.path.join(directory, name))
if side:
found.append((side["key"], side))
return found
if __name__ == "__main__":
sys.exit(0)
+60
View File
@@ -541,3 +541,63 @@ class Flickr(TestCase):
self.assertEqual(rb.extensionfor("https://x/1_o.jpeg", ""), ".jpg")
self.assertEqual(rb.extensionfor("https://x/1_o", "image/gif"), ".gif")
self.assertEqual(rb.extensionfor("https://x/1_o.weird", "text/html"), ".jpg")
def sidecar(key="111", file="111.jpg", nodes=None, copyright="P", licence="https://creativecommons.org/licenses/by-sa/2.0/", flickr=True):
data = {"key": key, "file": file, "source": "flickr" if flickr else "url",
"rgl": {"copyright": copyright, "licence": licence,
"nodes": nodes or [{"param": "A,_E1", "name": "A, E1",
"url": "https://london.randomness.org.uk/wiki.cgi?A%2C_E1",
"lat": "51.5", "long": "-0.1"}]},
"fetched": "2026-09-09T21:14:00Z"}
if flickr:
data["flickr"] = {"id": key, "owner": "12345678@N00", "owner_name": "handle",
"owner_realname": "Real Name", "title": "A title",
"taken": "2009-07-22 14:03:11", "posted": "2009-07-23T10:40:00Z",
"licence_id": 5, "page": "https://www.flickr.com/photos/12345678@N00/%s/" % key,
"size_label": "Original", "width": 800, "height": 600}
else:
data["page"] = "https://www.geograph.org.uk/photos/1.jpg"
return data
class Sidecars(TestCase):
def test_round_trip_is_atomic(self):
path = rb.sidecarpath(self.dir, "111")
self.assertEqual(path, os.path.join(self.dir, "111.json"))
rb.writejson(path, sidecar())
self.assertEqual(rb.readsidecar(path)["key"], "111")
self.assertFalse(os.path.exists(path + ".part"))
def test_unreadable_sidecar_is_none(self):
self.assertIsNone(rb.readsidecar(os.path.join(self.dir, "none.json")))
self.write("bad.json", "{not json")
self.assertIsNone(rb.readsidecar(os.path.join(self.dir, "bad.json")))
def test_mediapath_finds_the_photo_not_the_sidecar(self):
self.write("111.json", "{}")
self.write("111.part", b"x")
self.assertIsNone(rb.mediapath(self.dir, "111"))
self.write("111.png", b"x")
self.assertEqual(rb.mediapath(self.dir, "111"), os.path.join(self.dir, "111.png"))
def test_iscomplete(self):
self.assertFalse(rb.iscomplete(self.dir, "111"))
rb.writejson(rb.sidecarpath(self.dir, "111"), sidecar())
self.assertFalse(rb.iscomplete(self.dir, "111")) # no file yet
self.write("111.jpg", b"x")
self.assertTrue(rb.iscomplete(self.dir, "111"))
self.write("111.part", b"x")
self.assertFalse(rb.iscomplete(self.dir, "111")) # interrupted refetch
def test_listsidecars_unreadable_directory_is_an_error(self):
with self.assertRaises(rb.BackupError):
rb.listsidecars(os.path.join(self.dir, "nowhere"))
def test_listsidecars_sorted_and_skips_hash_cache(self):
rb.writejson(rb.sidecarpath(self.dir, "222"), sidecar(key="222", file="222.jpg"))
rb.writejson(rb.sidecarpath(self.dir, "111"), sidecar())
rb.writejson(os.path.join(self.dir, rb.HASHES), {})
self.assertEqual([k for k, _ in rb.listsidecars(self.dir)], ["111", "222"])
def test_utcnow_shape(self):
self.assertRegex(rb.utcnow(), r"^\d{4}-\d\d-\d\dT\d\d:\d\d:\d\dZ$")