dekun
66 lines
1.9 KiB
Python
66 lines
1.9 KiB
Python
"""Linux 上 vnpy_ctp 连接 CTP 前须设置有效 locale（否则 C++ 层 abort）。"""
|
||
from __future__ import annotations
|
||
|
||
import locale
|
||
import logging
|
||
import os
|
||
import subprocess
|
||
|
||
logger = logging.getLogger(__name__)
|
||
|
||
_LOCALE_DONE = False
|
||
_LOCALE_NAME = ""
|
||
|
||
|
||
def _list_locale_candidates() -> list[str]:
|
||
names: list[str] = []
|
||
for item in (
|
||
"C.UTF-8",
|
||
"en_US.UTF-8",
|
||
"en_US.utf8",
|
||
"POSIX",
|
||
"C",
|
||
):
|
||
if item not in names:
|
||
names.append(item)
|
||
try:
|
||
out = subprocess.check_output(["locale", "-a"], text=True, stderr=subprocess.DEVNULL)
|
||
for line in out.splitlines():
|
||
loc = line.strip()
|
||
if not loc:
|
||
continue
|
||
low = loc.lower()
|
||
if "utf" in low and loc not in names:
|
||
names.insert(0, loc)
|
||
except (OSError, subprocess.SubprocessError):
|
||
pass
|
||
return names
|
||
|
||
|
||
def ensure_process_locale() -> str:
|
||
"""强制设置进程 locale，覆盖系统里无效的旧值。"""
|
||
global _LOCALE_DONE, _LOCALE_NAME
|
||
if _LOCALE_DONE:
|
||
return _LOCALE_NAME
|
||
|
||
last_err: locale.Error | None = None
|
||
for name in _list_locale_candidates():
|
||
try:
|
||
locale.setlocale(locale.LC_ALL, name)
|
||
os.environ["LANG"] = name
|
||
os.environ["LC_ALL"] = name
|
||
os.environ["LC_CTYPE"] = name
|
||
_LOCALE_DONE = True
|
||
_LOCALE_NAME = name
|
||
logger.info("进程 locale 已设置: %s", name)
|
||
return name
|
||
except locale.Error as exc:
|
||
last_err = exc
|
||
continue
|
||
|
||
raise RuntimeError(
|
||
"未找到可用 locale，vnpy_ctp 会在 CTP 登录后崩溃。"
|
||
"请执行: apt install -y locales && locale-gen en_US.UTF-8 && "
|
||
"update-locale LANG=en_US.UTF-8 LC_ALL=en_US.UTF-8"
|
||
) from last_err
|