feat: profiles export/import (portable zip)
Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
This commit is contained in:
@@ -98,3 +98,43 @@ def duplicate_profile(base, src_id, name, new_id, ts=0):
|
|||||||
reg["profiles"].append(entry)
|
reg["profiles"].append(entry)
|
||||||
write_registry(base, reg)
|
write_registry(base, reg)
|
||||||
return entry
|
return entry
|
||||||
|
|
||||||
|
|
||||||
|
def export_profile(base, pid, dest_zip):
|
||||||
|
src = Path(base) / pid
|
||||||
|
if not src.exists():
|
||||||
|
raise KeyError(pid)
|
||||||
|
entry = find_by_id(read_registry(base), pid)
|
||||||
|
name = entry["name"] if entry else pid
|
||||||
|
with zipfile.ZipFile(dest_zip, "w", zipfile.ZIP_DEFLATED) as z:
|
||||||
|
z.writestr("profile_meta.json", json.dumps({"name": name}))
|
||||||
|
for f in src.rglob("*"):
|
||||||
|
if f.is_file():
|
||||||
|
z.write(f, arcname=str(Path("pool") / f.relative_to(src)))
|
||||||
|
return dest_zip
|
||||||
|
|
||||||
|
|
||||||
|
def import_profile(base, src_zip, new_id, name=None, ts=0):
|
||||||
|
reg = read_registry(base)
|
||||||
|
meta_name = None
|
||||||
|
dst = Path(base) / new_id
|
||||||
|
dst.mkdir(parents=True, exist_ok=True)
|
||||||
|
with zipfile.ZipFile(src_zip) as z:
|
||||||
|
names = z.namelist()
|
||||||
|
if "profile_meta.json" in names:
|
||||||
|
meta_name = json.loads(z.read("profile_meta.json")).get("name")
|
||||||
|
for n in names:
|
||||||
|
if n.startswith("pool/") and not n.endswith("/"):
|
||||||
|
target = dst / n[len("pool/"):]
|
||||||
|
target.parent.mkdir(parents=True, exist_ok=True)
|
||||||
|
with z.open(n) as srcf, open(target, "wb") as out:
|
||||||
|
shutil.copyfileobj(srcf, out)
|
||||||
|
final = name or meta_name or new_id
|
||||||
|
candidate, i = final, 2
|
||||||
|
while find_by_name(reg, candidate):
|
||||||
|
candidate = f"{final} ({i})"
|
||||||
|
i += 1
|
||||||
|
entry = {"id": new_id, "name": candidate, "created": ts}
|
||||||
|
reg["profiles"].append(entry)
|
||||||
|
write_registry(base, reg)
|
||||||
|
return entry
|
||||||
|
|||||||
@@ -67,3 +67,26 @@ def test_duplicate_duplicate_name_raises(tmp_path):
|
|||||||
pr.create_profile(str(tmp_path), "src", "id1")
|
pr.create_profile(str(tmp_path), "src", "id1")
|
||||||
with pytest.raises(ValueError):
|
with pytest.raises(ValueError):
|
||||||
pr.duplicate_profile(str(tmp_path), "id1", "src", "id2")
|
pr.duplicate_profile(str(tmp_path), "id1", "src", "id2")
|
||||||
|
|
||||||
|
def test_export_import_roundtrip(tmp_path):
|
||||||
|
src_base = str(tmp_path / "a"); dst_base = str(tmp_path / "b")
|
||||||
|
pr.create_profile(src_base, "setA", "id1", ts=1)
|
||||||
|
from pathlib import Path
|
||||||
|
(Path(src_base) / "id1" / "img_0001.png").write_bytes(b"hello")
|
||||||
|
zpath = str(tmp_path / "setA.zip")
|
||||||
|
pr.export_profile(src_base, "id1", zpath)
|
||||||
|
assert (tmp_path / "setA.zip").exists()
|
||||||
|
# import into a different base, fresh id
|
||||||
|
e = pr.import_profile(dst_base, zpath, "id99", ts=2)
|
||||||
|
assert e["id"] == "id99"
|
||||||
|
assert e["name"] == "setA" # name carried in zip meta
|
||||||
|
assert (Path(dst_base) / "id99" / "img_0001.png").read_bytes() == b"hello"
|
||||||
|
|
||||||
|
def test_import_name_collision_suffixes(tmp_path):
|
||||||
|
base = str(tmp_path)
|
||||||
|
pr.create_profile(base, "setA", "id1")
|
||||||
|
from pathlib import Path
|
||||||
|
(Path(base) / "id1" / "f.png").write_bytes(b"x")
|
||||||
|
z = str(tmp_path / "e.zip"); pr.export_profile(base, "id1", z)
|
||||||
|
e = pr.import_profile(base, z, "id2")
|
||||||
|
assert e["name"] == "setA (2)"
|
||||||
|
|||||||
Reference in New Issue
Block a user