fix: 修复 ConfigStore 写入顺序、JSON 原子写入、JSON 损坏不回填、清理未使用导入

app/api/ai_config.py

@@ -1,9 +1,7 @@
 """
 AI模型配置接口 - 管理AI分析模型的配置
 """
-import json
 import logging
-from pathlib import Path
 from typing import Optional
 
 from fastapi import APIRouter, Depends, HTTPException

app/api/config.py

@@ -3,12 +3,9 @@
 """
 import json
 import logging
-import shutil
-from pathlib import Path
 from typing import Optional
 
 from fastapi import APIRouter, Depends, HTTPException, UploadFile, File, Body, Request
-from fastapi.responses import JSONResponse
 from sqlalchemy.orm import Session
 from pydantic import BaseModel
 

app/api/futures_analysis.py

@@ -3,7 +3,6 @@
 """
 import json
 import logging
-from pathlib import Path
 from typing import Optional
 import threading
 

app/config_store.py

@@ -65,30 +65,34 @@ class ConfigStore:
                     return db_value
 
                 # 数据库未命中：从 JSON 读取并回填
-                json_value = self._load_json(key, fallback)
+                json_value = self._load_json_from_file(key)
-                self._save_to_mysql(key, json_value)
+                if json_value is not None:
-                return json_value
+                    self._save_to_mysql(key, json_value)
+                    return json_value
+                # JSON 不存在或损坏，使用 fallback 但不回填
+                return fallback if fallback is not None else DEFAULT_CONFIGS.get(key, {})
             except Exception as e:
                 logger.warning(f"从 MySQL 读取配置 [{key}] 失败，降级到 JSON: {e}")
 
-        return self._load_json(key, fallback)
+        json_value = self._load_json_from_file(key)
+        if json_value is not None:
+            return json_value
+        return fallback if fallback is not None else DEFAULT_CONFIGS.get(key, {})
 
     def set_config(self, key: str, value: dict) -> bool:
-        """写入配置，始终写 JSON，MySQL 可用时同时写数据库。"""
+        """写入配置，MySQL 可用时先写数据库再写 JSON；MySQL 不可用时只写 JSON。"""
-        if key not in CONFIG_FILES:
+        if key not in self.config_files:
             raise ValueError(f"未知配置键: {key}")
 
-        # 始终写入 JSON 文件，保证 fallback 最新
-        self._save_to_json(key, value)
-
         if self.storage_manager.check_mysql():
             try:
                 self._save_to_mysql(key, value)
-                return True
             except Exception as e:
-                logger.warning(f"写入 MySQL 配置 [{key}] 失败，已保存到 JSON: {e}")
+                logger.warning(f"写入 MySQL 配置 [{key}] 失败，保留 JSON 旧值: {e}")
                 return False
 
+        # MySQL 写入成功或 MySQL 不可用，都更新 JSON fallback
+        self._save_to_json(key, value)
         return True
 
     def _get_from_mysql(self, key: str) -> Optional[dict]:
@@ -117,8 +121,8 @@ class ConfigStore:
             db.commit()
         return True
 
-    def _load_json(self, key: str, fallback: Optional[dict] = None) -> dict:
+    def _load_json_from_file(self, key: str) -> Optional[dict]:
-        """从 JSON 文件读取配置。"""
+        """从 JSON 文件读取配置，文件不存在或损坏时返回 None。"""
         file_path = self.config_files[key]
         if file_path.exists():
             try:
@@ -126,15 +130,28 @@ class ConfigStore:
                     return json.load(f)
             except Exception as e:
                 logger.warning(f"读取 JSON 配置 [{key}] 失败: {e}")
+        return None
 
+    def _load_json(self, key: str, fallback: Optional[dict] = None) -> dict:
+        """从 JSON 文件读取配置，文件不存在或损坏时返回 fallback。"""
+        value = self._load_json_from_file(key)
+        if value is not None:
+            return value
         return fallback if fallback is not None else DEFAULT_CONFIGS.get(key, {})
 
     def _save_to_json(self, key: str, value: dict) -> None:
-        """保存配置到 JSON 文件。"""
+        """保存配置到 JSON 文件（原子写入）。"""
         file_path = self.config_files[key]
         self.config_dir.mkdir(parents=True, exist_ok=True)
-        with open(file_path, "w", encoding="utf-8") as f:
+        tmp_path = file_path.with_suffix(".tmp")
-            json.dump(value, f, ensure_ascii=False, indent=4)
+        try:
+            with open(tmp_path, "w", encoding="utf-8") as f:
+                json.dump(value, f, ensure_ascii=False, indent=4)
+            tmp_path.replace(file_path)
+        except Exception:
+            if tmp_path.exists():
+                tmp_path.unlink()
+            raise
 
     def _config_exists_in_mysql(self, key: str) -> bool:
         """检查 MySQL 中是否已存在配置。"""

app/services/plan_generator.py

@@ -4,7 +4,6 @@ V2 交易计划生成器 - 5维度综合评分 + 多周期共振分析
 import json
 import logging
 from collections import Counter
-from pathlib import Path
 from typing import Dict, List, Optional, Tuple
 from datetime import datetime
 
@@ -17,9 +16,6 @@ from app.analysis_models import (
 
 logger = logging.getLogger(__name__)
 
-# 品种配置加载
-CONFIG_DIR = Path(__file__).resolve().parent.parent.parent / "config"
-
 # 板块分类
 SECTOR_MAP = {
     "贵金属": ["沪银", "沪金"],

docs/superpowers/specs/2026-07-04-config-to-mysql-design.md

@@ -65,9 +65,10 @@ class AppConfig(Base):
 
 ### 3.2 写入流程
 
-1. 始终写入 JSON 文件（保证 fallback 最新）
+1. 检查 MySQL 可用性
-2. MySQL 可用时同时写入 `app_config`
+2. MySQL 可用时先写入 `app_config`；失败时保留 JSON 旧值并返回 False
-3. MySQL 不可用时跳过数据库写入
+3. MySQL 写入成功后，再原子写入 JSON 文件
+4. MySQL 不可用时只写入 JSON 文件
 
 ### 3.3 降级检测