codearchiver/codearchiver/modules/git.py

import codearchiver.core
import codearchiver.subprocess
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}')
		codearchiver.subprocess.run_with_log(['git', 'clone', '--verbose', '--mirror', self._url, directory])

		if self._extraBranches:
			for branch, commit in self._extraBranches.items():
				logger.info(f'Fetching commit {commit} as {branch}')
				r = codearchiver.subprocess.run_with_log(['git', 'fetch', '--verbose', '--progress', 'origin', commit], cwd = directory, check = False)
				if r.returncode == 0:
					r2 = codearchiver.subprocess.run_with_log(['git', 'update-ref', f'refs/codearchiver/{branch}', commit, ''], cwd = directory, check = False)
					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}')
		codearchiver.subprocess.run_with_log(['git', 'bundle', 'create', '--progress', f'../{bundle}', '--all'], cwd = directory)

		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})'