Thomas/Piwigo_Bratonien_Tools
1
0
Fork 0
mirror of https://github.com/Terranom674/Piwigo_Bratonien_Tools.git synced 2026-09-19 15:14:34 +00:00
Code Issues Packages Projects Releases Wiki Activity

Make NC shadow tree replacement rollback-safe

Browse Source
This commit is contained in:
Terranom674
2026-08-18 08:47:37 +02:00
parent 73a13a87d3
commit 6d278c2d0d
1 changed files with 53 additions and 42 deletions
Download Patch File Download Diff File Expand all files Collapse all files

95
runtime/lib/shadow_tree.py
View File

@@ -56,12 +56,7 @@ def load_manifest(path: Path) -> list[dict[str, str]]:
share_id, item_type, display_name, source_path = fields share_id, item_type, display_name, source_path = fields
if item_type not in {"folder", "file"}: if item_type not in {"folder", "file"}:
raise ValueError(f"{path}:{line_number}: unsupported item_type {item_type!r}") raise ValueError(f"{path}:{line_number}: unsupported item_type {item_type!r}")
entries.append({ entries.append({"share_id": share_id, "item_type": item_type, "display_name": display_name, "source_path": source_path})
"share_id": share_id,
"item_type": item_type,
"display_name": display_name,
"source_path": source_path,
})
return entries return entries
@@ -87,8 +82,7 @@ def preferred_target(source_key: str, raw_name: str, parent_target: Path, old_ma
def mirror_directory(source: Path, target: Path, source_key: str, target_key: Path, old_map: dict[str, str], new_map: dict[str, str]) -> None: def mirror_directory(source: Path, target: Path, source_key: str, target_key: Path, old_map: dict[str, str], new_map: dict[str, str]) -> None:
target.mkdir(parents=True, exist_ok=True) target.mkdir(parents=True, exist_ok=True)
used: set[str] = set() used: set[str] = set()
children = sorted(source.iterdir(), key=lambda item: (item.name.casefold(), item.name)) for child in sorted(source.iterdir(), key=lambda item: (item.name.casefold(), item.name)):
for child in children:
if child.is_symlink(): if child.is_symlink():
continue continue
child_source_key = f"{source_key}/{child.name}" child_source_key = f"{source_key}/{child.name}"
@@ -103,51 +97,68 @@ def mirror_directory(source: Path, target: Path, source_key: str, target_key: Pa
child_target.symlink_to(child.resolve()) child_target.symlink_to(child.resolve())
def remove_tree(path: Path) -> None:
if path.is_symlink() or path.is_file():
path.unlink(missing_ok=True)
elif path.exists():
shutil.rmtree(path)
def build(manifest: Path, destination: Path, state_file: Path) -> None: def build(manifest: Path, destination: Path, state_file: Path) -> None:
entries = load_manifest(manifest) entries = load_manifest(manifest)
old_map = load_map(state_file) old_map = load_map(state_file)
new_map: dict[str, str] = {} new_map: dict[str, str] = {}
staging = destination.with_name(f".{destination.name}.next") staging = destination.with_name(f".{destination.name}.next")
previous = destination.with_name(f".{destination.name}.previous") previous = destination.with_name(f".{destination.name}.previous")
state_staging = state_file.with_suffix(state_file.suffix + ".next")
if staging.exists(): remove_tree(staging)
shutil.rmtree(staging) remove_tree(previous)
if state_staging.exists():
state_staging.unlink()
staging.mkdir(parents=True) staging.mkdir(parents=True)
used_roots: set[str] = set() try:
for entry in sorted(entries, key=lambda item: (item["display_name"].casefold(), item["share_id"])): used_roots: set[str] = set()
source = Path(entry["source_path"]) for entry in sorted(entries, key=lambda item: (item["display_name"].casefold(), item["share_id"])):
source_key = f"share:{entry['share_id']}" source = Path(entry["source_path"])
preferred = preferred_target(source_key, entry["display_name"], Path("."), old_map) source_key = f"share:{entry['share_id']}"
is_file_share = entry["item_type"] == "file" preferred = preferred_target(source_key, entry["display_name"], Path("."), old_map)
root_name = unique_name(preferred, used_roots, is_file_share) is_file_share = entry["item_type"] == "file"
root_key = Path(root_name) root_name = unique_name(preferred, used_roots, is_file_share)
new_map[source_key] = root_key.as_posix() root_key = Path(root_name)
new_map[source_key] = root_key.as_posix()
if is_file_share:
if not source.is_file():
raise FileNotFoundError(f"source is not a readable file: {source}")
(staging / root_name).symlink_to(source.resolve())
else:
if not source.is_dir():
raise FileNotFoundError(f"source is not a readable directory: {source}")
mirror_directory(source, staging / root_name, source_key, root_key, old_map, new_map)
if is_file_share: state_staging.parent.mkdir(parents=True, exist_ok=True)
if not source.is_file(): with state_staging.open("w", encoding="utf-8") as handle:
raise FileNotFoundError(f"source is not a readable file: {source}") json.dump(new_map, handle, ensure_ascii=False, indent=2, sort_keys=True)
(staging / root_name).symlink_to(source.resolve()) handle.write("\n")
continue
if not source.is_dir(): had_destination = destination.exists()
raise FileNotFoundError(f"source is not a readable directory: {source}") if had_destination:
mirror_directory(source, staging / root_name, source_key, root_key, old_map, new_map) destination.rename(previous)
try:
state_staging = state_file.with_suffix(state_file.suffix + ".next") staging.rename(destination)
state_staging.parent.mkdir(parents=True, exist_ok=True) state_staging.replace(state_file)
with state_staging.open("w", encoding="utf-8") as handle: except Exception:
json.dump(new_map, handle, ensure_ascii=False, indent=2, sort_keys=True) remove_tree(destination)
handle.write("\n") if had_destination and previous.exists():
previous.rename(destination)
if previous.exists(): raise
shutil.rmtree(previous) remove_tree(previous)
if destination.exists(): except Exception:
destination.rename(previous) remove_tree(staging)
staging.rename(destination) if state_staging.exists():
state_staging.replace(state_file) state_staging.unlink()
if previous.exists(): raise
shutil.rmtree(previous)
def main() -> int: def main() -> int: