from argparse import ArgumentParser import contextlib from enum import Enum import logging import os import os.path import shutil import socket import sys import subprocess from pull.registry import MY_REPOS env = os.environ.copy() env['GIT_SSH_COMMAND'] = 'ssh -oStrictHostKeyChecking=no -oBatchMode=yes' DEFAULT_OUTPUT_DIR = 'output' DEFAULT_CGIT_CLONE_USER = 'egor' DEFAULT_CGIT_CLONE_HOST = 'tensin-ext1.home' DEFAULT_CGIT_CLONE_PORT = 8080 def set_up_logging(): logging.basicConfig( level=logging.DEBUG, datefmt='%Y-%m-%d %H:%M:%S', format='%(asctime)s | %(levelname)s | %(message)s') def parse_args(argv=None): if argv is None: argv = sys.argv[1:] parser = ArgumentParser() parser.add_argument('--output', default=DEFAULT_OUTPUT_DIR, help='output directory path') parser.add_argument('--cgit-user', default=DEFAULT_CGIT_CLONE_USER, help='cgit clone username') parser.add_argument('--cgit-host', default=DEFAULT_CGIT_CLONE_HOST, help='cgit clone host') parser.add_argument('--cgit-port', default=DEFAULT_CGIT_CLONE_PORT, help='cgit clone port number', type=int) return parser.parse_args(argv) def check_output(*args, stdout=subprocess.PIPE): result = subprocess.run(args, stdout=stdout, stderr=subprocess.STDOUT, env=env, encoding='utf-8') try: result.check_returncode() if stdout != subprocess.DEVNULL: if result.stdout is None: logging.debug('%s', args) else: logging.debug('%s\n%s', args, result.stdout) return result.returncode == 0, result.stdout except subprocess.CalledProcessError as e: if stdout != subprocess.DEVNULL: logging.error('%s\n%s', e, e.output) return e.returncode == 0, e.output def run(*args, discard_output=False): if discard_output: success, _ = check_output(*args, stdout=subprocess.DEVNULL) else: success, _ = check_output(*args) return success @contextlib.contextmanager def chdir(new_cwd): old_cwd = os.getcwd() os.chdir(new_cwd) try: yield finally: os.chdir(old_cwd) class CGit: def __init__(self, user, host, port): self.user = user self.host = host self.ip = socket.gethostbyname(self.host) self.port = port def get_clone_url(self, repo): return f'http://{self.user}@{self.ip}:{self.port}/git/{repo.repo_id}' class CGitRC: def __init__(self, cgit): self.cgit = cgit def write(self, path, repo): with open(path, 'w') as fd: self._write_field(fd, 'clone-url', self._build_clone_url(repo)) self._write_field(fd, 'owner', repo.owner) self._write_field(fd, 'desc', repo.desc) self._write_field(fd, 'homepage', repo.homepage) @staticmethod def _write_field(fd, field, value): if value is None: return fd.write(f'{field}={value}\n') def _build_clone_url(self, repo): clone_urls = [] if repo.clone_url is not None: clone_urls.append(repo.clone_url) clone_urls.append(self.cgit.get_clone_url(repo)) clone_urls = ' '.join(clone_urls) return clone_urls class Output: def __init__(self, output_dir, cgit): self.output_dir = self._make_dir(output_dir) self.cgitrc = CGitRC(cgit) @staticmethod def _make_dir(rel_path): abs_path = os.path.abspath(rel_path) os.makedirs(abs_path, exist_ok=True) return abs_path def get_repo_dir(self, repo): return os.path.join(self.output_dir, repo.repo_id) def get_cgitrc_path(self, repo): return os.path.join(self.get_repo_dir(repo), 'cgitrc') def pull(self, repo): success = False verdict = self.judge(repo) if verdict is RepoVerdict.SHOULD_MIRROR: success = self.mirror(repo) elif verdict is RepoVerdict.SHOULD_UPDATE: success = self.update(repo) elif verdict is RepoVerdict.CANT_DECIDE: success = False else: raise NotImplementedError(f'Unknown repository verdict: {verdict}') if success: self.cgitrc.write(self.get_cgitrc_path(repo), repo) return success def judge(self, repo): repo_dir = self.get_repo_dir(repo) if not os.path.isdir(repo_dir): return RepoVerdict.SHOULD_MIRROR with chdir(repo_dir): if not run('git', 'rev-parse', '--is-inside-work-tree', discard_output=True): logging.warning('Not a repository, so going to mirror: %s', repo_dir) return RepoVerdict.SHOULD_MIRROR success, output = check_output('git', 'config', '--get', 'remote.origin.url') if not success: # Every repository managed by this script should have the # 'origin' remote. If it doesn't, it's trash. return RepoVerdict.SHOULD_MIRROR if f'{repo.clone_url}\n' != output: logging.warning("Existing repository '%s' URL doesn't match the specified clone" \ " URL: %s", repo.repo_id, repo.clone_url) return RepoVerdict.CANT_DECIDE # Looks like a legit clone of the specified remote. return RepoVerdict.SHOULD_UPDATE def mirror(self, repo): logging.info("Mirroring repository '%s' from: %s", repo.repo_id, repo.clone_url) repo_dir = self.get_repo_dir(repo) if os.path.isdir(repo_dir): try: shutil.rmtree(repo_dir) except Exception as e: logging.exception(e) return False return run('git', 'clone', '--mirror', repo.clone_url, repo_dir) def update(self, repo): logging.info("Updating repository '%s'", repo.repo_id) repo_dir = self.get_repo_dir(repo) with chdir(repo_dir): if not run('git', 'remote', 'update', '--prune'): return False if run('git', 'rev-parse', '--verify', '--quiet', 'origin/master', discard_output=True): if not run('git', 'reset', '--soft', 'origin/master'): return False return True class RepoVerdict(Enum): SHOULD_MIRROR = 1 SHOULD_UPDATE = 2 CANT_DECIDE = 3 def main(args=None): set_up_logging() try: args = parse_args(args) cgit = CGit(args.cgit_user, args.cgit_host, args.cgit_port) output = Output(args.output, cgit) success = True for repo in MY_REPOS: if not output.pull(repo): success = False if success: logging.info('All repositories were updated successfully') return 0 else: logging.warning("Some repositories couldn't be updated!") return 1 except Exception as e: logging.exception(e) raise if __name__ == '__main__': sys.exit(main())