initial toolsv2 commit

toolsv2/.pdm-python

@@ -0,0 +1 @@
+/home/felix/tmp/yunohost/repo_apps/toolsv2/.venv/bin/python

toolsv2/pyproject.toml

@@ -0,0 +1,48 @@
+[project]
+name = "tools"
+version = "0.1.0"
+description = "Default template for PDM package"
+authors = [
+    {name = "Salamandar", email = "felix@piedallu.me"},
+]
+dependencies = [
+    "gitpython>=3.1.42",
+    "tomlkit>=0.12.4",
+]
+requires-python = ">=3.8"
+readme = "README.md"
+license = {text = "GPLv3"}
+
+
+[tool.pdm]
+distribution = false
+
+[tool.pdm.dev-dependencies]
+dev = [
+    "black>=24",
+    "ruff>=0.3",
+    "mypy>=1.9",
+    "types-toml>=0.10",
+]
+
+[tool.black]
+line-length = 120
+
+[tool.ruff]
+line-length = 120
+
+[tool.ruff.lint]
+select = [
+  "F",   # pyflakes
+  "E",   # pycodestyle
+  "W",   # pycodestyle
+  "I",   # isort
+  "N",   # pep8-naming
+  "B",   # flake8-ubgbear
+  "ANN", # flake8-annotations
+  "Q",   # flake8-quotes
+  "PTH", # flake8-use-pathlib
+  "UP",  # pyupgrade,
+]
+
+ignore = ["ANN101", "ANN102", "ANN401"]

toolsv2/tools/forge/github.py

@@ -0,0 +1,47 @@
+#!/usr/bin/env python3
+
+import logging
+from functools import cache
+
+import github
+
+from ..utils.paths import APPS_REPO_ROOT
+
+
+@cache
+def github_login() -> str | None:
+    if (file := APPS_REPO_ROOT / ".github_login").exists():
+        return file.open(encoding="utf-8").read().strip()
+    return None
+
+
+@cache
+def github_email() -> str | None:
+    if (file := APPS_REPO_ROOT / ".github_email").exists():
+        return file.open(encoding="utf-8").read().strip()
+    return None
+
+
+@cache
+def github_token() -> str | None:
+    if (file := APPS_REPO_ROOT / ".github_token").exists():
+        return file.open(encoding="utf-8").read().strip()
+    return None
+
+
+@cache
+def github_auth() -> github.Auth.Auth | None:
+    token = github_token()
+    if token is None:
+        logging.warning("Could not get Github token authentication.")
+        return None
+    return github.Auth.Token(token)
+
+
+@cache
+def github_api() -> github.Github:
+    auth = github_auth()
+    if auth is None:
+        logging.warning("Returning unauthenticated Github API.")
+        return github.Github()
+    return github.Github(auth=auth)

toolsv2/tools/forge/rest_api.py

