list_builder: start using pathlib, typing

list_builder.py

@@ -5,17 +5,22 @@ import sys
 import os
 import re
 import json
+from shutil import which
 import toml
 import subprocess
 import time
+from typing import TextIO, Generator, Any
+from pathlib import Path
 
 from collections import OrderedDict
 from tools.packaging_v2.convert_v1_manifest_to_v2_for_catalog import convert_v1_manifest_to_v2_for_catalog
 
 now = time.time()
 
+REPO_APPS_PATH = Path(__file__).parent
+
 # Load categories and reformat the structure to have a list with an "id" key
-categories = toml.load(open("categories.toml"))
+categories = toml.load((REPO_APPS_PATH / "categories.toml").open("r", encoding="utf-8"))
 for category_id, infos in categories.items():
     infos["id"] = category_id
     for subtag_id, subtag_infos in infos.get("subtags", {}).items():
@@ -25,13 +30,13 @@ for category_id, infos in categories.items():
 categories = list(categories.values())
 
 # (Same for antifeatures)
-antifeatures = toml.load(open("antifeatures.toml"))
+antifeatures = toml.load((REPO_APPS_PATH / "antifeatures.toml").open("r", encoding="utf-8"))
 for antifeature_id, infos in antifeatures.items():
     infos["id"] = antifeature_id
 antifeatures = list(antifeatures.values())
 
 # Load the app catalog and filter out the non-working ones
-catalog = toml.load(open("apps.toml"))
+catalog = toml.load((REPO_APPS_PATH / "apps.toml").open("r", encoding="utf-8"))
 catalog = {
     app: infos for app, infos in catalog.items() if infos.get("state") != "notworking"
 }
@@ -39,19 +44,20 @@ catalog = {
 my_env = os.environ.copy()
 my_env["GIT_TERMINAL_PROMPT"] = "0"
 
-os.makedirs(".apps_cache", exist_ok=True)
-os.makedirs("builds/", exist_ok=True)
+(REPO_APPS_PATH / ".apps_cache").mkdir(exist_ok=True)
+(REPO_APPS_PATH / "builds").mkdir(exist_ok=True)
 
 
-def error(msg):
+def error(msg: str) -> None:
     msg = "[Applist builder error] " + msg
-    if os.path.exists("/usr/bin/sendxmpppy"):
+    if which("sendxmpppy") is not None:
         subprocess.call(["sendxmpppy", msg], stdout=open(os.devnull, "wb"))
     print(msg + "\n")
 
 
 # Progress bar helper, stolen from https://stackoverflow.com/a/34482761
-def progressbar(it, prefix="", size=60, file=sys.stdout):
+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=""):
@@ -75,23 +81,22 @@ def progressbar(it, prefix="", size=60, file=sys.stdout):
 ###################################
 
 
-def app_cache_folder(app):
-    return os.path.join(".apps_cache", app)
+def app_cache_folder(app: str) -> Path:
+    return REPO_APPS_PATH / ".apps_cache" / app
 
 
-def git(cmd, in_folder=None):
+def git(cmd: str, in_folder: Path | None = None):
 
     if in_folder:
-        cmd = "-C " + in_folder + " " + cmd
+        cmd = "-C " + str(in_folder) + " " + cmd
     cmd = "git " + cmd
     return subprocess.check_output(cmd.split(), env=my_env).strip().decode("utf-8")
 
 
-def refresh_all_caches():
-
+def refresh_all_caches() -> None:
     for app, infos in progressbar(sorted(catalog.items()), "Updating git clones: ", 40):
         app = app.lower()
-        if not os.path.exists(app_cache_folder(app)):
+        if not app_cache_folder(app).exists():
             try:
                 init_cache(app, infos)
             except Exception as e:
@@ -103,7 +108,7 @@ def refresh_all_caches():
                 error("Failed to not refresh cache for %s" % app)
 
 
-def init_cache(app, infos):
+def init_cache(app: str, infos: dict[str, str]) -> None:
 
     if infos["state"] == "notworking":
         depth = 5
@@ -122,11 +127,11 @@ def init_cache(app, infos):
     )
 
 
-def refresh_cache(app, infos):
+def refresh_cache(app: str, infos: dict[str, str]) -> None:
 
     # Don't refresh if already refreshed during last hour
-    fetch_head = app_cache_folder(app) + "/.git/FETCH_HEAD"
-    if os.path.exists(fetch_head) and now - os.path.getmtime(fetch_head) < 3600:
+    fetch_head = app_cache_folder(app) / ".git" / "FETCH_HEAD"
+    if fetch_head.exists() and (now - fetch_head.stat().st_mtime) < 3600:
         return
 
     branch = infos.get("branch", "master")
@@ -136,7 +141,7 @@ def refresh_cache(app, infos):
         # With git >= 2.22
         # current_branch = git("branch --show-current", in_folder=app_cache_folder(app))
         current_branch = git(
-            "rev-parse --abbrev-ref HEAD", in_folder=app_cache_folder(app)
+            "rev-parse --abbrev-ref HEAD", app_cache_folder(app)
         )
         if current_branch != branch:
             # With git >= 2.13
@@ -160,10 +165,7 @@ def refresh_cache(app, infos):
     except:
         # Sometimes there are tmp issue such that the refresh cache ..
         # we don't trigger an error unless the cache hasnt been updated since more than 24 hours
-        if (
-            os.path.exists(fetch_head)
-            and now - os.path.getmtime(fetch_head) < 24 * 3600
-        ):
+        if (fetch_head.exists() and (now - fetch_head.stat().st_mtime) < 24 * 3600):
             pass
         else:
             raise
@@ -229,7 +231,7 @@ def build_catalog():
 
     for appid, app in result_dict_with_manifest_v2.items():
         appid = appid.lower()
-        if os.path.exists(f"logos/{appid}.png"):
+        if (REPO_APPS_PATH / "logos" / f"{appid}.png").exists():
             logo_hash = subprocess.check_output(["sha256sum", f"logos/{appid}.png"]).strip().decode("utf-8").split()[0]
             os.system(f"cp logos/{appid}.png builds/default/v3/logos/{logo_hash}.png")
             # FIXME: implement something to cleanup old logo stuf in the builds/.../logos/ folder somehow
@@ -291,7 +293,7 @@ def build_app_dict(app, infos):
 
     # Make sure we have some cache
     this_app_cache = app_cache_folder(app)
-    assert os.path.exists(this_app_cache), "No cache yet for %s" % app
+    assert this_app_cache.exists(), "No cache yet for %s" % app
 
     commit_timestamps_for_this_app_in_catalog = git(f'log -G "{app}"|\[{app}\] --first-parent --reverse --date=unix --format=%cd -- apps.json apps.toml')
     # Assume the first entry we get (= the oldest) is the time the app was added
@@ -334,10 +336,10 @@ def build_app_dict(app, infos):
     timestamp = int(timestamp)
 
     # Build the dict with all the infos
-    if os.path.exists(this_app_cache + "/manifest.toml"):
-        manifest = toml.load(open(this_app_cache + "/manifest.toml"), _dict=OrderedDict)
+    if (this_app_cache / "manifest.toml").exists():
+        manifest = toml.load((this_app_cache / "manifest.toml").open("r"), _dict=OrderedDict)
     else:
-        manifest = json.load(open(this_app_cache + "/manifest.json"))
+        manifest = json.load((this_app_cache / "manifest.json").open("r"))
 
     return {
         "id": manifest["id"],