diff --git a/appslib/apps_cache.py b/appslib/apps_cache.py new file mode 100644 index 0000000..b8fd1e4 --- /dev/null +++ b/appslib/apps_cache.py @@ -0,0 +1,68 @@ +#!/usr/bin/env python3 + +import logging +from pathlib import Path + +import utils +from git import Repo + + +def apps_cache_path() -> Path: + path = apps_repo_root() / ".apps_cache" + path.mkdir() + return path + + +def app_cache_path(app: str) -> Path: + path = apps_cache_path() / app + path.mkdir() + return path + + +# def refresh_all_caches(catalog: dict[str, dict[str, str]]): +# for app, infos +# pass + + +def app_cache_clone(app: str, infos: dict[str, str]) -> None: + git_depths = { + "notworking": 5, + "inprogress": 20, + "default": 40, + } + + Repo.clone_from( + infos["url"], + to_path=app_cache_path(app), + depth=git_depths.get(infos["state"], git_depths["default"]), + single_branch=True, branch=infos.get("branch", "master"), + ) + + +def app_cache_update(app: str, infos: dict[str, str]) -> None: + app_path = app_cache_path(app) + age = utils.git_repo_age(app_path) + if age is False: + return app_cache_clone(app, infos) + + if age < 3600: + logging.info(f"Skipping {app}, it's been updated recently.") + return + + repo = Repo(app_path) + repo.remote("origin").set_url(infos["url"]) + + branch = infos.get("branch", "master") + if repo.active_branch != branch: + all_branches = [str(b) for b in repo.branches] + if branch in all_branches: + repo.git.checkout(branch, "--force") + else: + repo.git.remote("set-branches", "--add", "origin", branch) + repo.remote("origin").fetch(f"{branch}:{branch}") + + repo.remote("origin").fetch(refspec=branch, force=True) + repo.git.reset("--hard", f"origin/{branch}") + + +def cache_all_apps(catalog: dict[str, dict[str, str]]) -> None: diff --git a/appslib/utils.py b/appslib/utils.py new file mode 100644 index 0000000..f80650e --- /dev/null +++ b/appslib/utils.py @@ -0,0 +1,72 @@ +#!/usr/bin/env python3 + +import sys +import subprocess +from typing import Any, TextIO, Generator +import time +from functools import cache +from pathlib import Path +from git import Repo + +import toml + +REPO_APPS_ROOT = Path(Repo(__file__, search_parent_directories=True).working_dir) + + +@cache +def apps_repo_root() -> Path: + return Path(__file__).parent.parent.parent + + +def git(cmd: list[str], cwd: Path | None = None) -> str: + full_cmd = ["git"] + if cwd: + full_cmd.extend(["-C", str(cwd)]) + full_cmd.extend(cmd) + return subprocess.check_output( + full_cmd, + # env=my_env, + ).strip().decode("utf-8") + + +def git_repo_age(path: Path) -> bool | int: + for file in [path / ".git" / "FETCH_HEAD", path / ".git" / "HEAD"]: + if file.exists(): + return int(time.time() - file.stat().st_mtime) + return False + + +# Progress bar helper, stolen from https://stackoverflow.com/a/34482761 +def progressbar( + it: list[Any], + prefix: str = "", + size: int = 60, + file: TextIO = sys.stdout) -> Generator[Any, None, None]: + count = len(it) + + def show(j, name=""): + name += " " + x = int(size * j / count) + file.write( + "%s[%s%s] %i/%i %s\r" % (prefix, "#" * x, "." * (size - x), j, count, name) + ) + file.flush() + + show(0) + for i, item in enumerate(it): + yield item + show(i + 1, item[0]) + file.write("\n") + file.flush() + + +@cache +def get_catalog(working_only=False): + """Load the app catalog and filter out the non-working ones""" + catalog = toml.load((REPO_APPS_ROOT / "apps.toml").open("r", encoding="utf-8")) + if working_only: + catalog = { + app: infos for app, infos in catalog.items() + if infos.get("state") != "notworking" + } + return catalog