aboutsummaryrefslogtreecommitdiffstatshomepage
path: root/pull.py
diff options
context:
space:
mode:
Diffstat (limited to 'pull.py')
-rwxr-xr-xpull.py300
1 files changed, 300 insertions, 0 deletions
diff --git a/pull.py b/pull.py
new file mode 100755
index 0000000..3649139
--- /dev/null
+++ b/pull.py
@@ -0,0 +1,300 @@
+#!/usr/bin/env python3
+
+import contextlib
+from enum import Enum
+import logging
+import os
+import os.path
+import shutil
+import socket
+import sys
+import subprocess
+
+env = os.environ.copy()
+env['GIT_SSH_COMMAND'] = 'ssh -oStrictHostKeyChecking=no -oBatchMode=yes'
+
+CGIT_CLONE_USER = 'egor'
+CGIT_CLONE_HOST = 'tensin-ext1.home'
+CGIT_CLONE_IP = '127.0.0.1'
+
+REPOS_DIR = 'repos'
+
+DEFAULT_OWNER = 'Egor Tensin'
+DEFAULT_GITHUB_USER = 'egor-tensin'
+DEFAULT_BITBUCKET_USER = 'egor-tensin'
+
+
+def set_up_logging():
+ logging.basicConfig(
+ level=logging.DEBUG,
+ datefmt='%Y-%m-%d %H:%M:%S',
+ format='%(asctime)s | %(levelname)s | %(message)s')
+
+
+def make_dir(rel_path):
+ script_path = os.path.abspath(__file__)
+ script_dir = os.path.dirname(script_path)
+ abs_path = os.path.join(script_dir, rel_path)
+ os.makedirs(abs_path, exist_ok=True)
+ return abs_path
+
+
+@contextlib.contextmanager
+def chdir(new_cwd):
+ old_cwd = os.getcwd()
+ os.chdir(new_cwd)
+ try:
+ yield
+ finally:
+ os.chdir(old_cwd)
+
+
+def extract_repo_name(repo_id):
+ return os.path.basename(repo_id)
+
+
+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 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:
+ 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
+
+
+class RepoVerdict(Enum):
+ SHOULD_MIRROR = 1
+ SHOULD_UPDATE = 2
+ CANT_DECIDE = 3
+
+
+class Repo:
+ def __init__(self, repo_id, clone_url, owner=None, desc=None,
+ homepage=None):
+ self.repo_id = repo_id
+ self.repo_name = extract_repo_name(repo_id)
+ self.repo_dir = os.path.join(REPOS_DIR, self.repo_id)
+ self.clone_url = clone_url
+ if owner is None:
+ owner = DEFAULT_OWNER
+ self.owner = owner
+ if desc is None:
+ if homepage is not None:
+ desc = homepage
+ elif clone_url is not None:
+ desc = clone_url
+ else:
+ desc = self.repo_name
+ self.desc = desc
+ self.homepage = homepage
+
+ def write_cgitrc(self):
+ with open(self.get_cgitrc_path(), 'w') as fd:
+ self.write_cgitrc_field(fd, 'clone-url', self.build_cgitrc_clone_url())
+ self.write_cgitrc_field(fd, 'owner', self.owner)
+ self.write_cgitrc_field(fd, 'desc', self.desc)
+ self.write_cgitrc_field(fd, 'homepage', self.homepage)
+
+ def build_cgitrc_clone_url(self):
+ clone_urls = []
+ if self.clone_url is not None:
+ clone_urls.append(self.clone_url)
+ clone_urls.append(self.build_cgit_clone_url())
+ clone_urls = ' '.join(clone_urls)
+ return clone_urls
+
+ def write_cgitrc_field(self, fd, field, value):
+ if value is None:
+ return
+ fd.write(f'{field}={value}\n')
+
+ def build_cgit_clone_url(self):
+ return f'http://{CGIT_CLONE_USER}@{CGIT_CLONE_IP}:8080/git/{self.repo_id}'
+
+ def get_cgitrc_path(self):
+ return os.path.join(self.repo_dir, 'cgitrc')
+
+ def pull(self):
+ success = False
+ verdict = self.judge()
+ if verdict is RepoVerdict.SHOULD_MIRROR:
+ success = self.mirror()
+ elif verdict is RepoVerdict.SHOULD_UPDATE:
+ success = self.update()
+ elif verdict is RepoVerdict.CANT_DECIDE:
+ success = False
+ else:
+ raise NotImplementedError(f'Unknown repository verdict: {verdict}')
+ if success:
+ self.write_cgitrc()
+ return success
+
+ def judge(self):
+ if not os.path.isdir(self.repo_dir):
+ return RepoVerdict.SHOULD_MIRROR
+ with chdir(self.repo_dir):
+ if not run('git', 'rev-parse', '--is-inside-work-tree', discard_output=True):
+ # What is this directory?
+ 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'{self.clone_url}\n' != output:
+ logging.warning("Existing repository '%s' URL doesn't match" \
+ " the specified clone URL: %s", self.repo_id,
+ self.clone_url)
+ return RepoVerdict.CANT_DECIDE
+ # Looks like a legit clone of the specified remote.
+ return RepoVerdict.SHOULD_UPDATE
+
+ def mirror(self):
+ logging.info("Mirroring repository '%s' from: %s", self.repo_id,
+ self.clone_url)
+ if os.path.isdir(self.repo_dir):
+ try:
+ shutil.rmtree(self.repo_dir)
+ except Exception as e:
+ logging.exception(e)
+ return False
+ return run('git', 'clone', '--mirror', self.clone_url, self.repo_dir)
+
+ def update(self):
+ logging.info("Updating repository '%s'", self.repo_id)
+ with chdir(self.repo_dir):
+ return run('git', 'remote', 'update', '--prune')
+
+
+class GithubRepo(Repo):
+ def __init__(self, repo_id, clone_url=None, owner=None, desc=None,
+ homepage=None, github_user=DEFAULT_GITHUB_USER):
+ if clone_url is None:
+ clone_url = self.build_clone_url(github_user, repo_id)
+ if homepage is None:
+ homepage = self.build_homepage_url(github_user, repo_id)
+ super().__init__(repo_id, clone_url, owner=owner, desc=desc,
+ homepage=homepage)
+
+ @staticmethod
+ def build_clone_url(user, repo_id):
+ name = extract_repo_name(repo_id)
+ return f'ssh://git@github.com/{user}/{name}.git'
+
+ @staticmethod
+ def build_homepage_url(user, repo_id):
+ name = extract_repo_name(repo_id)
+ return f'https://github.com/egor-tensin/{name}'
+
+
+class BitbucketRepo(Repo):
+ def __init__(self, repo_id, clone_url=None, owner=None, desc=None,
+ homepage=None, bitbucket_user=DEFAULT_BITBUCKET_USER):
+ if clone_url is None:
+ clone_url = self.build_clone_url(bitbucket_user, repo_id)
+ if homepage is None:
+ homepage = self.build_homepage_url(bitbucket_user, repo_id)
+ super().__init__(repo_id, clone_url, owner=owner, desc=desc,
+ homepage=homepage)
+
+ @staticmethod
+ def build_clone_url(user, repo_id):
+ name = extract_repo_name(repo_id)
+ return f'ssh://git@bitbucket.org/{user}/{name}.git'
+
+ @staticmethod
+ def build_homepage_url(user, repo_id):
+ name = extract_repo_name(repo_id)
+ return f'https://bitbucket.org/egor-tensin/{name.lower()}'
+
+
+repos = (
+ GithubRepo('personal/aes-tools'),
+ GithubRepo('personal/blog'),
+ GithubRepo('personal/chess-games'),
+ GithubRepo('personal/cmake-common'),
+ GithubRepo('personal/config-links'),
+ GithubRepo('personal/cv'),
+ GithubRepo('personal/egor-tensin.github.io'),
+ GithubRepo('personal/filters'),
+ GithubRepo('personal/linux-home'),
+ BitbucketRepo('personal/measure_temp'),
+ GithubRepo('personal/notes'),
+ GithubRepo('personal/pdb-repo'),
+ GithubRepo('personal/privilege-check'),
+ GithubRepo('personal/simple-interpreter'),
+ GithubRepo('personal/sorting-algorithms'),
+ GithubRepo('personal/vk-scripts'),
+ GithubRepo('personal/windows-env'),
+ GithubRepo('personal/windows-home'),
+ GithubRepo('personal/windows-tmp'),
+ GithubRepo('personal/windows7-drivers'),
+ GithubRepo('personal/writable-dirs'),
+
+ BitbucketRepo('etc/etc-tensin-laptop1'),
+ BitbucketRepo('etc/etc-tensin-laptop2'),
+ BitbucketRepo('etc/etc-tensin-pc1'),
+ BitbucketRepo('etc/etc-tensin-raspi1'),
+ BitbucketRepo('etc/etc-tensin-raspi2'),
+ BitbucketRepo('fr24/fr24-cover-letter'),
+ BitbucketRepo('fr24/fr24-home'),
+ BitbucketRepo('fr24/fr24-tmp'),
+ BitbucketRepo('netwrix/etc-wiki'),
+ BitbucketRepo('netwrix/netwrix-copyright'),
+ BitbucketRepo('netwrix/netwrix-lab'),
+ BitbucketRepo('netwrix/netwrix-logs'),
+ #BitbucketRepo('netwrix/netwrix-webapi'),
+ BitbucketRepo('netwrix/netwrix-xml'),
+ BitbucketRepo('netwrix/netwrix.sh'),
+ BitbucketRepo('netwrix/wiki-backup'),
+ BitbucketRepo('shadow'),
+ BitbucketRepo('staging/361_Tensin_E_D_report'),
+ BitbucketRepo('staging/361_Tensin_E_D_slides'),
+ BitbucketRepo('staging/461_Tensin_E_D_report'),
+ BitbucketRepo('staging/461_Tensin_E_D_slides'),
+ BitbucketRepo('staging/deposit-calculator'),
+ BitbucketRepo('staging/x64-decoder'),
+
+ Repo('fr24/key_mgmt', 'ssh://egor@tensin-raspi2/~/tmp/key_mgmt.git'),
+ Repo('fr24/openfortivpn', 'ssh://egor@tensin-raspi2/~/tmp/openfortivpn.git'),
+)
+
+
+def main():
+ set_up_logging()
+ try:
+ global REPOS_DIR
+ REPOS_DIR = make_dir(REPOS_DIR)
+ global CGIT_CLONE_IP
+ CGIT_CLONE_IP = socket.gethostbyname(CGIT_CLONE_HOST)
+ success = True
+ for repo in repos:
+ if not repo.pull():
+ 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())