add appslib

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:

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