refactoring

gotify.py

@@ -1,4 +1,5 @@
 import requests
+import subprocess
 
 
 class Gotify:
@@ -20,5 +21,20 @@ class Gotify:
         return self.send(priority=5, title=title, text=text)
 
     def send_error(self, title="", text=""):
-        return self.send(priority=10, title=title, text=text)
+        return self.send(priority=10, title=f"❗💀❗ {title}", text=text)
+
+    def send_subprocess_error(self, title, error: subprocess.CalledProcessError):
+        text = f"stdout:\n{error.stdout}\nstderr:\n{error.stderr}"
+        return self.send_error(title=title, text=text)
+
+    def send_backup_successful(self, borg_result: subprocess.CompletedProcess):
+        return self.send_success("Backup completed", borg_result)
+
+    def send_success(self, title, process_result: subprocess.CompletedProcess):
+        text = "\n".join([process_result.stdout, process_result.stderr])
+        return self.send_info(f"✅ {title}", text)
+
+    def send_warning(self, title, text):
+        return self.send(priority=5, title=f"❗ {title}", text=text)
+
 

nextcloud_backup

@@ -27,19 +27,17 @@ class BackupManager:
                 cmd, shell=True, text=True, check=True, capture_output=True
             )
         except subprocess.CalledProcessError as e:
-            self._gotify.send_error(
+            self._gotify.send_subprocess_error(
-                "❗💀❗ Enabling maintenace failed",
+                title="Enabling maintenance failed.", error=e
-                f"stdout:\n{e.stdout}\nstderr:\n{e.stderr}",
             )
             return False
         if "Maintenance mode already enabled" in result.stdout:
-            self._gotify.send_info(
+            title = "Maintenance unexpectedly enabled"
-                "❗ Maintenance unexpectedly enabled",
+            text = (
-                (
                 "Maintenance mode was already enabled. "
                 "Did not expect that. Will continue."
-                ),
             )
+            self._gotify.send_warning(title, text)
         return True
 
     def disable_maintenance(self):
@@ -49,19 +47,15 @@ class BackupManager:
                 cmd, shell=True, text=True, check=True, capture_output=True
             )
         except subprocess.CalledProcessError as e:
-            self._gotify.send_error(
+            self._gotify.send_subprocess_error("Disabling maintenance failed", e)
-                "❗💀❗ Disabling maintenace failed",
-                f"stdout:\n{e.stdout}\nstderr:\n{e.stderr}",
-            )
             return False
         if "Maintenance mode already disabled" in result.stdout:
-            self._gotify.send_info(
+            title = "Maintenance mode unexpectedly already disabled"
-                "❗ Maintenance mode unexpectedly already disabled",
+            text = (
-                (
                 "Maintenance mode was already disabled. "
                 "Did not expect that. Will continue."
-                ),
             )
+            self._gotify.send_warning(title, text)
         return True
 
     def dump_database(self):
@@ -79,13 +73,9 @@ class BackupManager:
                 capture_output=True,
             )
         except subprocess.CalledProcessError as e:
-            self._gotify.send_error(
+            self._gotify.send_subprocess_error("Dumping database failed", e)
-                "❗💀❗ Dumping Database failed",
-                f"stdout:\n{e.stdout}\nstderr:\n{e.stderr}",
-            )
             return False
-        text = "\n".join([result.stdout, result.stderr])
+        self._gotify.send_backup_successful(result)
-        self._gotify.send_info("✅ Database dumped", f"Result:\n{text}")
         return True
 
     def borg_backup(self):
@@ -110,13 +100,9 @@ class BackupManager:
                 env=borg_env,
             )
         except subprocess.CalledProcessError as e:
-            self._gotify.send_error(
+            self._gotify.send_subprocess_error("Backup failed", e)
-                title="❗💀❗ Backup failed!",
-                text=f"stdout: \n{e.stdout}\nsterr: \n{e.stderr}",
-            )
             return False
-        text = "\n".join([result.stdout, result.stderr])
+        self._gotify.send_backup_successful(result)
-        self._gotify.send_info("✅ Backup completed", f"Result:\n{text}\n")
         return True
 
 

paperless-ngx_backup

@@ -23,16 +23,15 @@ class BackupManager:
     def export_data(self):
         cmd = "docker compose exec -it webserver document_exporter ../export -d -f"
         try:
-            result = subprocess.run(cmd, shell=True, text=True, check=True, capture_output=True)
+            result = subprocess.run(
-        except subprocess.CalledProcessError as e:
+                cmd, shell=True, text=True, check=True, capture_output=True
-            self._gotify.send_error(
-                "❗💀❗ Exporting data failed",
-                f"stdout:\n{e.stdout}\nstderr:\n{e.stderr}",
             )
+        except subprocess.CalledProcessError as e:
+            self._gotify.send_subprocess_error("Exporting data failed", e)
             return False
+        self._gotify.send_success("Data exported.", result)
         return True
 
-
     def borg_backup(self):
         backup_dirs = " ".join(self._config["BACKUP_DIRS"])
         exclude_dirs = " ".join(self._config["EXCLUDE_DIRS"])
@@ -55,13 +54,9 @@ class BackupManager:
                 env=borg_env,
             )
         except subprocess.CalledProcessError as e:
-            self._gotify.send_error(
+            self._gotify.send_subprocess_error("Backup failed", e)
-                title="❗💀❗ Backup failed!",
-                text=f"stdout: \n{e.stdout}\nsterr: \n{e.stderr}",
-            )
             return False
-        text = "\n".join([result.stdout, result.stderr])
+        self._gotify.send_backup_successful(result)
-        self._gotify.send_info("✅ Backup completed", f"Result:\n{text}\n")
         return True
 
 
@@ -78,4 +73,3 @@ def main():
 
 if __name__ == "__main__":
     main()
-