110 lines
4.5 KiB
Python
Executable File
110 lines
4.5 KiB
Python
Executable File
#!/usr/bin/env python3
|
|
"""Synchronize skill symlinks without taking ownership of unrelated entries."""
|
|
import argparse
|
|
from pathlib import Path
|
|
import os
|
|
import re
|
|
import sys
|
|
|
|
|
|
def lexical(path):
|
|
s = str(os.path.abspath(path))
|
|
if s.startswith('\\\\?\\'):
|
|
s = s[4:]
|
|
return Path(s)
|
|
|
|
|
|
def link_target(path):
|
|
target = os.readlink(path)
|
|
if os.path.isabs(target):
|
|
return lexical(target)
|
|
return lexical(path.parent / target)
|
|
|
|
|
|
def inside(path, root):
|
|
return path != root and root in path.parents
|
|
|
|
|
|
def main(argv=None):
|
|
parser = argparse.ArgumentParser(description=__doc__)
|
|
parser.add_argument('--dry-run', action='store_true', help='show changes without writing')
|
|
parser.add_argument('--check', action='store_true', help='exit 1 if links differ; do not write')
|
|
parser.add_argument('--global-dir', type=Path, default=Path(__file__).resolve().parents[1] / 'skills')
|
|
parser.add_argument('--jev-dir', type=Path, default=Path.home() / 'Projects/work/JEV/.agents/skills')
|
|
parser.add_argument('--target', action='append', type=Path, help='override target directories; repeatable')
|
|
args = parser.parse_args(argv)
|
|
sources = [lexical(args.global_dir), lexical(args.jev_dir)]
|
|
targets = args.target if args.target is not None else [Path.home() / p for p in ('.agents/skills', '.codex/skills', '.claude/skills')]
|
|
targets = [lexical(p) for p in targets]
|
|
# Reject destinations inside catalogs (or containing them) before any mutation.
|
|
for target in targets:
|
|
resolved_target = target.resolve()
|
|
if any(resolved_target == source.resolve() or
|
|
inside(resolved_target, source.resolve()) or
|
|
inside(source.resolve(), resolved_target) for source in sources):
|
|
parser.error(f'target overlaps a source catalog: {target}')
|
|
if not sources[0].is_dir():
|
|
parser.error(f'global skill source unavailable: {sources[0]}')
|
|
available = [p for p in sources if p.is_dir()]
|
|
expected = {}
|
|
errors = []
|
|
for root in available:
|
|
for child in sorted(root.iterdir()):
|
|
if not child.is_dir():
|
|
continue
|
|
if not re.fullmatch(r'[a-z0-9]+(?:-[a-z0-9]+)*', child.name) or not (child / 'SKILL.md').is_file():
|
|
errors.append(f'invalid skill directory: {child}')
|
|
continue
|
|
if child.name in expected:
|
|
errors.append(f'colliding skill name: {child.name} ({expected[child.name]}, {child})')
|
|
else:
|
|
expected[child.name] = child
|
|
if errors:
|
|
print('\n'.join(errors), file=sys.stderr)
|
|
return 2
|
|
if len(available) != len(sources):
|
|
print(f'WARN: JEV source unavailable; preserving its existing links: {sources[1]}', file=sys.stderr)
|
|
changes = 0
|
|
conflicts = 0
|
|
readonly = args.check or args.dry_run
|
|
for target in targets:
|
|
if target.exists() and not target.is_dir():
|
|
print(f'SKIP: target is not a directory: {target}', file=sys.stderr)
|
|
conflicts += 1
|
|
continue
|
|
if not readonly:
|
|
target.mkdir(parents=True, exist_ok=True)
|
|
# Only prune links to sources that were successfully inventoried.
|
|
for dest in sorted(target.iterdir()) if target.is_dir() else []:
|
|
if dest.is_symlink() and any(inside(link_target(dest), root) for root in available) and dest.name not in expected:
|
|
print(f'remove stale: {dest}')
|
|
changes += 1
|
|
if not readonly:
|
|
dest.unlink()
|
|
for name, src in sorted(expected.items()):
|
|
dest = target / name
|
|
if dest.is_symlink():
|
|
current = link_target(dest)
|
|
if current == src:
|
|
continue
|
|
if not any(inside(current, root) for root in available):
|
|
print(f'SKIP: unrelated symlink: {dest}', file=sys.stderr)
|
|
conflicts += 1
|
|
continue
|
|
elif dest.exists():
|
|
print(f'SKIP: unrelated entry: {dest}', file=sys.stderr)
|
|
conflicts += 1
|
|
continue
|
|
print(f'link: {dest} -> {src}')
|
|
changes += 1
|
|
if not readonly:
|
|
if dest.is_symlink():
|
|
dest.unlink()
|
|
dest.symlink_to(src, target_is_directory=True)
|
|
print(f'{"check" if args.check else "dry-run" if args.dry_run else "sync"}: {changes} changes, {conflicts} conflicts')
|
|
return 1 if conflicts or (args.check and changes) else 0
|
|
|
|
|
|
if __name__ == '__main__':
|
|
sys.exit(main())
|