fix: decode local python output from bytes

astrbot/core/computer/booters/local.py

@@ -79,8 +79,15 @@ def _try_decode(encoding: str) -> str | None:
     return output.decode("utf-8", errors="replace")
 
 
-def _decode_shell_output(output: bytes | None) -> str:
-    return _decode_bytes_with_fallback(output, preferred_encoding="utf-8")
+def _decode_process_output(
+    output: bytes | None,
+    *,
+    normalize_newlines: bool = False,
+) -> str:
+    decoded = _decode_bytes_with_fallback(output, preferred_encoding="utf-8")
+    if normalize_newlines:
+        decoded = decoded.replace("\r\n", "\n")
+    return decoded
 
 
 @dataclass
@@ -127,8 +134,8 @@ def _run() -> dict[str, Any]:
                 capture_output=True,
             )
             return {
-                "stdout": _decode_shell_output(result.stdout),
-                "stderr": _decode_shell_output(result.stderr),
+                "stdout": _decode_process_output(result.stdout),
+                "stderr": _decode_process_output(result.stderr),
                 "exit_code": result.returncode,
             }
 
@@ -150,10 +157,20 @@ def _run() -> dict[str, Any]:
                     [os.environ.get("PYTHON", sys.executable), "-c", code],
                     timeout=timeout,
                     capture_output=True,
-                    text=True,
+                    text=False,
+                )
+                stdout = (
+                    ""
+                    if silent
+                    else _decode_process_output(
+                        result.stdout,
+                        normalize_newlines=True,
+                    )
+                )
+                stderr = _decode_process_output(
+                    result.stderr,
+                    normalize_newlines=True,
                 )
-                stdout = "" if silent else result.stdout
-                stderr = result.stderr if result.returncode != 0 else ""
                 return {
                     "data": {
                         "output": {"text": stdout, "images": []},

tests/unit/test_computer.py

@@ -5,6 +5,7 @@
 """
 
 import sys
+import subprocess
 from unittest.mock import AsyncMock, MagicMock, patch
 
 import pytest
@@ -225,6 +226,64 @@ async def test_exec_return_value(self):
         result = await python.exec("result = 1 + 1\nprint(result)")
         assert "2" in result["data"]["output"]["text"]
 
+    @pytest.mark.asyncio
+    async def test_exec_decodes_non_utf8_stdout_with_fallback(self):
+        """Test Python execution decodes captured bytes with fallback encodings."""
+        python = LocalPythonComponent()
+
+        def fake_run(*args, **kwargs):
+            assert kwargs.get("text") is False
+            return subprocess.CompletedProcess(
+                args=args[0],
+                returncode=0,
+                stdout="中文输出\n".encode("gbk"),
+                stderr=b"",
+            )
+
+        with patch("astrbot.core.computer.booters.local.subprocess.run", fake_run):
+            result = await python.exec("print('中文输出')")
+
+        assert result["data"]["output"]["text"] == "中文输出\n"
+
+    @pytest.mark.asyncio
+    async def test_exec_preserves_lone_carriage_returns(self):
+        """Test Python execution preserves lone carriage returns used by progress output."""
+        python = LocalPythonComponent()
+
+        def fake_run(*args, **kwargs):
+            assert kwargs.get("text") is False
+            return subprocess.CompletedProcess(
+                args=args[0],
+                returncode=0,
+                stdout=b"progress 10%\rprogress 20%\r\ncomplete\r",
+                stderr=b"",
+            )
+
+        with patch("astrbot.core.computer.booters.local.subprocess.run", fake_run):
+            result = await python.exec("print('progress')")
+
+        assert result["data"]["output"]["text"] == "progress 10%\rprogress 20%\ncomplete\r"
+
+    @pytest.mark.asyncio
+    async def test_exec_keeps_success_stderr(self):
+        """Test Python execution keeps diagnostic stderr even on success."""
+        python = LocalPythonComponent()
+
+        def fake_run(*args, **kwargs):
+            assert kwargs.get("text") is False
+            return subprocess.CompletedProcess(
+                args=args[0],
+                returncode=0,
+                stdout=b"ok\n",
+                stderr=b"warning\r\n",
+            )
+
+        with patch("astrbot.core.computer.booters.local.subprocess.run", fake_run):
+            result = await python.exec("print('ok')")
+
+        assert result["data"]["output"]["text"] == "ok\n"
+        assert result["data"]["error"] == "warning\n"
+
 
 class TestLocalFileSystemComponent:
     """Tests for LocalFileSystemComponent."""