mirror of
https://github.com/cexll/myclaude.git
synced 2026-02-05 02:30:26 +08:00
feat: add hooks management to install.py
- Add merge_hooks_to_settings: merge module hooks into settings.json - Add unmerge_hooks_from_settings: remove module hooks from settings.json - Add find_module_hooks: detect hooks.json in module directories - Update execute_module: auto-merge hooks after install - Update uninstall_module: auto-remove hooks on uninstall - Add __module__ marker to track hook ownership Generated with SWE-Agent.ai Co-Authored-By: SWE-Agent.ai <noreply@swe-agent.ai>
This commit is contained in:
147
install.py
147
install.py
@@ -23,6 +23,7 @@ except ImportError: # pragma: no cover
|
||||
jsonschema = None
|
||||
|
||||
DEFAULT_INSTALL_DIR = "~/.claude"
|
||||
SETTINGS_FILE = "settings.json"
|
||||
|
||||
|
||||
def _ensure_list(ctx: Dict[str, Any], key: str) -> List[Any]:
|
||||
@@ -91,6 +92,132 @@ def _load_json(path: Path) -> Any:
|
||||
raise ValueError(f"Invalid JSON in {path}: {exc}") from exc
|
||||
|
||||
|
||||
def _save_json(path: Path, data: Any) -> None:
|
||||
"""Save data to JSON file with proper formatting."""
|
||||
path.parent.mkdir(parents=True, exist_ok=True)
|
||||
with path.open("w", encoding="utf-8") as fh:
|
||||
json.dump(data, fh, indent=2, ensure_ascii=False)
|
||||
fh.write("\n")
|
||||
|
||||
|
||||
# =============================================================================
|
||||
# Hooks Management
|
||||
# =============================================================================
|
||||
|
||||
def load_settings(ctx: Dict[str, Any]) -> Dict[str, Any]:
|
||||
"""Load settings.json from install directory."""
|
||||
settings_path = ctx["install_dir"] / SETTINGS_FILE
|
||||
if settings_path.exists():
|
||||
try:
|
||||
return _load_json(settings_path)
|
||||
except (ValueError, FileNotFoundError):
|
||||
return {}
|
||||
return {}
|
||||
|
||||
|
||||
def save_settings(ctx: Dict[str, Any], settings: Dict[str, Any]) -> None:
|
||||
"""Save settings.json to install directory."""
|
||||
settings_path = ctx["install_dir"] / SETTINGS_FILE
|
||||
_save_json(settings_path, settings)
|
||||
|
||||
|
||||
def find_module_hooks(module_name: str, cfg: Dict[str, Any], ctx: Dict[str, Any]) -> Optional[Dict[str, Any]]:
|
||||
"""Find hooks.json for a module if it exists."""
|
||||
# Check for hooks in operations (copy_dir targets)
|
||||
for op in cfg.get("operations", []):
|
||||
if op.get("type") == "copy_dir":
|
||||
target_dir = ctx["install_dir"] / op["target"]
|
||||
hooks_file = target_dir / "hooks" / "hooks.json"
|
||||
if hooks_file.exists():
|
||||
try:
|
||||
return _load_json(hooks_file)
|
||||
except (ValueError, FileNotFoundError):
|
||||
pass
|
||||
|
||||
# Also check source directory during install
|
||||
for op in cfg.get("operations", []):
|
||||
if op.get("type") == "copy_dir":
|
||||
source_dir = ctx["config_dir"] / op["source"]
|
||||
hooks_file = source_dir / "hooks" / "hooks.json"
|
||||
if hooks_file.exists():
|
||||
try:
|
||||
return _load_json(hooks_file)
|
||||
except (ValueError, FileNotFoundError):
|
||||
pass
|
||||
|
||||
return None
|
||||
|
||||
|
||||
def _create_hook_marker(module_name: str) -> str:
|
||||
"""Create a marker to identify hooks from a specific module."""
|
||||
return f"__module:{module_name}__"
|
||||
|
||||
|
||||
def merge_hooks_to_settings(module_name: str, hooks_config: Dict[str, Any], ctx: Dict[str, Any]) -> None:
|
||||
"""Merge module hooks into settings.json."""
|
||||
settings = load_settings(ctx)
|
||||
settings.setdefault("hooks", {})
|
||||
|
||||
module_hooks = hooks_config.get("hooks", {})
|
||||
marker = _create_hook_marker(module_name)
|
||||
|
||||
for hook_type, hook_entries in module_hooks.items():
|
||||
settings["hooks"].setdefault(hook_type, [])
|
||||
|
||||
for entry in hook_entries:
|
||||
# Add marker to identify this hook's source module
|
||||
entry_copy = dict(entry)
|
||||
entry_copy["__module__"] = module_name
|
||||
|
||||
# Check if already exists (avoid duplicates)
|
||||
exists = False
|
||||
for existing in settings["hooks"][hook_type]:
|
||||
if existing.get("__module__") == module_name:
|
||||
# Same module, check if same hook
|
||||
if _hooks_equal(existing, entry_copy):
|
||||
exists = True
|
||||
break
|
||||
|
||||
if not exists:
|
||||
settings["hooks"][hook_type].append(entry_copy)
|
||||
|
||||
save_settings(ctx, settings)
|
||||
write_log({"level": "INFO", "message": f"Merged hooks for module: {module_name}"}, ctx)
|
||||
|
||||
|
||||
def unmerge_hooks_from_settings(module_name: str, ctx: Dict[str, Any]) -> None:
|
||||
"""Remove module hooks from settings.json."""
|
||||
settings = load_settings(ctx)
|
||||
|
||||
if "hooks" not in settings:
|
||||
return
|
||||
|
||||
modified = False
|
||||
for hook_type in list(settings["hooks"].keys()):
|
||||
original_len = len(settings["hooks"][hook_type])
|
||||
settings["hooks"][hook_type] = [
|
||||
entry for entry in settings["hooks"][hook_type]
|
||||
if entry.get("__module__") != module_name
|
||||
]
|
||||
if len(settings["hooks"][hook_type]) < original_len:
|
||||
modified = True
|
||||
|
||||
# Remove empty hook type arrays
|
||||
if not settings["hooks"][hook_type]:
|
||||
del settings["hooks"][hook_type]
|
||||
|
||||
if modified:
|
||||
save_settings(ctx, settings)
|
||||
write_log({"level": "INFO", "message": f"Removed hooks for module: {module_name}"}, ctx)
|
||||
|
||||
|
||||
def _hooks_equal(hook1: Dict[str, Any], hook2: Dict[str, Any]) -> bool:
|
||||
"""Compare two hooks ignoring the __module__ marker."""
|
||||
h1 = {k: v for k, v in hook1.items() if k != "__module__"}
|
||||
h2 = {k: v for k, v in hook2.items() if k != "__module__"}
|
||||
return h1 == h2
|
||||
|
||||
|
||||
def load_config(path: str) -> Dict[str, Any]:
|
||||
"""Load config and validate against JSON Schema.
|
||||
|
||||
@@ -345,7 +472,7 @@ def interactive_select_modules(config: Dict[str, Any]) -> Dict[str, Any]:
|
||||
|
||||
|
||||
def uninstall_module(name: str, cfg: Dict[str, Any], ctx: Dict[str, Any]) -> Dict[str, Any]:
|
||||
"""Uninstall a module by removing its files."""
|
||||
"""Uninstall a module by removing its files and hooks."""
|
||||
result: Dict[str, Any] = {
|
||||
"module": name,
|
||||
"status": "success",
|
||||
@@ -371,6 +498,13 @@ def uninstall_module(name: str, cfg: Dict[str, Any], ctx: Dict[str, Any]) -> Dic
|
||||
except Exception as exc:
|
||||
write_log({"level": "WARNING", "message": f"Failed to remove {op.get('target', 'unknown')}: {exc}"}, ctx)
|
||||
|
||||
# Remove module hooks from settings.json
|
||||
try:
|
||||
unmerge_hooks_from_settings(name, ctx)
|
||||
result["hooks_removed"] = True
|
||||
except Exception as exc:
|
||||
write_log({"level": "WARNING", "message": f"Failed to remove hooks for {name}: {exc}"}, ctx)
|
||||
|
||||
result["removed_paths"] = removed_paths
|
||||
return result
|
||||
|
||||
@@ -572,6 +706,17 @@ def execute_module(name: str, cfg: Dict[str, Any], ctx: Dict[str, Any]) -> Dict[
|
||||
)
|
||||
raise
|
||||
|
||||
# Handle hooks: find and merge module hooks into settings.json
|
||||
hooks_config = find_module_hooks(name, cfg, ctx)
|
||||
if hooks_config:
|
||||
try:
|
||||
merge_hooks_to_settings(name, hooks_config, ctx)
|
||||
result["operations"].append({"type": "merge_hooks", "status": "success"})
|
||||
result["has_hooks"] = True
|
||||
except Exception as exc:
|
||||
write_log({"level": "WARNING", "message": f"Failed to merge hooks for {name}: {exc}"}, ctx)
|
||||
result["operations"].append({"type": "merge_hooks", "status": "failed", "error": str(exc)})
|
||||
|
||||
return result
|
||||
|
||||
|
||||
|
||||
Reference in New Issue
Block a user