allow sending diagnostic logs from client

mergin/client.py

@@ -19,7 +19,7 @@
 
 from typing import List
 
-from .common import ClientError, LoginError, WorkspaceRole, ProjectRole
+from .common import ClientError, LoginError, WorkspaceRole, ProjectRole, LOG_FILE_SIZE_TO_SEND, MERGIN_DEFAULT_LOGS_URL
 from .merginproject import MerginProject
 from .client_pull import (
     download_file_finalize,
@@ -1360,3 +1360,64 @@ def remove_project_collaborator(self, project_id: str, user_id: int):
         Remove a user from project collaborators
         """
         self.delete(f"v2/projects/{project_id}/collaborators/{user_id}")
+
+    def server_config(self) -> dict:
+        """Get server configuration as dictionary."""
+        response = self.get("/config")
+        return json.load(response)
+
+    def send_logs(
+        self,
+        logfile: str,
+        global_log_file: typing.Optional[str] = None,
+        application: typing.Optional[str] = None,
+        meta: typing.Optional[str] = None,
+    ):
+        """Send logs to configured server or the default Mergin server."""
+
+        if application is None:
+            application = "mergin-client-{}".format(__version__)
+
+        params = {"app": application, "username": self.username()}
+
+        config = self.server_config()
+        diagnostic_logs_url = config.get("diagnostic_logs_url", None)
+
+        use_server_api = False
+        if is_version_acceptable(self.server_version(), "2025.4.1") and (
+            diagnostic_logs_url is None or diagnostic_logs_url == ""
+        ):
+            url = "v2/diagnostic-logs" + "?" + urllib.parse.urlencode(params)
+            use_server_api = True
+        else:
+            if diagnostic_logs_url:
+                url = diagnostic_logs_url + "?" + urllib.parse.urlencode(params)
+            else:
+                # fallback to default logs URL
+                url = MERGIN_DEFAULT_LOGS_URL + "?" + urllib.parse.urlencode(params)
+
+        if meta is None:
+            meta = "Python API Client\nSystem: {} \nMergin Maps URL: {} \nMergin Maps user: {} \n--------------------------------\n\n".format(
+                platform.system(), self.url, self.username()
+            )
+
+        global_logs = b""
+        if global_log_file and os.path.exists(global_log_file):
+            with open(global_log_file, "rb") as f:
+                if os.path.getsize(global_log_file) > LOG_FILE_SIZE_TO_SEND:
+                    f.seek(-LOG_FILE_SIZE_TO_SEND, os.SEEK_END)
+                global_logs = f.read() + b"\n--------------------------------\n\n"
+
+        with open(logfile, "rb") as f:
+            if os.path.getsize(logfile) > 512 * 1024:
+                f.seek(-512 * 1024, os.SEEK_END)
+            logs = f.read()
+
+        payload = meta.encode() + global_logs + logs
+        header = {"content-type": "text/plain"}
+
+        if use_server_api:
+            return self.post(url, data=payload, headers=header)
+        else:
+            request = urllib.request.Request(url, data=payload, headers=header)
+            return self._do_request(request)

mergin/common.py

@@ -6,6 +6,11 @@
 # there is an upper limit for chunk size on server, ideally should be requested from there once implemented
 UPLOAD_CHUNK_SIZE = 10 * 1024 * 1024
 
+# size of the log file part to send (if file is larger only this size from end will be sent)
+LOG_FILE_SIZE_TO_SEND = 100 * 1024
+
+# default URL for submitting logs
+MERGIN_DEFAULT_LOGS_URL = "https://g4pfq226j0.execute-api.eu-west-1.amazonaws.com/mergin_client_log_submit"
 
 this_dir = os.path.dirname(os.path.realpath(__file__))
 

mergin/test/test_client.py

@@ -2833,3 +2833,41 @@ def test_access_management(mc: MerginClient, mc2: MerginClient):
     with pytest.raises(ClientError) as exc_info:
         mc.remove_workspace_member(workspace_id, new_user["id"])
     assert exc_info.value.http_error == 404
+
+
+def test_server_config(mc: MerginClient):
+    """Test retrieving server configuration and some keys."""
+    config = mc.server_config()
+
+    assert config
+    assert isinstance(config, dict)
+
+    assert "server_type" in config
+    assert "version" in config
+    assert "server_configured" in config
+
+
+def test_send_logs(mc: MerginClient, monkeypatch):
+    """Test that logs can be send to the server."""
+    test_project = "test_logs_send"
+    project = API_USER + "/" + test_project
+    project_dir = os.path.join(TMP_DIR, test_project)
+
+    cleanup(mc, project, [project_dir])
+    # prepare local project
+    shutil.copytree(TEST_DATA_DIR, project_dir)
+
+    # create remote project
+    mc.create_project_and_push(project, directory=project_dir)
+
+    # patch mc.server_config() to return empty config which means that logs will be send to the server
+    # but it is not configured to accept them so client error with message will be raised
+    def server_config(self):
+        return {}
+
+    monkeypatch.setattr(mc, "server_config", server_config.__get__(mc))
+
+    logs_path = os.path.join(project_dir, ".mergin", "client-log.txt")
+
+    with pytest.raises(ClientError, match="The requested URL was not found on the server"):
+        mc.send_logs(logs_path)