|
| 1 | +#!/usr/bin/env -S uv run --no-config --script |
| 2 | +# /// script |
| 3 | +# dependencies = ["gitignore-parser"] |
| 4 | +# /// |
| 5 | +"""Set up a worktree by copying or symlinking files from a root workspace.""" |
| 6 | + |
| 7 | +from __future__ import annotations |
| 8 | + |
| 9 | +import argparse |
| 10 | +import os |
| 11 | +import shutil |
| 12 | +import sys |
| 13 | +import tempfile |
| 14 | +from concurrent.futures import ThreadPoolExecutor, as_completed |
| 15 | +from pathlib import Path |
| 16 | + |
| 17 | +from gitignore_parser import parse_gitignore |
| 18 | + |
| 19 | + |
| 20 | +def parse_args() -> argparse.Namespace: |
| 21 | + parser = argparse.ArgumentParser( |
| 22 | + description="Copy or symlink files from a root workspace into the current directory." |
| 23 | + ) |
| 24 | + parser.add_argument("root_path", type=Path, help="Root workspace path") |
| 25 | + parser.add_argument( |
| 26 | + "-s", |
| 27 | + "--symlink", |
| 28 | + action="append", |
| 29 | + default=[], |
| 30 | + metavar="PATTERN", |
| 31 | + help="Glob pattern for files to symlink (repeatable)", |
| 32 | + ) |
| 33 | + parser.add_argument( |
| 34 | + "-c", |
| 35 | + "--copy", |
| 36 | + action="append", |
| 37 | + default=[], |
| 38 | + metavar="PATTERN", |
| 39 | + help="Glob pattern for files to copy (repeatable)", |
| 40 | + ) |
| 41 | + return parser.parse_args() |
| 42 | + |
| 43 | + |
| 44 | +def _canonicalize(p: Path) -> Path: |
| 45 | + """Canonicalize a path by resolving each symlink component via readlink.""" |
| 46 | + result = Path(p.anchor) |
| 47 | + for part in p.relative_to(p.anchor).parts: |
| 48 | + if part == "..": |
| 49 | + result = result.parent |
| 50 | + continue |
| 51 | + if part == ".": |
| 52 | + continue |
| 53 | + candidate = result / part |
| 54 | + while candidate.is_symlink(): |
| 55 | + link = Path(os.readlink(candidate)) |
| 56 | + candidate = link if link.is_absolute() else candidate.parent / link |
| 57 | + result = candidate |
| 58 | + return result |
| 59 | + |
| 60 | + |
| 61 | +def _pattern_targets_dotgit(pattern: str) -> bool: |
| 62 | + stripped = pattern.lstrip("!").lstrip("/") |
| 63 | + return stripped == ".git" or stripped.startswith(".git/") |
| 64 | + |
| 65 | + |
| 66 | +def _is_relative_to(path: Path, parent: Path) -> bool: |
| 67 | + try: |
| 68 | + path.relative_to(parent) |
| 69 | + return True |
| 70 | + except ValueError: |
| 71 | + return False |
| 72 | + |
| 73 | + |
| 74 | +def _read_patterns(path: Path) -> list[str]: |
| 75 | + """Read gitignore-format patterns from a file, returning raw pattern lines.""" |
| 76 | + patterns = [] |
| 77 | + for line in path.read_text().splitlines(): |
| 78 | + stripped = line.strip() |
| 79 | + if stripped and not stripped.startswith("#"): |
| 80 | + patterns.append(stripped) |
| 81 | + return patterns |
| 82 | + |
| 83 | + |
| 84 | +def _build_matcher(patterns: list[str], root: Path): |
| 85 | + """Build a gitignore-style matcher from a list of patterns.""" |
| 86 | + if not patterns: |
| 87 | + return None |
| 88 | + with tempfile.NamedTemporaryFile(mode="w", suffix=".gitignore", delete=False) as f: |
| 89 | + for p in patterns: |
| 90 | + f.write(p + "\n") |
| 91 | + tmp_path = f.name |
| 92 | + try: |
| 93 | + return parse_gitignore(tmp_path, base_dir=str(root)) |
| 94 | + finally: |
| 95 | + os.unlink(tmp_path) |
| 96 | + |
| 97 | + |
| 98 | +def _lowest_common_ancestor(a: Path, b: Path) -> Path: |
| 99 | + """Return the deepest common ancestor directory of two absolute paths.""" |
| 100 | + common: list[str] = [] |
| 101 | + for pa, pb in zip(a.parts, b.parts): |
| 102 | + if pa == pb: |
| 103 | + common.append(pa) |
| 104 | + else: |
| 105 | + break |
| 106 | + return Path(*common) if common else Path("/") |
| 107 | + |
| 108 | + |
| 109 | +def _readjust_symlink_target(link_target: Path, root: Path, cwd: Path) -> Path: |
| 110 | + """Decide whether a symlink target inside root should be readjusted for cwd. |
| 111 | +
|
| 112 | + Returns the (possibly adjusted) target path. |
| 113 | + """ |
| 114 | + if not _is_relative_to(cwd, root): |
| 115 | + return cwd / link_target.relative_to(root) |
| 116 | + lca = _lowest_common_ancestor(cwd, link_target) |
| 117 | + if lca == root: |
| 118 | + return cwd / link_target.relative_to(root) |
| 119 | + return link_target |
| 120 | + |
| 121 | + |
| 122 | +def _dst_exists(dst: Path) -> bool: |
| 123 | + """Check if dst exists (includes broken symlinks).""" |
| 124 | + return dst.exists() or dst.is_symlink() |
| 125 | + |
| 126 | + |
| 127 | +def copy_entry(src: Path, dst: Path, root: Path, cwd: Path) -> None: |
| 128 | + """Copy a file or directory from src to dst, handling symlink readjustment.""" |
| 129 | + if src.is_symlink(): |
| 130 | + if _dst_exists(dst): |
| 131 | + return |
| 132 | + |
| 133 | + raw_target = Path(os.readlink(src)) |
| 134 | + |
| 135 | + if raw_target.is_absolute(): |
| 136 | + link_target = _canonicalize(raw_target) |
| 137 | + if _is_relative_to(link_target, root): |
| 138 | + new_target = _readjust_symlink_target(link_target, root, cwd) |
| 139 | + else: |
| 140 | + new_target = link_target |
| 141 | + else: |
| 142 | + resolved = _canonicalize(src.parent / raw_target) |
| 143 | + if _is_relative_to(resolved, root): |
| 144 | + # Inside root: preserve the relative target unchanged |
| 145 | + new_target = raw_target |
| 146 | + else: |
| 147 | + # Outside root: convert to absolute |
| 148 | + new_target = resolved |
| 149 | + |
| 150 | + dst.parent.mkdir(parents=True, exist_ok=True) |
| 151 | + os.symlink(new_target, dst) |
| 152 | + elif src.is_dir(): |
| 153 | + if dst.exists() and (not dst.is_dir() or any(dst.iterdir())): |
| 154 | + return |
| 155 | + if dst.exists(): |
| 156 | + shutil.rmtree(dst) |
| 157 | + shutil.copytree(src, dst, symlinks=True, copy_function=shutil.copy2) |
| 158 | + else: |
| 159 | + if dst.exists() and (not dst.is_file() or dst.stat().st_size > 0): |
| 160 | + return |
| 161 | + dst.parent.mkdir(parents=True, exist_ok=True) |
| 162 | + shutil.copy2(src, dst) |
| 163 | + |
| 164 | + |
| 165 | +def symlink_entry(src: Path, dst: Path) -> None: |
| 166 | + """Create an absolute symlink at dst pointing to src.""" |
| 167 | + if _dst_exists(dst): |
| 168 | + return |
| 169 | + dst.parent.mkdir(parents=True, exist_ok=True) |
| 170 | + os.symlink(_canonicalize(src), dst) |
| 171 | + |
| 172 | + |
| 173 | +def main() -> None: |
| 174 | + args = parse_args() |
| 175 | + root = _canonicalize(args.root_path.absolute()) |
| 176 | + cwd = _canonicalize(Path.cwd()) |
| 177 | + |
| 178 | + if not root.is_dir(): |
| 179 | + print(f"error: root path is not a directory: {root}", file=sys.stderr) |
| 180 | + sys.exit(1) |
| 181 | + |
| 182 | + # Collect patterns: .worktreeinclude first, then CLI (CLI overrides) |
| 183 | + copy_patterns = [] |
| 184 | + worktreeinclude = root / ".worktreeinclude" |
| 185 | + if worktreeinclude.is_file(): |
| 186 | + copy_patterns.extend(_read_patterns(worktreeinclude)) |
| 187 | + copy_patterns.extend(args.copy) |
| 188 | + symlink_patterns = list(args.symlink) |
| 189 | + |
| 190 | + if not copy_patterns and not symlink_patterns: |
| 191 | + print( |
| 192 | + "error: no patterns specified (use -c, -s, or .worktreeinclude)", |
| 193 | + file=sys.stderr, |
| 194 | + ) |
| 195 | + sys.exit(1) |
| 196 | + |
| 197 | + # Build matchers |
| 198 | + copy_matcher = _build_matcher(copy_patterns, root) |
| 199 | + symlink_matcher = _build_matcher(symlink_patterns, root) |
| 200 | + copy_allows_dotgit = any(_pattern_targets_dotgit(p) for p in copy_patterns) |
| 201 | + symlink_allows_dotgit = any(_pattern_targets_dotgit(p) for p in symlink_patterns) |
| 202 | + any_allows_dotgit = copy_allows_dotgit or symlink_allows_dotgit |
| 203 | + |
| 204 | + # Determine cwd exclusion: only if cwd is inside root |
| 205 | + cwd_exclude = cwd if _is_relative_to(cwd, root) else None |
| 206 | + |
| 207 | + # Walk root and collect work items |
| 208 | + work: list[tuple[str, Path, Path]] = [] |
| 209 | + seen_dsts: set[Path] = set() |
| 210 | + |
| 211 | + for dirpath, dirnames, filenames in os.walk(root): |
| 212 | + dp = Path(dirpath) |
| 213 | + |
| 214 | + # Skip root's .git unless any pattern allows it |
| 215 | + if dp == root and not any_allows_dotgit: |
| 216 | + try: |
| 217 | + dirnames.remove(".git") |
| 218 | + except ValueError: |
| 219 | + pass |
| 220 | + |
| 221 | + # cwd exclusion |
| 222 | + if cwd_exclude is not None: |
| 223 | + canon = _canonicalize(dp) |
| 224 | + if canon == cwd_exclude or _is_relative_to(canon, cwd_exclude): |
| 225 | + dirnames.clear() |
| 226 | + continue |
| 227 | + |
| 228 | + # Check subdirectories for whole-directory matches |
| 229 | + skip_dirs: set[str] = set() |
| 230 | + for dname in list(dirnames): |
| 231 | + src = dp / dname |
| 232 | + rel = src.relative_to(root) |
| 233 | + in_dotgit = bool(rel.parts) and rel.parts[0] == ".git" |
| 234 | + dst = cwd / rel |
| 235 | + |
| 236 | + if dst in seen_dsts: |
| 237 | + skip_dirs.add(dname) |
| 238 | + continue |
| 239 | + |
| 240 | + if copy_matcher and copy_matcher(str(src)): |
| 241 | + if not (in_dotgit and not copy_allows_dotgit): |
| 242 | + seen_dsts.add(dst) |
| 243 | + work.append(("copy", src, dst)) |
| 244 | + skip_dirs.add(dname) |
| 245 | + continue |
| 246 | + |
| 247 | + if symlink_matcher and symlink_matcher(str(src)): |
| 248 | + if not (in_dotgit and not symlink_allows_dotgit): |
| 249 | + seen_dsts.add(dst) |
| 250 | + work.append(("symlink", src, dst)) |
| 251 | + skip_dirs.add(dname) |
| 252 | + |
| 253 | + dirnames[:] = [d for d in dirnames if d not in skip_dirs] |
| 254 | + |
| 255 | + # Check files (includes symlinks to files) |
| 256 | + for fname in filenames: |
| 257 | + src = dp / fname |
| 258 | + rel = src.relative_to(root) |
| 259 | + in_dotgit = bool(rel.parts) and rel.parts[0] == ".git" |
| 260 | + dst = cwd / rel |
| 261 | + |
| 262 | + if dst in seen_dsts: |
| 263 | + continue |
| 264 | + |
| 265 | + if copy_matcher and copy_matcher(str(src)): |
| 266 | + if not (in_dotgit and not copy_allows_dotgit): |
| 267 | + seen_dsts.add(dst) |
| 268 | + work.append(("copy", src, dst)) |
| 269 | + continue |
| 270 | + |
| 271 | + if symlink_matcher and symlink_matcher(str(src)): |
| 272 | + if not (in_dotgit and not symlink_allows_dotgit): |
| 273 | + seen_dsts.add(dst) |
| 274 | + work.append(("symlink", src, dst)) |
| 275 | + |
| 276 | + if not work: |
| 277 | + print("warning: no files matched any pattern", file=sys.stderr) |
| 278 | + return |
| 279 | + |
| 280 | + errors: list[tuple[str, Path, Exception]] = [] |
| 281 | + |
| 282 | + with ThreadPoolExecutor() as executor: |
| 283 | + futures = {} |
| 284 | + for action, src, dst in work: |
| 285 | + if action == "copy": |
| 286 | + fut = executor.submit(copy_entry, src, dst, root, cwd) |
| 287 | + else: |
| 288 | + fut = executor.submit(symlink_entry, src, dst) |
| 289 | + futures[fut] = (action, src) |
| 290 | + |
| 291 | + for fut in as_completed(futures): |
| 292 | + action, src = futures[fut] |
| 293 | + try: |
| 294 | + fut.result() |
| 295 | + except Exception as exc: |
| 296 | + errors.append((action, src, exc)) |
| 297 | + |
| 298 | + if errors: |
| 299 | + for action, src, exc in errors: |
| 300 | + print(f"error: {action} {src}: {exc}", file=sys.stderr) |
| 301 | + sys.exit(1) |
| 302 | + |
| 303 | + |
| 304 | +if __name__ == "__main__": |
| 305 | + main() |
0 commit comments