feat(storage): add StorageManager skeleton with fallback detection

app/storage_manager.py

@@ -0,0 +1,87 @@
+import logging
+import time
+
+from sqlalchemy import text
+
+logger = logging.getLogger(__name__)
+
+
+class StorageManager:
+    """存储管理器骨架，封装 Redis / MySQL 可用性检测与三级降级状态。"""
+
+    def __init__(self, redis_client=None, mysql_engine=None, check_interval=30):
+        self.redis_client = redis_client
+        self.mysql_engine = mysql_engine
+        self.check_interval = check_interval
+
+        self.redis_available = False
+        self.mysql_available = False
+        self.last_redis_check = 0
+        self.last_mysql_check = 0
+
+    def check_redis(self):
+        """惰性检测 Redis 可用性，30 秒间隔内返回缓存结果。"""
+        now = time.time()
+        if now - self.last_redis_check < self.check_interval:
+            return self.redis_available
+
+        if self.redis_client is None:
+            self.redis_available = False
+        else:
+            try:
+                self.redis_client.ping()
+                self.redis_available = True
+            except Exception as e:
+                self.redis_available = False
+
+        self.last_redis_check = now
+        return self.redis_available
+
+    def check_mysql(self):
+        """惰性检测 MySQL 可用性，30 秒间隔内返回缓存结果。"""
+        now = time.time()
+        if now - self.last_mysql_check < self.check_interval:
+            return self.mysql_available
+
+        if self.mysql_engine is None:
+            self.mysql_available = False
+        else:
+            try:
+                with self.mysql_engine.connect() as conn:
+                    conn.execute(text("SELECT 1"))
+                self.mysql_available = True
+            except Exception as e:
+                self.mysql_available = False
+
+        self.last_mysql_check = now
+        return self.mysql_available
+
+    def is_available(self):
+        """判断 Redis 或 MySQL 至少一个可用。"""
+        return self.check_redis() or self.check_mysql()
+
+    def initialize(self):
+        """启动时调用，检测各后端可用性并输出日志。"""
+        redis_ok = self.check_redis()
+        mysql_ok = self.check_mysql()
+
+        if redis_ok:
+            logger.info("Redis 可用")
+        else:
+            logger.warning("Redis 不可用")
+
+        if mysql_ok:
+            logger.info("MySQL 可用")
+        else:
+            logger.warning("MySQL 不可用")
+
+
+_storage_manager = None
+
+
+def get_storage_manager():
+    """获取全局 StorageManager 单例。"""
+    global _storage_manager
+    if _storage_manager is None:
+        _storage_manager = StorageManager()
+    return _storage_manager