|
34 | 34 |
|
35 | 35 | import re |
36 | 36 | from pathlib import Path |
37 | | - |
| 37 | +from contextlib import suppress |
| 38 | +import os, sys, shutil, subprocess |
| 39 | +import stat |
| 40 | + |
| 41 | +def _is_windows_junction(p: Path) -> bool: |
| 42 | + """Return True if path is a directory junction (reparse point mount point).""" |
| 43 | + try: |
| 44 | + st = p.lstat() |
| 45 | + return getattr(st, "st_reparse_tag", 0) == stat.IO_REPARSE_TAG_MOUNT_POINT |
| 46 | + except (OSError, AttributeError): |
| 47 | + return False |
| 48 | + |
| 49 | +def _safe_remove_path(p: Path) -> None: |
| 50 | + """Remove file/dir/symlink/junction at p without following links.""" |
| 51 | + if not os.path.lexists(str(p)): |
| 52 | + return |
| 53 | + with suppress(FileNotFoundError): |
| 54 | + if p.is_symlink(): |
| 55 | + p.unlink() |
| 56 | + elif _is_windows_junction(p): |
| 57 | + os.rmdir(p) |
| 58 | + elif p.is_dir(): |
| 59 | + shutil.rmtree(p) |
| 60 | + else: |
| 61 | + p.unlink() |
| 62 | + |
| 63 | +def _make_latest_windows(latest: Path, target: Path) -> None: |
| 64 | + # Clean previous latest (symlink/junction/dir/file) |
| 65 | + _safe_remove_path(latest) |
| 66 | + |
| 67 | + tmp = latest.with_name(latest.name + "_tmp") |
| 68 | + _safe_remove_path(tmp) |
| 69 | + |
| 70 | + # Try junction first (no admin needed), fallback to copy |
| 71 | + try: |
| 72 | + subprocess.run( |
| 73 | + ["cmd", "/c", "mklink", "/J", str(tmp), str(target.resolve())], |
| 74 | + check=True, |
| 75 | + capture_output=True, |
| 76 | + ) |
| 77 | + except Exception: |
| 78 | + shutil.copytree(target, tmp) |
| 79 | + |
| 80 | + os.replace(tmp, latest) |
38 | 81 |
|
39 | 82 | def create_versioned_dir(root_dir: str | Path) -> Path: |
40 | 83 | """Create a new version directory and update the ``latest`` symbolic link. |
@@ -100,10 +143,12 @@ def create_versioned_dir(root_dir: str | Path) -> Path: |
100 | 143 |
|
101 | 144 | # Update the 'latest' symbolic link to point to the new version directory |
102 | 145 | latest_link_path = root_dir / "latest" |
103 | | - if latest_link_path.is_symlink() or latest_link_path.exists(): |
104 | | - latest_link_path.unlink() |
105 | | - latest_link_path.symlink_to(new_version_dir, target_is_directory=True) |
106 | | - |
| 146 | + if sys.platform.startswith("win"): |
| 147 | + _make_latest_windows(latest_link_path, new_version_dir) |
| 148 | + else: |
| 149 | + if latest_link_path.is_symlink() or latest_link_path.exists(): |
| 150 | + latest_link_path.unlink() |
| 151 | + latest_link_path.symlink_to(new_version_dir, target_is_directory=True) |
107 | 152 | return latest_link_path |
108 | 153 |
|
109 | 154 |
|
|
0 commit comments