dekun
97 lines
3.0 KiB
Python
97 lines
3.0 KiB
Python
# Copyright (c) 2025-2026 马建军. All rights reserved.
|
||
# 专有软件 — 未经授权禁止复制、传播、转售。
|
||
# 严禁用于：带单/代客理财、向他人推荐期货品种或买卖建议、融资配资等业务。
|
||
# 详见 LICENSE.zh-CN.txt 与 docs/软件购买与使用协议.md
|
||
|
||
"""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
|