neon/scripts/git-upload

#!/usr/bin/env python3

import argparse
import os
import shlex
import shutil
import subprocess
import sys
import textwrap
from contextlib import contextmanager
from distutils.dir_util import copy_tree
from pathlib import Path
from tempfile import TemporaryDirectory
from typing import Optional


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, branch: Optional[str] = None):
        self.url = url
        self.cwd = TemporaryDirectory()
        self.branch = branch

        args = [
            'git',
            'clone',
            '--single-branch',
        ]
        if self.branch:
            args.extend(['--branch', self.branch])

        subprocess.check_call([
            *args,
            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

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

            for _ in range(5):
                try:
                    run(['git', 'fetch', 'origin', branch])
                    run(git_with_user + ['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():
        if not args.merge:
            shutil.rmtree(dst, ignore_errors=True)
        # distutils is deprecated, but this is a temporary workaround before python version bump
        # here we need dir_exists_ok=True from shutil.copytree which is available in python 3.8+
        copy_tree(str(src), str(dst))
    else:
        shutil.copy(src, dst)

    if args.run_cmd:
        run(shlex.split(args.run_cmd))


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')
    parser.add_argument('--branch', type=str, metavar='TEXT', help='target git repo branch')

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

    p_copy = commands.add_parser(
        'copy',
        help='copy file into the repo',
        formatter_class=argparse.RawTextHelpFormatter,
    )
    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')
    p_copy.add_argument(
        '--merge',
        action='store_true',
        help='when copying a directory do not delete existing data, but add new files')
    p_copy.add_argument('--run-cmd',
                        help=textwrap.dedent('''\
                run arbitrary cmd on top of copied files,
                example usage is static content generation
                based on current repository state\
            '''))

    args = parser.parse_args()

    commands = {
        'copy': do_copy,
    }

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


if __name__ == '__main__':
    main()