GitHub Release Backup Script

mirror-releases.py

# Use this with a slow cron - I had mine set to run every 24 bours and keep copies of the last 5 releases

from os import environ, listdir, makedirs
from os.path import join, exists, getmtime, isdir
from shutil import rmtree
from json import loads
from asyncio import run, create_task, sleep, create_subprocess_shell
from asyncio.subprocess import PIPE, STDOUT 
from aiohttp import ClientSession
from logging import basicConfig, INFO, DEBUG, WARNING, getLogger

basicConfig(level=INFO, format='%(asctime)s %(levelname)s %(funcName)s:%(lineno)s %(message)s')
log = getLogger(__name__)

PROJECTS = environ.get('PROJECTS', 'Ryujinx/release-channel-master').split(',')
ARCHIVE_PATH = environ.get('ARCHIVE_PATH', 'releases')
RELEASE_COUNT = int(environ.get('RELEASE_COUNT', 5))

async def fetch(path: str, url: str) -> None: 
    proc = await create_subprocess_shell(f"wget -P {path} -N {url} -qnv", stdout=PIPE, stderr=PIPE)
    log.info(f"fetching {url} to {path}")
    stdout, stderr = await proc.communicate()
    log.debug(stdout)
    log.debug(stderr)

async def main() -> None:
    async with ClientSession() as session:
        for project in PROJECTS:
            log.debug(f'Fetching releases for {project}')
            async with session.get(f'https://api.github.com/repos/{project}/releases') as response:
                data = await response.json()
                filtered_data = [element for element in data if ((element["draft"] != True) and (element["prerelease"] != True))][:RELEASE_COUNT]
                for release in filtered_data:
                    folder = join(ARCHIVE_PATH, project, release['tag_name'])
                    if not exists(folder):
                        makedirs(folder)
                    for asset in release['assets']:
                        if not exists(join(folder, asset['name'])):
                            create_task(fetch(folder, asset['browser_download_url']))
                            await sleep(0.1)
                # get modification dates for release folders and remove the oldest ones
                folders = [f for f in listdir(join(ARCHIVE_PATH, project)) if isdir(join(ARCHIVE_PATH, project, f))]
                log.info(f"checking for old releases in {project}")
                if len(folders) > RELEASE_COUNT:
                    #folders.sort(key=lambda f: (f, getmtime(join(ARCHIVE_PATH, project, f))), reverse=True)
                    folders.sort(reverse=True)
                    folders = folders[RELEASE_COUNT:]
                    for folder in folders:
                        log.warning(f"removing {project} release {folder}")
                        rmtree(join(ARCHIVE_PATH, project, folder))
                        
if __name__ == '__main__':
    run(main())