@@ -0,0 +1,178 @@
+#!/usr/bin/env python3
+
+import re
+from enum import Enum
+from typing import Any, Optional
+
+import requests
+
+
+class RefType(Enum):
+    tags = 1
+    commits = 2
+
+
+class GithubAPI:
+    def __init__(self, upstream: str, auth: Optional[tuple[str, str]] = None) -> None:
+        self.upstream = upstream
+        self.upstream_repo = upstream.replace("https://github.com/", "").strip("/")
+        assert (
+            len(self.upstream_repo.split("/")) == 2
+        ), f"'{upstream}' doesn't seem to be a github repository ?"
+        self.auth = auth
+
+    def internal_api(self, uri: str) -> Any:
+        url = f"https://api.github.com/{uri}"
+        r = requests.get(url, auth=self.auth)
+        r.raise_for_status()
+        return r.json()
+
+    def tags(self) -> list[dict[str, str]]:
+        """Get a list of tags for project."""
+        return self.internal_api(f"repos/{self.upstream_repo}/tags")
+
+    def commits(self) -> list[dict[str, Any]]:
+        """Get a list of commits for project."""
+        return self.internal_api(f"repos/{self.upstream_repo}/commits")
+
+    def releases(self) -> list[dict[str, Any]]:
+        """Get a list of releases for project."""
+        return self.internal_api(f"repos/{self.upstream_repo}/releases")
+
+    def url_for_ref(self, ref: str, ref_type: RefType) -> str:
+        """Get a URL for a ref."""
+        if ref_type == RefType.tags:
+            return f"{self.upstream}/archive/refs/tags/{ref}.tar.gz"
+        elif ref_type == RefType.commits:
+            return f"{self.upstream}/archive/{ref}.tar.gz"
+        else:
+            raise NotImplementedError
+
+
+class GitlabAPI:
+    def __init__(self, upstream: str) -> None:
+        # Find gitlab api root...
+        self.forge_root = self.get_forge_root(upstream).rstrip("/")
+        self.project_path = upstream.replace(self.forge_root, "").lstrip("/")
+        self.project_id = self.find_project_id(self.project_path)
+
+    def get_forge_root(self, project_url: str) -> str:
+        """A small heuristic based on the content of the html page..."""
+        r = requests.get(project_url)
+        r.raise_for_status()
+        match = re.search(r"const url = `(.*)/api/graphql`", r.text)
+        assert match is not None
+        return match.group(1)
+
+    def find_project_id(self, project: str) -> int:
+        try:
+            project = self.internal_api(f"projects/{project.replace('/', '%2F')}")
+        except requests.exceptions.HTTPError as err:
+            if err.response.status_code != 404:
+                raise
+            # Second chance for some buggy gitlab instances...
+            name = self.project_path.split("/")[-1]
+            projects = self.internal_api(f"projects?search={name}")
+            project = next(
+                filter(
+                    lambda x: x.get("path_with_namespace") == self.project_path,
+                    projects,
+                )
+            )
+
+        assert isinstance(project, dict)
+        project_id = project.get("id", None)
+        return project_id
+
+    def internal_api(self, uri: str) -> Any:
+        url = f"{self.forge_root}/api/v4/{uri}"
+        r = requests.get(url)
+        r.raise_for_status()
+        return r.json()
+
+    def tags(self) -> list[dict[str, str]]:
+        """Get a list of tags for project."""
+        return self.internal_api(f"projects/{self.project_id}/repository/tags")
+
+    def commits(self) -> list[dict[str, Any]]:
+        """Get a list of commits for project."""
+        return [
+            {
+                "sha": commit["id"],
+                "commit": {"author": {"date": commit["committed_date"]}},
+            }
+            for commit in self.internal_api(
+                f"projects/{self.project_id}/repository/commits"
+            )
+        ]
+
+    def releases(self) -> list[dict[str, Any]]:
+        """Get a list of releases for project."""
+        releases = self.internal_api(f"projects/{self.project_id}/releases")
+        retval = []
+        for release in releases:
+            r = {
+                "tag_name": release["tag_name"],
+                "prerelease": False,
+                "draft": False,
+                "html_url": release["_links"]["self"],
+                "assets": [
+                    {
+                        "name": asset["name"],
+                        "browser_download_url": asset["direct_asset_url"],
+                    }
+                    for asset in release["assets"]["links"]
+                ],
+            }
+            for source in release["assets"]["sources"]:
+                r["assets"].append(
+                    {
+                        "name": f"source.{source['format']}",
+                        "browser_download_url": source["url"],
+                    }
+                )
+            retval.append(r)
+
+        return retval
+
+    def url_for_ref(self, ref: str, ref_type: RefType) -> str:
+        name = self.project_path.split("/")[-1]
+        clean_ref = ref.replace("/", "-")
+        return f"{self.forge_root}/{self.project_path}/-/archive/{ref}/{name}-{clean_ref}.tar.bz2"
+
+
+class GiteaForgejoAPI:
+    def __init__(self, upstream: str):
+        # Find gitea/forgejo api root...
+        self.forge_root = self.get_forge_root(upstream).rstrip("/")
+        self.project_path = upstream.replace(self.forge_root, "").lstrip("/")
+
+    def get_forge_root(self, project_url: str) -> str:
+        """A small heuristic based on the content of the html page..."""
+        r = requests.get(project_url)
+        r.raise_for_status()
+        match = re.search(r"appUrl: '([^']*)',", r.text)
+        assert match is not None
+        return match.group(1).replace("\\", "")
+
+    def internal_api(self, uri: str):
+        url = f"{self.forge_root}/api/v1/{uri}"
+        r = requests.get(url)
+        r.raise_for_status()
+        return r.json()
+
+    def tags(self) -> list[dict[str, Any]]:
+        """Get a list of tags for project."""
+        return self.internal_api(f"repos/{self.project_path}/tags")
+
+    def commits(self) -> list[dict[str, Any]]:
+        """Get a list of commits for project."""
+        return self.internal_api(f"repos/{self.project_path}/commits")
+
+    def releases(self) -> list[dict[str, Any]]:
+        """Get a list of releases for project."""
+        return self.internal_api(f"repos/{self.project_path}/releases")
+
+    def url_for_ref(self, ref: str, ref_type: RefType) -> str:
+        """Get a URL for a ref."""
+        return f"{self.forge_root}/{self.project_path}/archive/{ref}.tar.gz"

