neon/scripts/git-upload

#!/usr/bin/env python3

from contextlib import contextmanager
from tempfile import TemporaryDirectory
from pathlib import Path

import argparse
import os
import shutil
import subprocess
import sys


def absolute_path(path):
    return Path(path).resolve()


def relative_path(path):
    path = Path(path)
    if path.is_absolute():
        raise Exception(f'path `{path}` must be relative!')
    return path


@contextmanager
def chdir(cwd: Path):
    old = os.getcwd()
    os.chdir(cwd)
    try:
        yield cwd
    finally:
        os.chdir(old)


def run(cmd, *args, **kwargs):
    print('$', ' '.join(cmd))
    subprocess.check_call(cmd, *args, **kwargs)


class GitRepo:
    def __init__(self, url):
        self.url = url
        self.cwd = TemporaryDirectory()

        subprocess.check_call([
            'git',
            'clone',
            str(url),
            self.cwd.name,
        ])

    def is_dirty(self):
        res = subprocess.check_output(['git', 'status', '--porcelain'], text=True).strip()
        return bool(res)

    def update(self, message, action, branch=None):
        with chdir(self.cwd.name):
            if not branch:
                cmd = ['git', 'branch', '--show-current']
                branch = subprocess.check_output(cmd, text=True).strip()

            # Run action in repo's directory
            action()

            run(['git', 'add', '.'])

            if not self.is_dirty():
                print('No changes detected, quitting')
                return

            run([
                'git',
                '-c',
                'user.name=vipvap',
                '-c',
                'user.email=vipvap@zenith.tech',
                'commit',
                '--author="vipvap <vipvap@zenith.tech>"',
                f'--message={message}',
            ])

            for _ in range(5):
                try:
                    run(['git', 'fetch', 'origin', branch])
                    run(['git', 'rebase', f'origin/{branch}'])
                    run(['git', 'push', 'origin', branch])
                    return

                except subprocess.CalledProcessError as e:
                    print(f'failed to update branch `{branch}`: {e}', file=sys.stderr)

            raise Exception(f'failed to update branch `{branch}`')


def do_copy(args):
    src = args.src
    dst = args.dst

    if args.forbid_overwrite and dst.exists():
        raise FileExistsError(f"File exists: '{dst}'")

    if src.is_dir():
        shutil.rmtree(dst, ignore_errors=True)
        shutil.copytree(src, dst)
    else:
        shutil.copy(src, dst)


def main():
    parser = argparse.ArgumentParser(description='Git upload tool')
    parser.add_argument('--repo', type=str, metavar='URL', required=True, help='git repo url')
    parser.add_argument('--message', type=str, metavar='TEXT', help='commit message')

    commands = parser.add_subparsers(title='commands', dest='subparser_name')

    p_copy = commands.add_parser('copy', help='copy file into the repo')
    p_copy.add_argument('src', type=absolute_path, help='source path')
    p_copy.add_argument('dst', type=relative_path, help='relative dest path')
    p_copy.add_argument('--forbid-overwrite', action='store_true', help='do not allow overwrites')

    args = parser.parse_args()

    commands = {
        'copy': do_copy,
    }

    action = commands.get(args.subparser_name)
    if action:
        message = args.message or 'update'
        GitRepo(args.repo).update(message, lambda: action(args))
    else:
        parser.print_usage()


if __name__ == '__main__':
    main()