neon/scripts/git-upload

#!/usr/bin/env python3

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

import argparse
import os
import shutil
import subprocess
import sys
import textwrap
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

            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)

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