⚒️ Migration scripts about i_want_to_help HOT 19 CLOSED

#!/usr/bin/env python3
import json
import re
from http.client import TOO_MANY_REQUESTS
from pathlib import Path
from time import sleep, time
from typing import List, Literal, TypeVar, Union

import click
import requests
from rich import print
from rich import print_json as _print_json
from rich import traceback

traceback.install()

T = TypeVar("T")


def inspect(any: T) -> T:
    print(any)
    return any


def inspect_json(any: T) -> T:
    _print_json(any)
    return any


def jprint(obj):
    # create a formatted string of the Python JSON object
    print(json.dumps(obj, sort_keys=True, indent=4))


def fetch(
    url: str,
    *path: Union[str, int],
    method: Literal["GET", "POST"] = "GET",
    pretty: bool = True,
    backoff: int = 0,
):
    print(f"===> fetching {method} {url}")
    resp = requests.request(method, url)
    if resp.status_code == TOO_MANY_REQUESTS:
        sleep_until = int(resp.headers["X-RateLimit-Reset"])
        sleep_for = int(sleep_until - time()) + 1
        print(f"backing off until {sleep_until} (i.e. {sleep_for} seconds)")
        sleep(sleep_for)
        return fetch(url, *path, method=method, pretty=pretty, backoff=backoff + 1)

    resp.raise_for_status()
    json = resp.json()
    if pretty:
        inspect(json)

    for p in path:
        json = json[p]

    return json


def fetch_repo(repo: str, no_wait: bool = False):
    # Utility to pretty-print json.
    destination = Path(f'~/Downloads/{repo.replace("/", "---")}.git.tar').expanduser()
    # if destination.exists():
    #     print(f"skipping {repo}, already fetched")
    #     return

    assert re.match(r"\w+/\w+", repo), "repo must be of format ORG/NAME"
    visits_url = f"https://archive.softwareheritage.org/api/1/origin/https://github.com/{repo}/visits/"

    snapshot_url = fetch(visits_url, 0, "snapshot_url")
    head_url = fetch(snapshot_url, "branches", "HEAD", "target_url")
    directory_id = fetch(head_url, "id")

    # target_url = fetch(directory_url, 0, "target_url")
    vault_url = (
        f"https://archive.softwareheritage.org/api/1/vault/git-bare/swh:1:rev:{directory_id}/"
    )
    
    meal_json = fetch(vault_url, method="POST")

    if no_wait:
        print("not waiting")
        return meal_json

    while meal_json["status"] != "done":
        meal_json = fetch(vault_url, pretty=False)
        print(meal_json)
        sleep(60)
    else:
        print("downloading!")
        resp = requests.get(meal_json["fetch_url"])
        print(f"saving at {destination}")
        print(resp.headers)
        destination.write_bytes(resp.content)


@click.command()
@click.argument("repos", nargs=-1)
@click.option("--no-wait/--wait")
def main(repos: List[str], no_wait: bool):
    for repo in repos:
        try:
            fetch_repo(repo, no_wait=no_wait)
        except Exception as e:
            print(e)


if __name__ == "__main__":
    main()```

assert re.match(r".+/.+", repo), "repo must be of format ORG/NAME"

    head_url = fetch(snapshot_url, "branches", "HEAD", "target_url")

#!/usr/bin/env python3
import json
import re
from http.client import TOO_MANY_REQUESTS
from os.path import basename
from pathlib import Path
from time import sleep, time
from typing import List, Literal, TypeVar, Union

import click
import requests
from rich import print
from rich import print_json as _print_json
from rich import traceback

traceback.install()

T = TypeVar("T")


def inspect(any: T) -> T:
    print(any)
    return any


def inspect_json(any: T) -> T:
    _print_json(any)
    return any


def jprint(obj):
    # create a formatted string of the Python JSON object
    print(json.dumps(obj, sort_keys=True, indent=4))


def fetch(
    url: str,
    *path: Union[str, int],
    method: Literal["GET", "POST"] = "GET",
    pretty: bool = True,
    backoff: int = 0,
):
    print(f"===> fetching {method} {url}")
    resp = requests.request(method, url)
    if resp.status_code == TOO_MANY_REQUESTS:
        sleep_until = int(resp.headers["X-RateLimit-Reset"])
        sleep_for = int(sleep_until - time()) + 1
        print(f"backing off until {sleep_until} (i.e. {sleep_for} seconds)")
        sleep(sleep_for)
        return fetch(url, *path, method=method, pretty=pretty, backoff=backoff + 1)

    resp.raise_for_status()
    json = resp.json()
    if pretty:
        inspect(json)

    for p in path:
        json = json[p]

    return json


def fetch_repo(repo: str, no_wait: bool = False):
    # Utility to pretty-print json.

    assert re.match(r".+/.+", repo), "repo must be of format ORG/NAME"
    visits_url = f"https://archive.softwareheritage.org/api/1/origin/https://github.com/{repo}/visits/"

    snapshot_url = fetch(visits_url, 0, "snapshot_url")
    branches = fetch(snapshot_url, "branches")
    for branch, branch_data in branches.items():
        print(f"Processing branch {branch}")
        if branch_data.get("target_type") == "alias":
            print(f"SKIP alias branch {branch}")
            continue
        branch_url = fetch(snapshot_url, "branches", branch, "target_url")
        directory_id = fetch(branch_url, "id")

        # target_url = fetch(directory_url, 0, "target_url")
        vault_url = f"https://archive.softwareheritage.org/api/1/vault/git-bare/swh:1:rev:{directory_id}/"

        destination = Path(
            f'~/Downloads/{repo.replace("/", "---")}---{basename(branch)}.git.tar'
        ).expanduser()
        if destination.exists():
            print(f"skipping {repo}, already fetched")
            return

        meal_json = fetch(vault_url, method="POST")

        if no_wait:
            print("not waiting")
            return meal_json

        while meal_json["status"] != "done":
            meal_json = fetch(vault_url, pretty=False)
            print(meal_json)
            print("Sleeping for 30s")
            sleep(30)
        else:
            print("downloading!")
            resp = requests.get(meal_json["fetch_url"])
            print(f"saving at {destination}")
            print(resp.headers)
            destination.write_bytes(resp.content)


@click.command()
@click.argument("repos", nargs=-1)
@click.option("--no-wait/--wait")
def main(repos: List[str], no_wait: bool):
    for repo in repos:
        try:
            fetch_repo(repo, no_wait=no_wait)
        except Exception as e:
            print(e)


if __name__ == "__main__":
    main()