Spaces:
Sleeping
Sleeping
File size: 3,002 Bytes
526950e | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 | """Staged sync: source dir -> diff vs target -> apply with deletion gate."""
from __future__ import annotations
import shutil
from dataclasses import dataclass, field
from pathlib import Path
_DIFF_PREVIEW_CAP = 5
class UnconfirmedDeletionsError(RuntimeError):
pass
@dataclass
class Diff:
additions: list[str] = field(default_factory=list)
modifications: list[str] = field(default_factory=list)
deletions: list[str] = field(default_factory=list)
def _walk_relative(d: Path) -> set[str]:
return {p.relative_to(d).as_posix() for p in d.rglob("*") if p.is_file()}
def stage_and_diff(src: Path, tgt: Path) -> Diff:
"""Compare src and tgt trees and return the diff."""
src_files = _walk_relative(src)
tgt_files = _walk_relative(tgt)
additions = sorted(src_files - tgt_files)
deletions = sorted(tgt_files - src_files)
modifications = sorted(
rel for rel in (src_files & tgt_files)
if (src / rel).read_bytes() != (tgt / rel).read_bytes()
)
return Diff(additions=additions, modifications=modifications, deletions=deletions)
def apply(diff: Diff, src: Path, tgt: Path, *, allow_deletes: bool) -> None:
"""Apply additions/modifications. Apply deletions only if allow_deletes."""
if diff.deletions and not allow_deletes:
raise UnconfirmedDeletionsError(
f"refusing to apply {len(diff.deletions)} deletions without allow_deletes=True; "
f"first few: {diff.deletions[:_DIFF_PREVIEW_CAP]}"
)
for rel in diff.additions + diff.modifications:
src_path = src / rel
tgt_path = tgt / rel
tgt_path.parent.mkdir(parents=True, exist_ok=True)
shutil.copy2(src_path, tgt_path)
if allow_deletes:
for rel in diff.deletions:
(tgt / rel).unlink()
parent = (tgt / rel).parent
while parent != tgt and parent.exists():
try:
if any(parent.iterdir()):
break
parent.rmdir()
except OSError:
break
parent = parent.parent
def _format_section(label_count: str, sigil: str, items: list[str]) -> list[str]:
if not items:
return []
lines = [f" {label_count}"]
for r in items[:_DIFF_PREVIEW_CAP]:
lines.append(f" {sigil} {r}")
if len(items) > _DIFF_PREVIEW_CAP:
lines.append(f" {sigil} ... and {len(items) - _DIFF_PREVIEW_CAP} more")
return lines
def format_diff_summary(diff: Diff) -> str:
"""Human-readable summary for confirmation prompts."""
lines: list[str] = []
lines.extend(_format_section(f"+{len(diff.additions)} additions", "+", diff.additions))
lines.extend(_format_section(f"~{len(diff.modifications)} modifications", "~", diff.modifications))
lines.extend(_format_section(f"-{len(diff.deletions)} deletions", "-", diff.deletions))
if not lines:
return " (no changes)"
return "\n".join(lines)
|