from git import Repo from git.exc import GitError import hashlib import inspect import os import shutil from SCons.Script import * Import('env') def _clone(env: Environment, repo_name: str, remote_url: str): repo_dir = os.path.join(env['CLONE_DIR'], 'git', repo_name, '_bare') try: repo = Repo(repo_dir) origin = repo.remotes['origin'] except GitError: print(f'Initializing git repository at {repo_dir}.') repo = Repo.init(repo_dir, bare=True) origin = repo.create_remote('origin', remote_url) return repo, origin def _git_branch(env: Environment, repo_name: str, remote_url: str, git_ref: str = 'main') -> dict: repo, origin = _clone(env, repo_name, remote_url) old_worktree_dir = os.path.join(env['CLONE_DIR'], 'git', repo_name, hashlib.shake_128(git_ref.encode('utf-8')).hexdigest(6)) worktree_dir = os.path.join(env['CLONE_DIR'], 'git', repo_name, git_ref.replace('/', '_')) if os.path.exists(old_worktree_dir) and not os.path.islink(old_worktree_dir): if not os.path.exists(worktree_dir): print(f'Found old Git worktree at {old_worktree_dir}, moving it to {worktree_dir}.') try: repo.git.worktree('move', old_worktree_dir, worktree_dir) except GitError: print('Error while moving worktree, manually moving and repairing it instead.') shutil.move(old_worktree_dir, worktree_dir) try: repo.git.worktree('repair', worktree_dir) except GitError: print('Also didn\'t work, removing and redownloading it.') try: repo.git.worktree('remove', '-f', worktree_dir) except GitError: ... try: repo.git.worktree('remove', '-f', old_worktree_dir) except GitError: ... if os.path.exists(worktree_dir): shutil.rmtree(worktree_dir, ignore_errors=True) # this is all we can do, I guess else: print(f'Found old Git worktree at {old_worktree_dir}, but the new one at {worktree_dir} already exists. Removing the old one.') repo.git.worktree('remove', '-f', old_worktree_dir) print('Attempting to create a symlink for older S++ versions.') try: os.symlink(worktree_dir, old_worktree_dir, target_is_directory=True) except Exception as e: print(f'Failed: {e}') update_submodules = False if not os.path.exists(worktree_dir): print(f'Checking out into {worktree_dir}.') origin.fetch(tags=True, force=True) os.makedirs(worktree_dir) repo.git.worktree('add', worktree_dir, git_ref) worktree_repo = Repo(worktree_dir) update_submodules = True elif env['UPDATE_REPOSITORIES']: worktree_repo = Repo(worktree_dir) if not worktree_repo.head.is_detached: print(f'Updating git repository at {worktree_dir}') worktree_origin = worktree_repo.remotes['origin'] worktree_origin.pull() update_submodules = True else: print(f'Not updating git repository {worktree_dir} as it is not on a branch.') else: worktree_repo = Repo(worktree_dir) if update_submodules: for submodule in worktree_repo.submodules: submodule.update(init=True) for submodule in worktree_repo.submodules: if os.listdir(submodule.abspath) == ['.git']: print(f'Submodule {submodule.name} seems borked, attempting to fix it.') worktree_repo.git.submodule('deinit', '-f', submodule.path) worktree_repo.git.submodule('init', submodule.path) worktree_repo.git.submodule('update', submodule.path) return { 'checkout_root': worktree_dir, 'repo': repo, 'origin': origin } def _git_tags(env: Environment, repo_name: str, remote_url: str, force_fetch: bool = False) -> 'list[str]': repo, origin = _clone(env, repo_name, remote_url) if force_fetch or env['UPDATE_REPOSITORIES']: try: origin.fetch(tags=True) except GitError: env.Warn(f'Error fetching tags from {repo_name} ({remote_url})') return [t.name for t in repo.tags] def _make_callable(val): if callable(val): return val else: def _wrapped(*args, **kwargs): return val return _wrapped def _git_recipe(env: Environment, globals: dict, repo_name, repo_url, cook_fn, versions = None, tag_pattern = None, tag_fn = None, ref_fn = None, dependencies: dict = {}) -> None: _repo_name = _make_callable(repo_name) _repo_url = _make_callable(repo_url) _tag_pattern = _make_callable(tag_pattern) versions_cb = versions and _make_callable(versions) dependencies_cb = _make_callable(dependencies) def _versions(env: Environment, update: bool = False, options: dict = {}): if 'ref' in options: return [(0, 0, 0)] # no versions if compiling from a branch pattern_signature = inspect.signature(_tag_pattern) kwargs = {} if 'options' in pattern_signature.parameters: kwargs['options'] = options pattern = _tag_pattern(env, **kwargs) if pattern: tags = env.GitTags(repo_name = _repo_name(env), remote_url = _repo_url(env), force_fetch=update) result = [] for tag in tags: match = pattern.match(tag) if match: result.append(tuple(int(part) for part in match.groups() if part is not None)) if len(result) == 0 and not update: return _versions(env, update=True) return result elif versions_cb: return versions_cb(env) else: return [(0, 0, 0)] def _dependencies(env: Environment, version, options: dict) -> 'dict': dependencies_signature = inspect.signature(dependencies_cb) kwargs = {} if 'options' in dependencies_signature.parameters: kwargs['options'] = options return dependencies_cb(env, version, **kwargs) def _cook(env: Environment, version, options: dict = {}) -> dict: if 'ref' in options: git_ref = options['ref'] elif tag_fn: tag_signature = inspect.signature(tag_fn) kwargs = {} if 'options' in tag_signature.parameters: kwargs['options'] = options git_ref = f'refs/tags/{tag_fn(version, **kwargs)}' else: assert ref_fn git_ref = ref_fn(env, version) repo = env.GitBranch(repo_name = _repo_name(env), remote_url = _repo_url(env), git_ref = git_ref) cook_signature = inspect.signature(cook_fn) kwargs = {} if 'options' in cook_signature.parameters: kwargs['options'] = options return cook_fn(env, repo, **kwargs) globals['versions'] = _versions globals['dependencies'] = _dependencies globals['cook'] = _cook env.AddMethod(_git_branch, 'GitBranch') env.AddMethod(_git_tags, 'GitTags') env.AddMethod(_git_recipe, 'GitRecipe') Return('env')