"""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 = "" # CTP C++ API 登录回调依赖中文 locale(见 vnpy/vnpy_ctp#24) _CTP_REQUIRED_LOCALES = ("zh_CN.GB18030", "zh_CN.gb18030") def _available_locales() -> set[str]: try: out = subprocess.check_output(["locale", "-a"], text=True, stderr=subprocess.DEVNULL) return {line.strip() for line in out.splitlines() if line.strip()} except (OSError, subprocess.SubprocessError): return set() def missing_ctp_locales() -> list[str]: """CTP 所需的 zh_CN.GB18030 是否已安装。""" avail = {x.lower() for x in _available_locales()} if any(x.lower() in avail for x in _CTP_REQUIRED_LOCALES): return [] return ["zh_CN.GB18030"] def _list_locale_candidates() -> list[str]: avail = _available_locales() names: list[str] = [] # CTP 回调优先尝试中文 locale for item in ( "zh_CN.GB18030", "zh_CN.gb18030", "zh_CN.UTF-8", "zh_CN.utf8", "en_US.UTF-8", "en_US.utf8", "C.UTF-8", "C.utf8", "POSIX", "C", ): if item in avail and item not in names: names.append(item) for loc in sorted(avail): low = loc.lower() if "utf" in low and loc not in names: names.append(loc) return names def ensure_process_locale() -> str: """强制设置进程 locale,覆盖系统里无效的旧值。""" global _LOCALE_DONE, _LOCALE_NAME if _LOCALE_DONE: return _LOCALE_NAME missing = missing_ctp_locales() if missing: raise RuntimeError( "CTP 需要中文 locale zh_CN.GB18030,当前系统未安装。" "请执行: sed -i '/^# zh_CN.GB18030/s/^# //' /etc/locale.gen && " "locale-gen zh_CN.GB18030" ) 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 zh_CN.GB18030 en_US.UTF-8" ) from last_err