NextcloudApp: new setup_nextcloud_logging function for transparent …

…logging

Signed-off-by: Alexander Piskun <bigcat88@icloud.com>

CHANGELOG.md

@@ -4,9 +4,13 @@ All notable changes to this project will be documented in this file.
 
 ## [0.17.1 - 2024-09-06]
 
+### Added
+
+- NextcloudApp: `setup_nextcloud_logging` function to support transparently sending logs to Nextcloud. #294
+
 ### Fixed
 
-- NextcloudApp: `nc.log` now suppresses all exceptions to safe call it anywhere in your app.
+- NextcloudApp: `nc.log` now suppresses all exceptions to safe call it anywhere in your app. #293
 
 ## [0.17.0 - 2024-09-05]
 

nc_py_api/ex_app/__init__.py

@@ -10,6 +10,7 @@
     set_handlers,
     talk_bot_msg,
 )
+from .logging import setup_nextcloud_logging
 from .misc import (
     get_computation_device,
     get_model_path,

nc_py_api/ex_app/logging.py

@@ -0,0 +1,47 @@
+"""Transparent logging support to store logs in the nextcloud.log."""
+
+import logging
+
+from ..nextcloud import NextcloudApp
+from .defs import LogLvl
+
+
+def _python_loglvl_to_nextcloud(levelno: int) -> LogLvl:
+    if levelno in (logging.NOTSET, logging.DEBUG):
+        return LogLvl.DEBUG
+    if levelno == logging.INFO:
+        return LogLvl.INFO
+    if levelno == logging.WARNING:
+        return LogLvl.WARNING
+    if levelno == logging.ERROR:
+        return LogLvl.ERROR
+    return LogLvl.FATAL
+
+
+class _NextcloudStorageHandler(logging.Handler):
+    def __init__(self):
+        super().__init__()
+        self.lock_flag = False
+
+    def emit(self, record):
+        if self.lock_flag:
+            return
+
+        try:
+            self.lock_flag = True
+            log_entry = self.format(record)
+            log_level = record.levelno
+            NextcloudApp().log(_python_loglvl_to_nextcloud(log_level), log_entry, fast_send=True)
+        except Exception:  # noqa pylint: disable=broad-exception-caught
+            self.handleError(record)
+        finally:
+            self.lock_flag = False
+
+
+def setup_nextcloud_logging(logger_name: str | None = None, logging_level: int = logging.DEBUG):
+    """Function to easily send all or selected log entries to Nextcloud."""
+    logger = logging.getLogger(logger_name)
+    nextcloud_handler = _NextcloudStorageHandler()
+    nextcloud_handler.setLevel(logging_level)
+    logger.addHandler(nextcloud_handler)
+    return nextcloud_handler

nc_py_api/nextcloud.py

@@ -348,15 +348,16 @@ def enabled_state(self) -> bool:
             return bool(self._session.ocs("GET", "/ocs/v1.php/apps/app_api/ex-app/state"))
         return False
 
-    def log(self, log_lvl: LogLvl, content: str) -> None:
+    def log(self, log_lvl: LogLvl, content: str, fast_send: bool = False) -> None:
         """Writes log to the Nextcloud log file."""
-        if self.check_capabilities("app_api"):
-            return
         int_log_lvl = int(log_lvl)
         if int_log_lvl < 0 or int_log_lvl > 4:
             raise ValueError("Invalid `log_lvl` value")
-        if int_log_lvl < self.capabilities["app_api"].get("loglevel", 0):
-            return
+        if not fast_send:
+            if self.check_capabilities("app_api"):
+                return
+            if int_log_lvl < self.capabilities["app_api"].get("loglevel", 0):
+                return
         with contextlib.suppress(Exception):
             self._session.ocs("POST", f"{self._session.ae_url}/log", json={"level": int_log_lvl, "message": content})
 
@@ -482,15 +483,16 @@ async def enabled_state(self) -> bool:
             return bool(await self._session.ocs("GET", "/ocs/v1.php/apps/app_api/ex-app/state"))
         return False
 
-    async def log(self, log_lvl: LogLvl, content: str) -> None:
+    async def log(self, log_lvl: LogLvl, content: str, fast_send: bool = False) -> None:
         """Writes log to the Nextcloud log file."""
-        if await self.check_capabilities("app_api"):
-            return
         int_log_lvl = int(log_lvl)
         if int_log_lvl < 0 or int_log_lvl > 4:
             raise ValueError("Invalid `log_lvl` value")
-        if int_log_lvl < (await self.capabilities)["app_api"].get("loglevel", 0):
-            return
+        if not fast_send:
+            if await self.check_capabilities("app_api"):
+                return
+            if int_log_lvl < (await self.capabilities)["app_api"].get("loglevel", 0):
+                return
         with contextlib.suppress(Exception):
             await self._session.ocs(
                 "POST", f"{self._session.ae_url}/log", json={"level": int_log_lvl, "message": content}