Merge pull request #6917 from kozlovsky/fix/handle_non_utf8_core_output

Handle non-utf8 Tribler Core output in Tribler GUI

src/tribler/gui/core_manager.py

@@ -107,7 +107,7 @@ def on_core_stdout_read_ready(self):
             return
 
         raw_output = bytes(self.core_process.readAllStandardOutput())
-        self.last_core_stdout_output = raw_output.decode("utf-8").strip()
+        self.last_core_stdout_output = self.decode_raw_core_output(raw_output).strip()
 
         try:
             print(self.last_core_stdout_output)  # print core output # noqa: T001
@@ -121,7 +121,7 @@ def on_core_stderr_read_ready(self):
             return
 
         raw_output = bytes(self.core_process.readAllStandardError())
-        self.last_core_stderr_output = raw_output.decode("utf-8").strip()
+        self.last_core_stderr_output = self.decode_raw_core_output(raw_output).strip()
 
         try:
             print(self.last_core_stderr_output, file=sys.stderr)  # print core output # noqa: T001
@@ -162,3 +162,14 @@ def on_core_finished(self, exit_code, exit_status):
                     self.events_manager.connect_timer.stop()
 
             raise CoreCrashedError(error_message)
+
+    @staticmethod
+    def decode_raw_core_output(output: bytes) -> str:
+        try:
+            # Let's optimistically try to decode from UTF8.
+            # If it is not UTF8, we should get UnicodeDecodeError "invalid continuation byte".
+            return output.decode('utf-8')
+        except UnicodeDecodeError:
+            # It may be hard to guess the real encoding on some systems,
+            # but by using the "backslashreplace" error handler we can keep all the received data.
+            return output.decode('ascii', errors='backslashreplace')

src/tribler/gui/tests/test_core_manager.py

@@ -61,3 +61,9 @@ def test_on_core_read_ready_os_error_suppressed(core_manager):
     core_manager.on_core_stdout_read_ready()
     core_manager.on_core_stderr_read_ready()
     assert print.call_count == 2
+
+
+def test_decode_raw_core_output(core_manager):
+    assert core_manager.decode_raw_core_output(b'test') == 'test'
+    assert core_manager.decode_raw_core_output('test привет'.encode('utf-8')) == 'test привет'
+    assert core_manager.decode_raw_core_output('test привет'.encode('cp1251')) == r'test \xef\xf0\xe8\xe2\xe5\xf2'