226 lines
9.1 KiB
Python
226 lines
9.1 KiB
Python
import os
|
|
import sys
|
|
import subprocess
|
|
import shutil
|
|
import logging
|
|
from pathlib import Path
|
|
import importlib.util
|
|
import traceback
|
|
|
|
class GitInstaller:
|
|
"""
|
|
A ComfyUI custom node that can install GitHub repositories and their requirements automatically.
|
|
"""
|
|
|
|
@classmethod
|
|
def INPUT_TYPES(cls):
|
|
return {
|
|
"required": {
|
|
"github_url": ("STRING", {
|
|
"default": "https://github.com/username/repo.git",
|
|
"multiline": False,
|
|
"placeholder": "Enter GitHub repository URL"
|
|
}),
|
|
"install_requirements": ("BOOLEAN", {"default": True}),
|
|
"force_reinstall": ("BOOLEAN", {"default": False}),
|
|
},
|
|
"optional": {
|
|
"custom_folder_name": ("STRING", {
|
|
"default": "",
|
|
"multiline": False,
|
|
"placeholder": "Optional: custom folder name"
|
|
}),
|
|
}
|
|
}
|
|
|
|
RETURN_TYPES = ("STRING", "BOOLEAN", "STRING")
|
|
RETURN_NAMES = ("status_message", "success", "installed_path")
|
|
FUNCTION = "install_repository"
|
|
CATEGORY = "utils/git"
|
|
|
|
def __init__(self):
|
|
# Get the custom_nodes directory relative to this file
|
|
self.custom_nodes_dir = Path(__file__).parent.parent.absolute()
|
|
|
|
def get_repo_name_from_url(self, github_url):
|
|
"""Extract repository name from GitHub URL"""
|
|
try:
|
|
# Handle different URL formats
|
|
if github_url.endswith('.git'):
|
|
repo_name = github_url.split('/')[-1][:-4] # Remove .git
|
|
else:
|
|
repo_name = github_url.split('/')[-1]
|
|
return repo_name
|
|
except:
|
|
return None
|
|
|
|
def check_git_available(self):
|
|
"""Check if git is available in the system"""
|
|
try:
|
|
subprocess.run(['git', '--version'], capture_output=True, check=True)
|
|
return True
|
|
except (subprocess.CalledProcessError, FileNotFoundError):
|
|
return False
|
|
|
|
def clone_repository(self, github_url, target_path):
|
|
"""Clone the GitHub repository"""
|
|
try:
|
|
cmd = ['git', 'clone', github_url, str(target_path)]
|
|
result = subprocess.run(cmd, capture_output=True, text=True, check=True)
|
|
return True, f"Repository cloned successfully: {result.stdout}"
|
|
except subprocess.CalledProcessError as e:
|
|
return False, f"Git clone failed: {e.stderr}"
|
|
except Exception as e:
|
|
return False, f"Unexpected error during clone: {str(e)}"
|
|
|
|
def install_requirements(self, repo_path):
|
|
"""Install requirements.txt if it exists"""
|
|
requirements_path = repo_path / "requirements.txt"
|
|
|
|
if not requirements_path.exists():
|
|
return True, "No requirements.txt found - skipping dependency installation"
|
|
|
|
try:
|
|
# Read requirements to check if it's empty or just comments
|
|
with open(requirements_path, 'r', encoding='utf-8') as f:
|
|
content = f.read().strip()
|
|
|
|
# Check if file has actual requirements (not just comments/empty lines)
|
|
lines = [line.strip() for line in content.split('\n')
|
|
if line.strip() and not line.strip().startswith('#')]
|
|
|
|
if not lines:
|
|
return True, "Requirements.txt contains no packages - skipping installation"
|
|
|
|
# Install requirements using pip
|
|
cmd = [sys.executable, '-m', 'pip', 'install', '-r', str(requirements_path)]
|
|
result = subprocess.run(cmd, capture_output=True, text=True, check=True)
|
|
|
|
return True, f"Requirements installed successfully: {result.stdout}"
|
|
|
|
except subprocess.CalledProcessError as e:
|
|
return False, f"Failed to install requirements: {e.stderr}"
|
|
except Exception as e:
|
|
return False, f"Error reading/installing requirements: {str(e)}"
|
|
|
|
def reload_custom_nodes(self):
|
|
"""Attempt to reload the custom nodes system"""
|
|
try:
|
|
# Import nodes module to trigger reload
|
|
import nodes
|
|
if hasattr(nodes, 'init_external_custom_nodes'):
|
|
# This would ideally reload custom nodes, but it's complex
|
|
# For now, we'll just notify the user
|
|
return "Note: Restart ComfyUI to load the newly installed custom node"
|
|
return ""
|
|
except Exception as e:
|
|
return f"Note: Could not auto-reload nodes: {str(e)}"
|
|
|
|
def clean_github_url(self, url):
|
|
"""Clean and normalize GitHub URL to proper git clone format"""
|
|
try:
|
|
url = url.strip()
|
|
|
|
# Remove common web interface suffixes
|
|
suffixes_to_remove = ['/tree/main/', '/tree/master/', '/tree/main', '/tree/master', '/blob/main/', '/blob/master/']
|
|
for suffix in suffixes_to_remove:
|
|
if url.endswith(suffix):
|
|
url = url[:-len(suffix)]
|
|
break
|
|
|
|
# Ensure .git extension for HTTPS URLs
|
|
if url.startswith(('https://github.com/', 'http://github.com/')):
|
|
if not url.endswith('.git'):
|
|
url += '.git'
|
|
|
|
return url
|
|
except:
|
|
return url
|
|
|
|
def validate_github_url(self, url):
|
|
"""Validate if the URL is a valid GitHub repository URL"""
|
|
try:
|
|
url = url.strip().lower()
|
|
if not url.startswith(('https://github.com/', 'http://github.com/', 'git@github.com:')):
|
|
return False, "URL must be a GitHub repository URL"
|
|
if url.startswith('git@github.com:') and not url.endswith('.git'):
|
|
return False, "SSH URLs must end with .git"
|
|
return True, ""
|
|
except:
|
|
return False, "Invalid URL format"
|
|
|
|
def install_repository(self, github_url, install_requirements=True, force_reinstall=False, custom_folder_name=""):
|
|
"""Main function to install a GitHub repository"""
|
|
|
|
try:
|
|
# Validate inputs
|
|
if not github_url or not github_url.strip():
|
|
return "Error: GitHub URL is required", False, ""
|
|
|
|
github_url = github_url.strip()
|
|
|
|
# Validate GitHub URL
|
|
url_valid, url_error = self.validate_github_url(github_url)
|
|
if not url_valid:
|
|
return f"Error: {url_error}", False, ""
|
|
|
|
# Check if git is available
|
|
if not self.check_git_available():
|
|
return "Error: Git is not available. Please install Git first.", False, ""
|
|
|
|
# Determine repository name and target path
|
|
if custom_folder_name and custom_folder_name.strip():
|
|
repo_name = custom_folder_name.strip()
|
|
# Sanitize folder name
|
|
repo_name = "".join(c for c in repo_name if c.isalnum() or c in ('-', '_', '.'))
|
|
if not repo_name:
|
|
return "Error: Invalid custom folder name", False, ""
|
|
else:
|
|
repo_name = self.get_repo_name_from_url(github_url)
|
|
if not repo_name:
|
|
return "Error: Could not extract repository name from URL", False, ""
|
|
|
|
target_path = self.custom_nodes_dir / repo_name
|
|
|
|
# Check if repository already exists
|
|
if target_path.exists():
|
|
if not force_reinstall:
|
|
return f"Repository '{repo_name}' already exists. Use force_reinstall=True to overwrite.", False, str(target_path)
|
|
else:
|
|
# Remove existing directory
|
|
try:
|
|
shutil.rmtree(target_path)
|
|
status_messages = [f"✓ Removed existing directory: {repo_name}"]
|
|
except Exception as e:
|
|
return f"Error: Could not remove existing directory: {str(e)}", False, ""
|
|
else:
|
|
status_messages = []
|
|
|
|
# Clone the repository
|
|
clone_success, clone_message = self.clone_repository(github_url, target_path)
|
|
if not clone_success:
|
|
return f"Clone failed: {clone_message}", False, ""
|
|
|
|
status_messages.append(f"✓ Repository cloned to: {target_path}")
|
|
|
|
# Install requirements if requested
|
|
if install_requirements:
|
|
req_success, req_message = self.install_requirements(target_path)
|
|
if req_success:
|
|
status_messages.append(f"✓ {req_message}")
|
|
else:
|
|
status_messages.append(f"⚠ {req_message}")
|
|
|
|
# Add reload note
|
|
reload_note = self.reload_custom_nodes()
|
|
if reload_note:
|
|
status_messages.append(reload_note)
|
|
|
|
final_message = "\n".join(status_messages)
|
|
return final_message, True, str(target_path)
|
|
|
|
except Exception as e:
|
|
error_msg = f"Unexpected error: {str(e)}\n{traceback.format_exc()}"
|
|
logging.error(f"GitInstaller error: {error_msg}")
|
|
return error_msg, False, ""
|