genmirrorz.py

#!/usr/bin/env python3
# -*- coding: utf-8 -*-


import json
import sys
import subprocess
import sys
import os
import logging

from utils import CONFIG_FOLDER, get_mirrorz_cname, get_resp_with_timeout
from gencontent import USER_CONFIG as gencontent_config


cname = get_mirrorz_cname()
logger = logging.getLogger(__name__)

with open(CONFIG_FOLDER / "genmirrorz.json") as f:
    options = json.load(f)
    base = options["base"]
    skiplist = options.get("skip", [])

metas = []
for yuki_endpoint in gencontent_config.get("yuki", []):
    if yuki_endpoint.get("mirrorz"):
        url = yuki_endpoint["url"]
        this_metas = get_resp_with_timeout(url)
        if not this_metas:
            logger.error(f"failed to get {url}")
            continue
        try:
            this_metas = this_metas.json()
        except json.JSONDecodeError:
            logger.error(f"failed to decode json from {url}")
            continue
        if not this_metas:
            logger.error(f"empty json from {url}")
            continue
        metas.extend(this_metas)


def name_func(name: str) -> str:
    if name in cname:
        return cname[name]
    else:
        return name


def iso(iso_orig: list) -> None:
    # modify iso_orig inplace
    for i in iso_orig:
        i["distro"] = name_func(i["distro"])
        if not i.get("category"):
            # fallback to OS (iso) if not exist.
            i["category"] = "os"


def size(bytes: int) -> str:
    mib = bytes / 1024 / 1024
    if mib < 1024:
        return f"{mib:.2f} MiB"
    gib = mib / 1024
    if gib < 1024:
        return f"{gib:.2f} GiB"
    tib = gib / 1024
    return f"{tib:.2f} TiB"


def parse_repo_with_meta(repolist: list, meta: dict) -> dict:
    content_list = []
    content_hash = {}
    for i in repolist:
        _, help_url, _, name = i
        cname = name_func(name)
        content_hash[cname.lower()] = len(content_list)
        content_list.append(
            {
                "cname": cname,
                "desc": "",  # now we don't have desc yet...
                "url": f"/{name}",
                "status": "U",
                "help": help_url,
                "upstream": "",
            }
        )
    # now we add data to content_list with meta!
    for i in meta:
        name = i["name"]
        if name in skiplist:
            continue
        name = name_func(name)
        try:
            try:
                ind = content_hash[name.lower()]
            except KeyError:
                ind = content_hash[
                    name.lower().split(".")[0]
                ]  # fix repo name like "kubernetes.apt"
            next_run = i.get("nextRun")
            last_success = i.get("lastSuccess")
            if next_run < 0:
                content_list[ind][
                    "status"
                ] = "P"  # negative next_run means cron date set to 2/30 or 2/31 (unreachable)
            elif i["syncing"]:
                content_list[ind]["status"] = "Y" + str(i.get("prevRun"))
                if last_success:
                    content_list[ind]["status"] += "O" + str(last_success)
            elif i["exitCode"] == 0:
                content_list[ind]["status"] = "S" + str(last_success)
            else:
                content_list[ind]["status"] = "F" + str(i.get("prevRun"))
                if last_success:
                    content_list[ind]["status"] += "O" + str(last_success)
            if next_run > 0:
                content_list[ind]["status"] += "X" + str(next_run)
            content_list[ind]["size"] = size(i["size"])
            content_list[ind]["upstream"] = i["upstream"]
        except KeyError:
            print(f"failed to parse {i['name']}", file=sys.stderr)
    return content_list


def disk_info(site: dict) -> None:
    # TODO(portability): This now only works on mirrors4
    repo_zfs = subprocess.check_output(
        "zfs get -Hp -o value used,available pool0/repo", shell=True
    ).decode("utf-8")
    used, available = repo_zfs.split()
    used = int(used)
    available = int(available)
    site["disk"] = f"{size(used)} / {size(used + available)}"


def getMirrorzJson(repolist, isolist):
    disk_info(base["site"])
    iso(isolist)
    mirrors = parse_repo_with_meta(repolist, metas)

    mirrorz = base
    mirrorz["info"] = isolist
    mirrorz["mirrors"] = mirrors

    return json.dumps(mirrorz)


if __name__ == "__main__":
    from utils import get_isolist
    import gencontent

    isolist = get_isolist()
    repolist = gencontent.genRepoList()

    print(getMirrorzJson(repolist, isolist))