| 1 | from __future__ import annotations |
| 2 | |
| 3 | import shutil |
| 4 | from pathlib import Path |
| 5 | |
| 6 | |
| 7 | def migrate_retired_state_tree( |
| 8 | *, |
| 9 | source: Path, |
| 10 | destination: Path, |
| 11 | owner: str, |
| 12 | migrated: list[str], |
| 13 | warnings: list[str], |
| 14 | errors: list[str], |
| 15 | ) -> None: |
| 16 | """Move retired plugin state into its plugin-owned state directory. |
| 17 | |
| 18 | Existing destination data wins. Colliding source entries are preserved under |
| 19 | a suffixed name in the destination instead of overwriting live data. |
| 20 | """ |
| 21 | |
| 22 | if not source.exists() and not source.is_symlink(): |
| 23 | return |
| 24 | if _same_path(source, destination): |
| 25 | return |
| 26 | |
| 27 | try: |
| 28 | if source.is_dir() and not source.is_symlink(): |
| 29 | destination.mkdir(parents=True, exist_ok=True) |
| 30 | for child in list(source.iterdir()): |
| 31 | try: |
| 32 | _move_path(child, destination / child.name, migrated) |
| 33 | except Exception as exc: |
| 34 | errors.append(f"{owner} state migration failed for {child}: {exc}") |
| 35 | _remove_empty_dir(source, owner=owner, warnings=warnings) |
| 36 | return |
| 37 | |
| 38 | _move_path(source, destination, migrated) |
| 39 | except Exception as exc: |
| 40 | errors.append(f"{owner} state migration failed from {source} to {destination}: {exc}") |
| 41 | |
| 42 | |
| 43 | def _move_path(source: Path, target: Path, migrated: list[str]) -> None: |
| 44 | if source.is_dir() and not source.is_symlink() and target.is_dir() and not target.is_symlink(): |
| 45 | for child in list(source.iterdir()): |
| 46 | _move_path(child, target / child.name, migrated) |
| 47 | source.rmdir() |
| 48 | return |
| 49 | |
| 50 | final_target = target |
| 51 | if target.exists() or target.is_symlink(): |
| 52 | final_target = _next_conflict_path(target) |
| 53 | final_target.parent.mkdir(parents=True, exist_ok=True) |
| 54 | shutil.move(str(source), str(final_target)) |
| 55 | migrated.append(f"{source} -> {final_target}") |
| 56 | |
| 57 | |
| 58 | def _next_conflict_path(path: Path) -> Path: |
| 59 | candidate = path.with_name(f"{path.name}.retired") |
| 60 | counter = 2 |
| 61 | while candidate.exists() or candidate.is_symlink(): |
| 62 | candidate = path.with_name(f"{path.name}.retired-{counter}") |
| 63 | counter += 1 |
| 64 | return candidate |
| 65 | |
| 66 | |
| 67 | def _remove_empty_dir(path: Path, *, owner: str, warnings: list[str]) -> None: |
| 68 | try: |
| 69 | path.rmdir() |
| 70 | except OSError: |
| 71 | warnings.append(f"Retired {owner} state directory was not empty after migration: {path}") |
| 72 | |
| 73 | |
| 74 | def _same_path(left: Path, right: Path) -> bool: |
| 75 | try: |
| 76 | return left.resolve(strict=False) == right.resolve(strict=False) |
| 77 | except OSError: |
| 78 | return False |