diff options
| -rw-r--r-- | src/util.py | 25 | ||||
| -rw-r--r-- | src/verbs/sync.py | 70 | ||||
| -rw-r--r-- | src/xi.py | 34 |
3 files changed, 103 insertions, 26 deletions
diff --git a/src/util.py b/src/util.py index 647f211..5ce93a9 100644 --- a/src/util.py +++ b/src/util.py @@ -1,4 +1,5 @@ import shutil +import csv import requests import colors import time @@ -129,7 +130,23 @@ def ask_confirmation(text, default=True, no_confirm=False): return reponse.lower() == "y" or len(reponse) == 0 -if __name__ == "__main__": - for i in range(1000): - loading_bar(i, 1000, "it is loading...") - time.sleep(0.01) +def get_distro(): + + RELEASE_DATA = {} + + with open("/etc/os-release") as f: + reader = csv.reader(f, delimiter="=") + for row in reader: + if row: + RELEASE_DATA[row[0]] = row[1] + + if RELEASE_DATA["ID"] in ["debian", "raspbian"]: + with open("/etc/debian_version") as f: + DEBIAN_VERSION = f.readline().strip() + major_version = DEBIAN_VERSION.split(".")[0] + version_split = RELEASE_DATA["VERSION"].split(" ", maxsplit=1) + if version_split[0] == major_version: + # Just major version shown, replace it with the full version + RELEASE_DATA["VERSION"] = " ".join([DEBIAN_VERSION] + version_split[1:]) + + return RELEASE_DATA diff --git a/src/verbs/sync.py b/src/verbs/sync.py index c1862e2..b0210af 100644 --- a/src/verbs/sync.py +++ b/src/verbs/sync.py @@ -8,6 +8,14 @@ import sys CACHE_DIR = "/var/cache/xipkg" def run_post_install(config, verbose=False, root="/"): + """ Scan and run postinstall scripts + + Args: + config: (dict) The xipkg config + verbose: (bool, optional) Whether to print debug messages + root: (str) The system root to begin searching + """ + if root == "/": installed_dir = util.add_path(root, config["dir"]["postinstall"]) if os.path.exists(installed_dir): @@ -25,41 +33,65 @@ def run_post_install(config, verbose=False, root="/"): util.loading_bar(len(files), len(files), f"Run Postinstalls") print(colors.RESET) -# returns a dictionary, and duration: -# key: package name -# value: list of info [checksum, size] + def list_packages(url): + """ List all of the packages available in a repo + + Will return a parsed version of /packages.list and the time it took to retrieve this + + Args: + url: (str) The repository's URL + + Returns: + dict: + A dictionary listing all packages and a string of their info summary + example: { + "linux" : "abcdef 100 200" + } + int: + The time in milliseconds for the request to complete + """ + start = time.time() status, response = util.curl(url + "/packages.list") duration = (time.time() - start) * 1000 + if status != 200: return {}, -1 - else: - if len(response) > 0: - duration /= len(response) - else: - duration = float('inf') - return { - line.split()[0].split(".")[0]: " ".join(line.split()[1:]) - for line in response.split("\n") if len(line.split()) > 0 - }, duration + + return { + line.split()[0].split(".")[0]: " ".join(line.split()[1:]) + for line in response.split("\n") if len(line.split()) > 0 + }, (duration / len(response)) if len(response) > 0 else float('inf') def sync_packages(repo, sources, verbose=False): - versions = {} + """ + Get a list of the versions available for all packages in a repo + + Args: + repo: (str) The name of the repo to search + sources: (dict) a dictionary of the sources and their urls + verbose: (bool, optional) Whether to print debug messages + Returns: + dict: + Versions of each available package + """ + versions = {} speeds = {} + for source,url in sources.items(): listed, speed = list_packages(url + repo if url[-1] == "/" else f"/{repo}") - if speed > 0: - speeds[source] = speed + if speed > 0: speeds[source] = speed if verbose: - if len(listed) == 0: - print(colors.RED + f"No packages found in {source}/{repo}" + colors.RESET) - else: - print(colors.BLACK + f"{len(listed)} packages found in {source}/{repo}" + colors.RESET) + print( + (colors.RED + f"No packages found in {source}/{repo}" + colors.RESET) + if len(listed) == 0 else + (colors.BLACK + f"{len(listed)} packages found in {source}/{repo}" + colors.RESET) + ) for p in listed: if not p in versions: versions[p] = [] @@ -1,13 +1,14 @@ import options import config import util +import os import colors from verbs.sync import sync from verbs.file import file from verbs.files import files from verbs.search import search -from verbs.info import info +from verbs.info import info, get_installed_info from verbs.remove import remove from verbs.install import install from verbs.update import update @@ -26,6 +27,34 @@ verbs = { v: globals()[v] for v in [ ] } +def print_stats(conf, opts): + pkg_count = {} + installed_dir = util.add_path(opts["r"], conf["dir"]["installed"]) + + for package in os.listdir(installed_dir): + installed_info = get_installed_info(package, conf, opts) + repo = installed_info["REPO"] + + if repo not in pkg_count: pkg_count[repo] = 0 + pkg_count[repo] += 1 + + key_count = len(os.listdir(util.add_path(opts["r"], conf["dir"]["keychain"]))) + + total = sum(pkg_count.values()) + + distro = util.get_distro()["NAME"] + + print(colors.LIGHT_CYAN + "xipkg", end="") + print(colors.CYAN + " on ", end="") + print(colors.LIGHT_CYAN + distro) + + print(colors.BLUE + f"Total key count: {colors.LIGHT_BLUE}{key_count}") + print(colors.BLUE + f"Total package count: {colors.LIGHT_BLUE}{total}") + + for repo,count in pkg_count.items(): + print(f"\t{colors.BLUE}{repo}: {colors.LIGHT_BLUE}{count}") + print(colors.CYAN + "================") + def main(): opts = options.parse_args() args = opts["args"] @@ -33,7 +62,6 @@ def main(): if opts["h"]: options.print_usage() return - conf = config.parse_file(opts["c"]) if len(args) > 0: @@ -48,7 +76,7 @@ def main(): except KeyboardInterrupt: print(colors.RESET + colors.CLEAR_LINE + colors.RED + "Action cancelled by user") else: - options.print_usage() + print_stats(conf, opts) return print(colors.RESET + colors.CLEAR_LINE, end="") |