logging_sender.py: permit to choose a matrix chan where to send the notification

appslib/logging_sender.py

@@ -1,16 +1,39 @@
 #!/usr/bin/env python3
 
 import subprocess
-from shutil import which
 import logging
 import logging.handlers
 
 
-def send_to_matrix(message: str) -> None:
+def notify(message, channel):
-    if which("sendxmpppy") is None:
+    print(f"{channel} -> {message}")
-        logging.warning("Could not send error via xmpp.")
+
-        return
+    chan_list = ["dev", "apps", "doc"]
-    subprocess.call(["sendxmpppy", message], stdout=subprocess.DEVNULL)
+
+    try:
+        any(channel in x for x in chan_list)
+    except False:
+        logging.warning(
+            f"Provided chan '{channel}' is not part of the available options ('dev', 'apps', 'doc')."
+        )
+
+    for char in ["'", "`", "!", ";", "$"]:
+        message = message.replace(char, "")
+
+    try:
+        subprocess.call(
+            [
+                "/var/www/webhooks/matrix-commander",
+                "--markdown -m '{message}' -c /var/www/webhooks/credentials.json --store /var/www/webhooks/store --room 'yunohost-{channel}'",
+            ],
+            stdout=subprocess.DEVNULL,
+        )
+    except subprocess.CalledProcessError as e:
+        logging.warning(
+            f"""Could not send a notification on {channel}.
+            Message: {message}
+            Error: {e}"""
+        )
 
 
 class LogSenderHandler(logging.Handler):
@@ -20,7 +43,7 @@ class LogSenderHandler(logging.Handler):
 
     def emit(self, record):
         msg = f"[Apps tools error] {record.msg}"
-        send_to_matrix(msg)
+        notify(msg, "dev")
 
     @classmethod
     def add(cls, level=logging.ERROR):

autoupdate_app_sources/autoupdate_app_sources.py

@@ -54,11 +54,13 @@ STRATEGIES = [
 
 
 @cache
-def get_github() -> tuple[
+def get_github() -> (
-    Optional[tuple[str, str]],
+    tuple[
-    Optional[github.Github],
+        Optional[tuple[str, str]],
-    Optional[github.InputGitAuthor],
+        Optional[github.Github],
-]:
+        Optional[github.InputGitAuthor],
+    ]
+):
     try:
         github_login = (
             (REPO_APPS_ROOT / ".github_login")
@@ -286,7 +288,6 @@ class AppAutoUpdater:
     def relevant_versions(
         tags: list[str], app_id: str, version_regex: Optional[str]
     ) -> tuple[str, str]:
-
         def apply_version_regex(tag: str) -> Optional[str]:
             # First preprocessing according to the manifest version_regex…
             if version_regex:
@@ -452,8 +453,8 @@ class AppAutoUpdater:
 
         api: Union[GithubAPI, GitlabAPI, GiteaForgejoAPI]
         if remote_type == "github":
-            assert upstream and upstream.startswith(
+            assert (
-                "https://github.com/"
+                upstream and upstream.startswith("https://github.com/")
             ), f"When using strategy {strategy}, having a defined upstream code repo on github.com is required"
             api = GithubAPI(upstream, auth=get_github()[0])
         if remote_type == "gitlab":
@@ -594,7 +595,6 @@ def paste_on_haste(data):
 
 
 class StdoutSwitch:
-
     class DummyFile:
         def __init__(self) -> None:
             self.result = ""
@@ -728,7 +728,7 @@ def main() -> None:
         paste_url = paste_on_haste(paste_message)
         matrix_message += f"\nSee the full log here: {paste_url}"
 
-    appslib.logging_sender.send_to_matrix(matrix_message)
+    appslib.logging_sender.notify(matrix_message, "apps")
     print(paste_message)