Create a package.zip from the files modified within a Github repo

package.py

import asyncio
import logging
import os
import shutil
import subprocess

logger = logging.getLogger(__name__)
logger.addHandler(hdlr=logging.StreamHandler())
logger.setLevel(level=logging.DEBUG)

PACKAGE = "package"


async def cleanup():
    """Cleanup left over repos and create a new one."""
    if os.path.isdir(PACKAGE):
        shutil.rmtree(PACKAGE)
    os.makedirs(PACKAGE)


async def get_modified():
    """Get modified files in current Git repo."""
    try:
        branch = subprocess.check_output("git rev-parse --abbrev-ref HEAD", shell=True)
    except subprocess.CalledProcessError as error:
        result = error.output.decode(encoding='UTF-8').strip()
        logging.error(f"[{error.returncode}]: {result}\n")
        return
    else:
        branch = branch.decode().strip()
    try:
        base_commit = subprocess.check_output("git merge-base %s HEAD" % branch, shell=True)
    except subprocess.CalledProcessError as error:
        result = error.output.decode(encoding='UTF-8').strip()
        logging.error(f"[{error.returncode}]: {result}\n")
        return
    else:
        base_commit = base_commit.decode().strip()
    try:
        # Ref: https://stackoverflow.com/a/6879568
        modified = subprocess.check_output("git diff --diff-filter=ACMR --name-only %s" % base_commit,
                                           shell=True)
    except subprocess.CalledProcessError as error:
        result = error.output.decode(encoding='UTF-8').strip()
        logging.error(f"[{error.returncode}]: {result}\n")
        return
    else:
        return modified.decode().splitlines()


async def create_package():
    """Create package.zip."""
    await cleanup()
    modified = await get_modified()
    for file in modified:
        dst_directory = os.path.dirname(os.path.join(os.getcwd(), file)).replace(os.getcwd(),
                                                                                 os.path.join(os.getcwd(), PACKAGE))
        if not os.path.isdir(dst_directory):
            os.makedirs(dst_directory)
        shutil.copyfile(src=os.path.join(os.getcwd(), file),
                        dst=os.path.join(os.getcwd(), PACKAGE, file))
    logger.info("Zipping %d files...", len(modified))
    shutil.make_archive(base_name=PACKAGE, format="zip", root_dir=PACKAGE)
    shutil.rmtree(PACKAGE)


async def unpack(filename: str = f'{PACKAGE}.zip'):
    """Unpacks package.zip file and moves the files into respective folder structure."""
    if not os.path.isfile(filename):
        raise FileNotFoundError(f'{filename!r} is missing')

    temp_dir = 'temp_unpack'

    if os.path.isdir(temp_dir):
        shutil.rmtree(temp_dir)
    os.mkdir(temp_dir)

    shutil.unpack_archive(filename=filename, extract_dir='temp_unpack')
    logger.info(f'Unzipping {filename!r}')
    files = 0
    for __path, __directory, __files in os.walk('temp_unpack'):
        for __file in __files:
            source = os.path.join(__path, __file)
            destination = os.path.join(os.getcwd(), __path, __file).replace(temp_dir + os.path.sep, '')
            logger.info(f'Moving {source!r} to {destination!r}')
            shutil.move(src=source, dst=destination)
            files += 1
    logger.info(f'Moved {files} files')
    shutil.rmtree('temp_unpack')


async def download_files_from_commit(sha, filename: str = f'{PACKAGE}.zip'):
    """Downloads the files added/copied/modified/renamed in a particular commit."""
    # https://stackoverflow.com/a/48503019
    cmd = "git archive --format=zip --output={filename} {sha} " \
          "`git diff-tree --diff-filter=ACMR --no-commit-id --name-only -r {sha}`"
    cmd = cmd.format(filename=filename, sha=sha)
    subprocess.check_output(cmd, shell=True)


if __name__ == '__main__':
    asyncio.run(create_package())