fix: 修复 MySQL session 创建失败降级和迁移对象 detached 问题

app/analysis_db.py

@@ -38,11 +38,25 @@ def get_analysis_db():
     storage = get_storage_manager()
 
     if MySQLSessionLocal is not None and storage.check_mysql():
+        db = None
         try:
             db = MySQLSessionLocal()
             yield db
+        except Exception as e:
+            logger.warning(f"创建 MySQL analysis session 失败，降级到 SQLite: {e}")
+            if db is not None:
+                try:
+                    db.close()
+                except Exception:
+                    pass
+            db = AnalysisSessionLocal()
+            yield db
         finally:
-            db.close()
+            if db is not None:
+                try:
+                    db.close()
+                except Exception:
+                    pass
     else:
         db = AnalysisSessionLocal()
         try:

app/analysis_migration.py

@@ -3,7 +3,7 @@
 """
 import logging
 
-from sqlalchemy.orm import sessionmaker
+from sqlalchemy.orm import sessionmaker, make_transient
 
 from app.analysis_db import analysis_engine, AnalysisBase
 from app.analysis_models import (
@@ -104,8 +104,8 @@ def _write_mysql_analysis_data(data):
                     continue
 
                 for record in records:
-                    # 创建新对象，避免绑定到 SQLite session
-                    db.expunge(record)
+                    # 将对象从原 session 中脱离，使其可加入新 session
+                    make_transient(record)
                     db.add(record)
 
             db.commit()

tests/test_analysis_storage.py

@@ -149,3 +149,23 @@ class TestAnalysisMigration:
 
                         assert result is True
                         mock_write.assert_called_once_with({"ai_analysis_cache": [{"id": 1, "symbol": "AG2606"}]})
+
+    def test_get_analysis_db_falls_back_to_sqlite_when_mysql_session_fails(self):
+        """MySQL Session 创建失败时降级到 SQLite。"""
+        from app.analysis_db import get_analysis_db
+
+        mysql_session_maker = MagicMock(side_effect=RuntimeError("MySQL 连接失败"))
+        sqlite_session = MagicMock()
+        sqlite_session_maker = MagicMock(return_value=sqlite_session)
+
+        with patch("app.analysis_db.MySQLSessionLocal", mysql_session_maker):
+            with patch("app.analysis_db.AnalysisSessionLocal", sqlite_session_maker):
+                with patch("app.analysis_db.get_storage_manager") as mock_sm:
+                    sm = MagicMock()
+                    sm.check_mysql.return_value = True
+                    mock_sm.return_value = sm
+
+                    db = next(get_analysis_db())
+
+                    assert db is sqlite_session
+                    mysql_session_maker.assert_called_once()