mirror of
https://github.com/AstrBotDevs/AstrBot.git
synced 2026-09-01 15:32:49 +08:00
fix: fall back on Windows skill file encodings (#6058)
Co-authored-by: stablegenius49 <185121704+stablegenius49@users.noreply.github.com>
This commit is contained in:
@@ -53,31 +53,45 @@ def _ensure_safe_path(path: str) -> str:
|
||||
return abs_path
|
||||
|
||||
|
||||
def _decode_shell_output(output: bytes | None) -> str:
|
||||
def _decode_bytes_with_fallback(
|
||||
output: bytes | None,
|
||||
*,
|
||||
preferred_encoding: str | None = None,
|
||||
) -> str:
|
||||
if output is None:
|
||||
return ""
|
||||
|
||||
preferred = locale.getpreferredencoding(False) or "utf-8"
|
||||
try:
|
||||
return output.decode("utf-8")
|
||||
except (LookupError, UnicodeDecodeError):
|
||||
pass
|
||||
attempted_encodings: list[str] = []
|
||||
|
||||
def _try_decode(encoding: str) -> str | None:
|
||||
normalized = encoding.lower()
|
||||
if normalized in attempted_encodings:
|
||||
return None
|
||||
attempted_encodings.append(normalized)
|
||||
try:
|
||||
return output.decode(encoding)
|
||||
except (LookupError, UnicodeDecodeError):
|
||||
return None
|
||||
|
||||
for encoding in filter(None, [preferred_encoding, "utf-8", "utf-8-sig"]):
|
||||
if decoded := _try_decode(encoding):
|
||||
return decoded
|
||||
|
||||
if os.name == "nt":
|
||||
for encoding in ("mbcs", "cp936", "gbk", "gb18030"):
|
||||
try:
|
||||
return output.decode(encoding)
|
||||
except (LookupError, UnicodeDecodeError):
|
||||
continue
|
||||
|
||||
try:
|
||||
return output.decode(preferred)
|
||||
except (LookupError, UnicodeDecodeError):
|
||||
pass
|
||||
for encoding in ("mbcs", "cp936", "gbk", "gb18030", preferred):
|
||||
if decoded := _try_decode(encoding):
|
||||
return decoded
|
||||
elif decoded := _try_decode(preferred):
|
||||
return decoded
|
||||
|
||||
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")
|
||||
|
||||
|
||||
@dataclass
|
||||
class LocalShellComponent(ShellComponent):
|
||||
async def exec(
|
||||
@@ -184,8 +198,12 @@ class LocalFileSystemComponent(FileSystemComponent):
|
||||
async def read_file(self, path: str, encoding: str = "utf-8") -> dict[str, Any]:
|
||||
def _run() -> dict[str, Any]:
|
||||
abs_path = _ensure_safe_path(path)
|
||||
with open(abs_path, encoding=encoding) as f:
|
||||
content = f.read()
|
||||
with open(abs_path, "rb") as f:
|
||||
raw_content = f.read()
|
||||
content = _decode_bytes_with_fallback(
|
||||
raw_content,
|
||||
preferred_encoding=encoding,
|
||||
)
|
||||
return {"success": True, "content": content}
|
||||
|
||||
return await asyncio.to_thread(_run)
|
||||
|
||||
@@ -0,0 +1,57 @@
|
||||
from __future__ import annotations
|
||||
|
||||
import asyncio
|
||||
from pathlib import Path
|
||||
|
||||
from astrbot.core.computer.booters import local as local_booter
|
||||
from astrbot.core.computer.booters.local import LocalFileSystemComponent
|
||||
|
||||
|
||||
def _allow_tmp_root(monkeypatch, tmp_path: Path) -> None:
|
||||
monkeypatch.setattr(local_booter, "get_astrbot_root", lambda: str(tmp_path))
|
||||
monkeypatch.setattr(local_booter, "get_astrbot_data_path", lambda: str(tmp_path))
|
||||
monkeypatch.setattr(local_booter, "get_astrbot_temp_path", lambda: str(tmp_path))
|
||||
|
||||
|
||||
def test_local_file_system_component_prefers_utf8_before_windows_locale(
|
||||
monkeypatch,
|
||||
tmp_path: Path,
|
||||
):
|
||||
_allow_tmp_root(monkeypatch, tmp_path)
|
||||
monkeypatch.setattr(local_booter.os, "name", "nt", raising=False)
|
||||
monkeypatch.setattr(
|
||||
local_booter.locale,
|
||||
"getpreferredencoding",
|
||||
lambda _do_setlocale=False: "cp936",
|
||||
)
|
||||
|
||||
skill_path = tmp_path / "skills" / "demo.txt"
|
||||
skill_path.parent.mkdir(parents=True, exist_ok=True)
|
||||
skill_path.write_bytes("技能内容".encode("utf-8"))
|
||||
|
||||
result = asyncio.run(LocalFileSystemComponent().read_file(str(skill_path)))
|
||||
|
||||
assert result["success"] is True
|
||||
assert result["content"] == "技能内容"
|
||||
|
||||
|
||||
def test_local_file_system_component_falls_back_to_gbk_on_windows(
|
||||
monkeypatch,
|
||||
tmp_path: Path,
|
||||
):
|
||||
_allow_tmp_root(monkeypatch, tmp_path)
|
||||
monkeypatch.setattr(local_booter.os, "name", "nt", raising=False)
|
||||
monkeypatch.setattr(
|
||||
local_booter.locale,
|
||||
"getpreferredencoding",
|
||||
lambda _do_setlocale=False: "cp1252",
|
||||
)
|
||||
|
||||
skill_path = tmp_path / "skills" / "weibo-hot.txt"
|
||||
skill_path.parent.mkdir(parents=True, exist_ok=True)
|
||||
skill_path.write_bytes("微博热搜".encode("gbk"))
|
||||
|
||||
result = asyncio.run(LocalFileSystemComponent().read_file(str(skill_path)))
|
||||
|
||||
assert result["success"] is True
|
||||
assert result["content"] == "微博热搜"
|
||||
Reference in New Issue
Block a user