Spaces:
Sleeping
Sleeping
| """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 | |
| 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) | |