| """app_config 键值配置表的读写。 |
| |
| 取代原先各 *_config_manager 的 config/*.json 文件存储。各 manager 的公开方法 |
| （get_config/update_config/get/set/list_all 等）签名与行为不变，只是底层 _read/_write |
| 改调这里，从而让配置随 Postgres 持久化、HF 容器重启不丢。 |
| |
| 兼容旧库：首次读某个 key 时若数据库里还没有、但旧的 config/<x>.json 文件存在， |
| 就把文件内容一次性导入数据库后返回（幂等）。这样桌面端 / 本地 SQLite 的既有配置不丢。 |
| """ |
| import json |
| from pathlib import Path |
| from typing import Any, Optional |
|
|
| from app.db.engine import SessionLocal |
| from app.db.models.app_config import AppConfig |
|
|
|
|
| def get_value(key: str) -> Optional[Any]: |
| """读某个配置域的整份 value（dict 或 list）；不存在返回 None。 |
| |
| 容错：启动早期（路由在 init_db 建表前就被 import，某些下载器会在导入期读 cookie 配置） |
| 或数据库暂不可用时，按「无配置」返回 None，而不是抛错——与旧的「文件不存在/读失败返回 {}」 |
| 语义一致，避免把启动 import 链炸掉。 |
| """ |
| try: |
| with SessionLocal() as db: |
| row = db.get(AppConfig, key) |
| return row.value if row is not None else None |
| except Exception: |
| return None |
|
|
|
|
| def set_value(key: str, value: Any) -> None: |
| """整份覆盖某个配置域的 value（upsert）。""" |
| with SessionLocal() as db: |
| row = db.get(AppConfig, key) |
| if row is None: |
| db.add(AppConfig(key=key, value=value)) |
| else: |
| row.value = value |
| db.commit() |
|
|
|
|
| def _read_legacy_file(legacy_path: str) -> Optional[Any]: |
| p = Path(legacy_path) |
| if not p.exists(): |
| return None |
| try: |
| return json.loads(p.read_text(encoding="utf-8")) |
| except Exception: |
| return None |
|
|
|
|
| def load_value(key: str, legacy_path: Optional[str] = None, default: Any = None) -> Any: |
| """读配置；数据库没有但旧 JSON 文件存在则一次性导入后返回；都没有返回 default。""" |
| value = get_value(key) |
| if value is not None: |
| return value |
| if legacy_path: |
| legacy = _read_legacy_file(legacy_path) |
| if legacy is not None: |
| try: |
| set_value(key, legacy) |
| except Exception: |
| |
| pass |
| return legacy |
| return default |
|
|
