fix(config): 修复非法系统配置阻断启动 (#6702)

app/db/oper/systemconfig.py

@@ -1,9 +1,11 @@
 import copy
+import json
+import logging
 import threading
 from collections.abc import Callable, Mapping
 from typing import Any, Optional, TypeVar, Union
 
-from sqlalchemy import select
+from sqlalchemy import select, text
 from sqlalchemy.orm import Session
 
 from app.db.base import DbOper
@@ -12,6 +14,7 @@ from app.foundation.singleton import Singleton
 from app.schemas.types import SystemConfigKey
 
 T = TypeVar("T")
+logger = logging.getLogger(__name__)
 
 
 class SystemConfigOper(DbOper, metaclass=Singleton):
@@ -30,17 +33,45 @@ class SystemConfigOper(DbOper, metaclass=Singleton):
     def load_snapshot(self, db: Optional[Session] = None) -> None:
         """从显式会话或 Oper 事务边界加载配置并发布内存快照。"""
         with self._write_lock:
-            items = SystemConfig.list(db) if db is not None else self._execute_sync_query(
-                SystemConfig.list
-            )
-            snapshot = {
-                item.key: copy.deepcopy(item.value)
-                for item in items
-            }
+            snapshot = self._load_snapshot_values(db)
             with self._snapshot_lock:
                 self.__SYSTEMCONF = snapshot
                 self._loaded = True
 
+    def _load_snapshot_values(self, db: Optional[Session]) -> dict[str, Any]:
+        """逐行读取配置并把历史非法 JSON 修复为可继续运行的空值。"""
+        def load(session: Session) -> dict[str, Any]:
+            """读取原始列值，避免单条坏记录触发 SQLAlchemy JSON 反序列化。"""
+            snapshot: dict[str, Any] = {}
+            rows = session.execute(
+                text("SELECT id, key, value FROM systemconfig ORDER BY id")
+            ).mappings()
+            for row in rows:
+                key = row["key"]
+                if not key:
+                    continue
+                raw_value = row["value"]
+                try:
+                    value = (
+                        json.loads(raw_value)
+                        if isinstance(raw_value, (str, bytes, bytearray))
+                        else raw_value
+                    )
+                except (TypeError, ValueError, UnicodeDecodeError) as error:
+                    # 配置快照是启动关键路径；归零坏行后仍允许管理员进入系统重新保存。
+                    logger.warning(
+                        f"系统配置 JSON 无法解析，已重置为空值：id={row['id']}，key={key}，error={error}"
+                    )
+                    session.execute(
+                        text("UPDATE systemconfig SET value = NULL WHERE id = :id"),
+                        {"id": row["id"]},
+                    )
+                    value = None
+                snapshot[str(key)] = copy.deepcopy(value)
+            return snapshot
+
+        return load(db) if db is not None else self._execute_sync_query(load)
+
     def _require_loaded(self) -> None:
         """阻止消费者读取尚未完成启动加载的半成品快照。"""
         if not self._loaded:

tests/test_systemconfig_oper.py

@@ -1,9 +1,9 @@
 import threading
 import uuid
 from concurrent.futures import ThreadPoolExecutor
-from types import SimpleNamespace
 
 import pytest
+from sqlalchemy import text
 
 from app.db.models.systemconfig import SystemConfig
 from app.db.oper.systemconfig import SystemConfigOper
@@ -50,19 +50,19 @@ def test_load_snapshot_publishes_complete_dictionary(monkeypatch):
     """重新加载期间读取方只会看到完整旧快照或完整新快照。"""
     Singleton._instances.pop((SystemConfigOper, (), frozenset()), None)
     oper = SystemConfigOper()
-    values = [SimpleNamespace(key="key", value="old")]
     entered = threading.Event()
     release = threading.Event()
 
-    monkeypatch.setattr(SystemConfig, "list", lambda _db: values)
+    monkeypatch.setattr(oper, "_load_snapshot_values", lambda _db: {"key": "old"})
     oper.load_snapshot()
 
     def load_new_snapshot(_db):
+        """阻塞新快照构造，供主线程确认旧快照仍可读取。"""
         entered.set()
         release.wait(1)
-        return [SimpleNamespace(key="key", value="new")]
+        return {"key": "new"}
 
-    monkeypatch.setattr(SystemConfig, "list", load_new_snapshot)
+    monkeypatch.setattr(oper, "_load_snapshot_values", load_new_snapshot)
     thread = threading.Thread(target=oper.load_snapshot)
     thread.start()
     assert entered.wait(1)
@@ -75,6 +75,31 @@ def test_load_snapshot_publishes_complete_dictionary(monkeypatch):
     assert oper.get("key") == "new"
 
 
+def test_load_snapshot_repairs_malformed_json_and_allows_follow_up_write():
+    """启动加载遇到单条非法 JSON 时应修复该行，并允许随后覆盖配置。"""
+    key = _unique_key()
+    with SessionFactory() as session:
+        session.execute(
+            text("INSERT INTO systemconfig (key, value) VALUES (:key, :value)"),
+            {"key": key, "value": "not-json"},
+        )
+        session.commit()
+
+    oper = _fresh_oper()
+
+    assert oper.get(key) is None
+    with SessionFactory() as session:
+        stored = session.execute(
+            text("SELECT value FROM systemconfig WHERE key = :key"),
+            {"key": key},
+        ).scalar_one()
+    assert stored is None
+
+    assert oper.set(key, {"recovered": True}) is True
+    assert oper.get(key) == {"recovered": True}
+    assert _stored_config(key).value == {"recovered": True}
+
+
 def test_read_does_not_wait_for_slow_write_transaction(monkeypatch):
     """数据库写入期间，内存读取仍返回最近一次已提交值。"""
     oper = _fresh_oper()