github-archiver/github_release_archiver.py

import os, sys, time
import requests
import json

import functools
import pathlib
import shutil
from tqdm.auto import tqdm
from urllib.parse import urlparse

try:
    import github_api_token
    if "ghb" not in github_api_token.personal_access_token:
        raise ImportError
except ImportError:
    print("Please create a github_api_token.py file with your personal access token first in github_api_token.py.")
    exit(1)

time_delay = 2

session = requests.Session()
session.headers.update({
    "Accept": "application/vnd.github+json",
    "Authorization": f'Bearer {github_api_token.personal_access_token}',
    "X-GitHub-Api-Version": "2022-11-28"
})

datas = {
    "per_page": 100,
    "page": 1
}

release_url = "https://api.github.com/repos"

def get_repo_params(url):
    try:
        if "github.com" not in url:
            print("This isn't a GitHub URL...")
            return None, None
        parsed_url = urlparse(url)
        path_parts = parsed_url.path.strip('/').split('/')
        if len(path_parts) >= 2:
            user, repository = path_parts[0], path_parts[1]
            return user, repository
        else:
            print("Invalid GitHub URL")
            return None, None
    except Exception as e:
        print(f"Error: {e}\nLink provided: {url}")
        return None, None

def download(url, filename):
    r = session.get(url, stream=True, allow_redirects=True)
    if r.status_code != 200:
        print(f"Request to {url} returned status code {r.status_code}")
        return 0
        # r.raise_for_status()  # Will only raise for 4xx codes, so...
        # raise RuntimeError(f"Request to {url} returned status code {r.status_code}")
    file_size = int(r.headers.get('Content-Length', 0))

    path = pathlib.Path(filename).expanduser().resolve()
    path.parent.mkdir(parents=True, exist_ok=True)

    desc = "(Unknown total file size)" if file_size == 0 else ""
    r.raw.read = functools.partial(r.raw.read, decode_content=True)  # Decompress if needed
    with tqdm.wrapattr(r.raw, "read", total=file_size, desc=desc) as r_raw:
        with path.open("wb") as f:
            shutil.copyfileobj(r_raw, f)

    return path

def get_github_releases(repo):
    url = f"{release_url}/{repo}/releases"
    cur_page = 1
    all_data = []

    while True:
        print(f"Getting page {cur_page} of releases")
        datas["page"] = cur_page
        response = session.get(url,params=datas)
        if response.status_code != 200:
            print(f"Request to {url} returned status code {response.status_code}")
            return 0
        
        json_resp = response.json()
        all_data += json_resp
        cur_page += 1

        print(f"Got {len(json_resp)} releases")

        if len(json_resp) < 100:
            print(f"Got all releases for {repo}. Total releases: {len(all_data)}, total pages: {cur_page-1}")
            break
        else:
            time.sleep(time_delay)

    return all_data

'''
Download all assets for a release; in this case, release is 
one element of the list returned by get_github_releases
    @param release: dict
    @return success: bool
'''
def download_release_assets(release):
    assets = release["assets"]
    if len(assets) == 0:
        print(f"No assets for release {release['name']}")
        return False
    else:
        for asset in assets:
            print(f" - Downloading asset {asset['name']}")
            # Check if file already exists
            if os.path.exists(f"{download_folder}/{release['tag_name']}/{asset['name']}"):
                print(f"   File {asset['name']} already exists, skipping")
                continue
            else:
                download(asset["browser_download_url"], f"{download_folder}/{release['tag_name']}/{asset['name']}")
            time.sleep(time_delay)

if __name__ == "__main__":
    # Ask user to provide a GitHub repository link and a download folder
    repository_link = input("Enter the GitHub repository link: ")
    repository = get_repo_params(repository_link)
    if repository[0] == None:
        print("Invalid repository")
        exit(1)

    # Check if download folder argument exists
    try:
        dir_in = sys.argv[1]
        if os.path.exists(dir_in) and os.path.isdir(dir_in):
            dir_in = os.path.abspath(dir_in) + '/'
        else:
            print("Invalid directory")
            print("Usage: python github_release_archiver.py <download_folder>\n<download_folder> optional, must be a valid directory")
            exit(1)
    except IndexError:
        download_folder = input(f"Enter the download folder ({repository[1]}-releases): ")
        if download_folder == "":
            download_folder = f"{repository[1]}-releases"
        dir_in = os.getcwd() + download_folder

    # Check if we downloaded the releases.json file
    json_filepath = os.path.join(dir_in, f"{repository[1]}.json")
    if os.path.exists(json_filepath):
        dat = json.loads(open(json_filepath).read())
    else:
        dat = get_github_releases("/".join(repository))
        open(json_filepath, "w").write(json.dumps(dat, indent=4))

    for release in dat:
        print(f"Downloading assets for release {release['tag_name']}")
        download_release_assets(release)