codearchiver/codearchiver/modules/git.py

import codearchiver.core
import datetime
import logging
import os.path
import shutil
import subprocess


logger = logging.getLogger(__name__)


class Git(codearchiver.core.Module):
	name = 'git'

	@staticmethod
	def matches(inputUrl):
		return inputUrl.url.endswith('.git')

	def __init__(self, *args, extraBranches = {}, **kwargs):
		super().__init__(*args, **kwargs)
		self._extraBranches = extraBranches

	def process(self):
		directory = self._url.rsplit('/', 1)[1]
		if os.path.exists(directory):
			logger.fatal(f'{directory!r} already exists')
			return
		startTime = datetime.datetime.utcnow()
		if self._id is None:
			self._id = f'git_{self._url.replace("/", "_")}_{startTime:%Y%m%dT%H%M%SZ}'
		bundle = f'{self._id}.bundle'
		if os.path.exists(bundle):
			logger.fatal(f'{bundle!r} already exists')
			return

		logger.info(f'Cloning {self._url} into {directory}')
		subprocess.run(['git', 'clone', '--verbose', '--mirror', self._url, directory], check = True)

		if self._extraBranches:
			for branch, commit in self._extraBranches.items():
				logger.info(f'Fetching commit {commit} as {branch}')
				r = subprocess.run(['git', 'fetch', '--verbose', 'origin', commit], cwd = directory)
				if r.returncode == 0:
					r2 = subprocess.run(['git', 'update-ref', f'refs/codearchiver/{branch}', commit, ''], cwd = directory)
					if r2.returncode != 0:
						logger.error(f'Failed to update-ref refs/codearchiver/{branch} to {commit}')
				else:
					logger.error(f'Failed to fetch {commit}')

		logger.info(f'Bundling into {bundle}')
		subprocess.run(['git', 'bundle', 'create', f'../{bundle}', '--all'], cwd = directory, check = True)

		logger.info(f'Removing clone')
		shutil.rmtree(directory)

		return codearchiver.core.Result(id = self._id, files = [bundle])

	def __repr__(self):
		return f'{type(self).__module__}.{type(self).__name__}({self._inputUrl!r}, extraBranches = {self._extraBranches!r})'