176 lines
7.1 KiB
Python
176 lines
7.1 KiB
Python
|
|
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') |