toolsv2/tools/utils/logging_sender.py

@@ -0,0 +1,39 @@
+#!/usr/bin/env python3
+
+import subprocess
+from shutil import which
+import logging
+import logging.handlers
+
+
+def send_to_matrix(message: str) -> None:
+    if which("sendxmpppy") is None:
+        logging.warning("Could not send error via xmpp.")
+        return
+    subprocess.call(["sendxmpppy", message], stdout=subprocess.DEVNULL)
+
+
+class LogSenderHandler(logging.Handler):
+    def __init__(self):
+        logging.Handler.__init__(self)
+        self.is_logging = False
+
+    def emit(self, record):
+        msg = f"[Apps tools error] {record.msg}"
+        send_to_matrix(msg)
+
+    @classmethod
+    def add(cls, level=logging.ERROR):
+        if not logging.getLogger().handlers:
+            logging.basicConfig()
+
+        # create handler
+        handler = cls()
+        handler.setLevel(level)
+        # add the handler
+        logging.getLogger().handlers.append(handler)
+
+
+def enable():
+    """Enables the LogSenderHandler"""
+    LogSenderHandler.add(logging.ERROR)

toolsv2/tools/utils/paths.py

@@ -0,0 +1,11 @@
+#!/usr/bin/env python3
+
+from pathlib import Path
+
+from git import Repo
+
+APPS_REPO_ROOT = Path(Repo(__file__, search_parent_directories=True).working_dir)
+
+
+def apps_repo_root() -> Path:
+    return APPS_REPO_ROOT

toolsv2/tools/utils/utils.py

@@ -0,0 +1,54 @@
+#!/usr/bin/env python3
+
+import time
+from functools import cache
+from pathlib import Path
+from typing import Any, Union
+
+import toml
+
+from .paths import APPS_REPO_ROOT
+
+
+def git_repo_age(path: Path) -> Union[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
+
+
+@cache
+def get_catalog(working_only: bool = False) -> dict[str, dict[str, Any]]:
+    """Load the app catalog and filter out the non-working ones"""
+    catalog = toml.load((APPS_REPO_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
+
+
+@cache
+def get_categories() -> dict[str, Any]:
+    categories_path = APPS_REPO_ROOT / "categories.toml"
+    return toml.load(categories_path)
+
+
+@cache
+def get_antifeatures() -> dict[str, Any]:
+    antifeatures_path = APPS_REPO_ROOT / "antifeatures.toml"
+    return toml.load(antifeatures_path)
+
+
+@cache
+def get_wishlist() -> dict[str, dict[str, str]]:
+    wishlist_path = APPS_REPO_ROOT / "wishlist.toml"
+    return toml.load(wishlist_path)
+
+
+@cache
+def get_graveyard() -> dict[str, dict[str, str]]:
+    wishlist_path = APPS_REPO_ROOT / "graveyard.toml"
+    return toml.load(wishlist_path)