Skip to content

[BUG]在 SQL Server 2008 + pymssql + 中文 varchar 数据场景下,查询命中行为与中文解码表现不一致。 #1036

@firo1603

Description

@firo1603

SQLBot Version
1.7.0

Run Mode
Docker 容器运行。

Describe the bug
在 SQL Server 2008 + pymssql + 中文 varchar 数据场景下,查询命中行为与中文解码表现不一致。

  • 当 extraJdbc 不设置 charset 时,中文条件通常能匹配到记录,但返回中文字段可能乱码。
  • 当 extraJdbc 设置 charset=GBK 时,中文显示可能改善,但相同中文 WHERE 条件有概率出现漏数据或结果不稳定。

这会导致“查询稳定性”和“中文可读性”难以同时满足。

To Reproduce
Steps to reproduce the behavior:

  1. 在 SQLBot Web 中新增 SQL Server 数据源,数据库为 SQL Server 2008,表中包含中文 varchar 数据(例如 country='智利'、fruit='樱桃')。
  2. 将 extraJdbc 留空,执行:
    SELECT * FROM table_name WHERE country = '智利' AND fruit = '樱桃';
  3. 观察结果:通常能命中数据,但中文字段可能乱码。
  4. 将同一数据源 extraJdbc 改为 charset=GBK,再执行相同 SQL。
  5. 观察结果:中文显示可能改善,但查询命中可能不稳定(部分预期记录丢失)。

Expected behavior
A clear and concise description of what you expected to happen.

  • 无论 extraJdbc 是否配置 charset=GBK,查询命中应保持一致、稳定。
  • 中文字段应正确显示,不应出现乱码。
  • charset 应仅作为结果解码优先级,不应影响连接层查询匹配行为。

Screenshots
If applicable, add screenshots to help explain your problem.

无。

Additional context
Add any other context about the problem here.

环境信息:

  • SQL Server 2008
  • 驱动:pymssql

根因分析:

  • extraJdbc 中 charset/encoding 被直接透传到 pymssql.connect,导致其不仅影响解码,还可能影响查询匹配行为。
  • bytes 解码策略过于单一,无法覆盖 SQL Server 2008 历史中文数据。
  • BIT 与短字节文本判定边界不清晰。

针对 db.py 的修复建议代码:

  1. 分离 SQL Server 连接参数与解码参数(不要把 charset 透传进 connect)。
SQLSERVER_DECODE_CHARSET_KEYS = ("charset", "encoding", "client_charset")


def _normalize_charset(charset: str | None) -> str | None:
   if not charset:
      return None
   value = charset.strip()
   if not value:
      return None
   lower = value.lower().replace("-", "")
   if lower in {"cp936", "ms936"}:
      return "gbk"
   return value


def get_sqlserver_result_charset(conf: DatasourceConf) -> str | None:
   extra_config = get_extra_config(conf)
   for key, value in extra_config.items():
      if key.lower() in SQLSERVER_DECODE_CHARSET_KEYS:
         return _normalize_charset(str(value))
   # SQL Server 2008 中文场景兜底
   return "gbk"


def get_sqlserver_connect_extra_config(conf: DatasourceConf):
   extra_config = get_extra_config(conf)
   connect_config = {}
   for key, value in extra_config.items():
      if key.lower() in SQLSERVER_DECODE_CHARSET_KEYS:
         continue
      connect_config[key] = value
   return connect_config
  1. 调整 get_origin_connect:连接时仅传 connect_config。
def get_origin_connect(type: str, conf: DatasourceConf):
   extra_config_dict = get_sqlserver_connect_extra_config(conf)
   if equals_ignore_case(type, "sqlServer"):
      if conf.lowVersion is None or conf.lowVersion:
         return pymssql.connect(
            server=conf.host,
            port=str(conf.port),
            user=conf.username,
            password=conf.password,
            database=conf.database,
            timeout=conf.timeout,
            tds_version="7.0",
            **extra_config_dict,
         )
      return pymssql.connect(
         server=conf.host,
         port=str(conf.port),
         user=conf.username,
         password=conf.password,
         database=conf.database,
         timeout=conf.timeout,
         **extra_config_dict,
      )
  1. 强化 extraJdbc 与参数校验解析(split("=", 1))。
def get_extra_config(conf: DatasourceConf):
   config_dict = {}
   if conf.extraJdbc:
      config_arr = conf.extraJdbc.split("&")
      for config in config_arr:
         if not config:
            continue
         kv = config.split("=", 1)
         if len(kv) == 2 and kv[0] and kv[1]:
            config_dict[kv[0]] = kv[1]
         else:
            raise Exception(f"param: {config} is error")
   return config_dict


def checkParams(extraParams: str, illegalParams: List[str]):
   if not extraParams:
      return
   kvs = extraParams.split("&")
   for kv in kvs:
      if kv and "=" in kv:
         k, v = kv.split("=", 1)
         if k in illegalParams:
            raise HTTPException(status_code=500, detail=f"Illegal Parameter: {k}")
  1. 强化 convert_value:优先使用 preferred_charset,且仅把单字节 0/1 当作 BIT。
def convert_value(value, datetime_format="space", preferred_charset: Optional[str] = None):
   if value is None:
      return None

   if isinstance(value, str):
      return _repair_mojibake_text(value, preferred_charset)

   if isinstance(value, bytes):
      if len(value) == 1 and value in (b"\x00", b"\x01"):
         return value == b"\x01"

      if preferred_charset:
         try:
            return _repair_mojibake_text(value.decode(preferred_charset), preferred_charset)
         except (LookupError, UnicodeDecodeError):
            pass

      try:
         return _repair_mojibake_text(value.decode("utf-8"), preferred_charset)
      except UnicodeDecodeError:
         if any(b < 32 and b not in (9, 10, 13) for b in value):
            return f"0x{value.hex()}"
         return _repair_mojibake_text(value.decode("latin-1"), preferred_charset)

   if isinstance(value, bytearray):
      return convert_value(bytes(value), datetime_format=datetime_format, preferred_charset=preferred_charset)

   return value
  1. 在 exec_sql 中统一注入 preferred_charset。
def exec_sql(ds: CoreDatasource | AssistantOutDsSchema, sql: str, origin_column=False):
   # ... SQL 读校验逻辑略
   db = DB.get_db(ds.type)
   preferred_charset = None
   if equals_ignore_case(ds.type, "sqlServer"):
      try:
         conf_for_decode = DatasourceConf(**json.loads(aes_decrypt(ds.configuration)))
         preferred_charset = get_sqlserver_result_charset(conf_for_decode)
      except Exception:
         preferred_charset = "gbk"

   # 后续每个结果映射都使用:
   # convert_value(value, preferred_charset=preferred_charset)

回归验证建议:

  1. extraJdbc 为空时执行中文条件查询,确认命中正常。
  2. extraJdbc=charset=GBK 时执行相同 SQL,确认命中结果一致。
  3. 对比两种配置返回的中文字段,确认无乱码或明显改善。
  4. 验证 BIT、日期、Decimal 返回类型是否符合预期。

Metadata

Metadata

Labels

bugSomething isn't working

Type

No type

Projects

No projects

Milestone

No milestone

Relationships

None yet

Development

No branches or pull requests

Issue actions