buffer_platform/app/storage_manager.py

import datetime as _datetime
import json
import logging
import time
from datetime import datetime

from sqlalchemy import text
from sqlalchemy.orm import sessionmaker

from app.config import CACHE_TTL_SECONDS, REDIS_TTL_SECONDS
from app.models import MarketData, SymbolTimestamp

logger = logging.getLogger(__name__)

# Redis 缓存 TTL：读取自 app.config.REDIS_TTL_SECONDS（默认 30 天）
# 业务缓存有效期：读取自 app.config.CACHE_TTL_SECONDS（默认 300 秒）


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 _parse_fetched_at(self, timestamp_raw):
        """优先使用输入 timestamp，解析失败时回退到当前时间。"""
        if not timestamp_raw:
            return _datetime.datetime.now()

        if isinstance(timestamp_raw, str):
            try:
                return _datetime.datetime.fromisoformat(timestamp_raw)
            except Exception:
                return _datetime.datetime.now()

        if hasattr(timestamp_raw, "isoformat"):
            return timestamp_raw

        return _datetime.datetime.now()

    def is_available(self):
        """判断 Redis 或 MySQL 至少一个可用。"""
        return self.check_redis() or self.check_mysql()

    def initialize(self, redis_client=None, mysql_engine=None):
        """启动时调用，可选注入 Redis/MySQL 客户端并检测各后端可用性。"""
        if redis_client is not None:
            self.redis_client = redis_client
        if mysql_engine is not None:
            self.mysql_engine = mysql_engine

        # 重置检测缓存，确保注入新客户端后重新探测后端可用性
        self.last_redis_check = 0
        self.last_mysql_check = 0

        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 不可用")

    def _discover_periods(self, symbol, data_type):
        """从 Redis 或 MySQL 发现该品种已缓存的所有周期。"""
        if self.check_redis():
            try:
                pattern = f"market_data:{symbol}:*"
                keys = self.redis_client.keys(pattern)
                periods = set()
                for key in keys:
                    key_str = key.decode("utf-8") if isinstance(key, bytes) else key
                    parts = key_str.split(":")
                    if len(parts) == 3:
                        periods.add(parts[2])
                if periods:
                    return sorted(periods)
            except Exception as e:
                logger.warning(f"Redis 发现周期失败 [{symbol}]: {e}")

        if not self.check_mysql():
            return []

        try:
            SessionLocal = sessionmaker(bind=self.mysql_engine)
            with SessionLocal() as db:
                rows = (
                    db.query(MarketData.period)
                    .filter_by(symbol=symbol, data_type=data_type)
                    .distinct()
                    .all()
                )
                return sorted([r.period for r in rows if r.period])
        except Exception as e:
            logger.warning(f"MySQL 发现周期失败 [{symbol}]: {e}")
            return []

    def _get_market_data_single_period(self, symbol, data_type, period):
        """读取单个周期的行情数据：Redis → MySQL → None。"""
        cache_key = f"market_data:{symbol}:{period}"

        # 1. 检查 Redis 缓存
        if self.check_redis():
            try:
                raw = self.redis_client.get(cache_key)
                if raw:
                    return json.loads(raw)
            except Exception as e:
                logger.warning(f"Redis 读取失败 [{cache_key}]: {e}")

        # 2. Redis 未命中，检查 MySQL 可用性
        if not self.check_mysql():
            return None

        # 3. 从 MySQL 读取
        try:
            SessionLocal = sessionmaker(bind=self.mysql_engine)
            with SessionLocal() as db:
                record = (
                    db.query(MarketData)
                    .filter_by(symbol=symbol, data_type=data_type, period=period)
                    .first()
                )
                if not record:
                    return None

                result = {
                    "current_price": record.current_price,
                    "timestamp": (
                        record.fetched_at.isoformat() if record.fetched_at else None
                    ),
                    "candles": (
                        json.loads(record.candles_json) if record.candles_json else []
                    ),
                }

                # 4. 回填 Redis（TTL 30 天）
                if self.check_redis():
                    try:
                        self.redis_client.setex(
                            cache_key,
                            REDIS_TTL_SECONDS,
                            json.dumps(result, ensure_ascii=False),
                        )
                    except Exception as e:
                        logger.warning(f"Redis 回填失败 [{cache_key}]: {e}")

                return result
        except Exception as e:
            logger.warning(f"MySQL 读取失败 [{cache_key}]: {e}")
            return None

    @staticmethod
    def _parse_candle_time(candle):
        """解析单根 K 线的时间字段，返回 naive datetime 或 None。"""
        candle_time = candle.get("datetime") or candle.get("time")
        if not candle_time:
            return None

        if isinstance(candle_time, str):
            try:
                dt = datetime.fromisoformat(candle_time.replace("Z", "+00:00"))
                if dt.tzinfo is not None:
                    dt = dt.replace(tzinfo=None)
                return dt
            except Exception:
                return None

        if hasattr(candle_time, "tzinfo") and candle_time.tzinfo is not None:
            return candle_time.replace(tzinfo=None)
        return candle_time

    def _filter_and_truncate(self, merged, end_time, max_candles):
        """在合并结果上按 end_time 与 max_candles 过滤/截断每个周期。"""
        if merged is None:
            return None

        filter_end_time = end_time if end_time else datetime.now()
        if filter_end_time.tzinfo is not None:
            filter_end_time = filter_end_time.replace(tzinfo=None)

        timeframes = {}
        for period, candles in merged.get("timeframes", {}).items():
            filtered = []
            for candle in candles:
                candle_dt = self._parse_candle_time(candle)
                if candle_dt is None or candle_dt <= filter_end_time:
                    filtered.append(candle)

            if len(filtered) > max_candles:
                filtered = filtered[-max_candles:]

            timeframes[period] = filtered

        merged["timeframes"] = timeframes
        return merged

    def _merge_period_results(self, symbol, data_type, period_results):
        """将按周期返回的结果合并为统一格式。"""
        timeframes = {}
        current_price = None
        newest = None

        for period, result in period_results.items():
            if not result:
                continue
            candles = result.get("candles", [])
            timeframes[period] = candles
            if current_price is None:
                current_price = result.get("current_price")
            timestamp_raw = result.get("timestamp")
            if timestamp_raw:
                try:
                    timestamp_dt = datetime.fromisoformat(timestamp_raw)
                    # 去掉时区信息，统一使用 naive datetime
                    if timestamp_dt.tzinfo is not None:
                        timestamp_dt = timestamp_dt.replace(tzinfo=None)
                    if newest is None or timestamp_dt > newest:
                        newest = timestamp_dt
                except Exception:
                    pass

        if not timeframes:
            return None

        if newest is None:
            newest = datetime.now()

        now = datetime.now()
        is_fresh = (now - newest).total_seconds() < CACHE_TTL_SECONDS

        return {
            "symbol": symbol,
            "type": data_type,
            "current_price": current_price,
            "timestamp": newest.isoformat(),
            "timeframes": timeframes,
            "is_fresh": is_fresh,
            "fetched_at": newest.isoformat(),
        }

    def get_market_data(self, symbol, data_type, periods=None, end_time=None, max_candles=100):
        """
        读取行情数据：Redis → MySQL → None（由调用方降级到 SQLite）。

        Args:
            symbol: 品种合约代码
            data_type: 数据类型，如 futures
            periods: 周期或周期列表，如 "5min" 或 ["5min", "15min"];
                     为 None 时自动发现全部周期
            end_time: 结束时间（可选），过滤该时间之前的 K 线
            max_candles: 每个周期最大 K 线数量，默认 100

        Returns:
            periods 为字符串时返回旧版单周期 dict;
            periods 为列表/None 时返回合并后的 dict（包含 timeframes）或 None
        """
        single_period = isinstance(periods, str)
        query_periods = [periods] if single_period else periods

        if query_periods is None:
            query_periods = self._discover_periods(symbol, data_type)
            if not query_periods:
                return None

        period_results = {}
        for period in query_periods:
            period_results[period] = self._get_market_data_single_period(
                symbol, data_type, period
            )

        merged = self._merge_period_results(symbol, data_type, period_results)
        if merged is None:
            return None

        merged = self._filter_and_truncate(merged, end_time, max_candles)

        if single_period:
            period = query_periods[0]
            return {
                "current_price": merged.get("current_price"),
                "timestamp": merged.get("timestamp"),
                "candles": merged["timeframes"].get(period, []),
            }

        return merged

    def get_symbol_timestamp(self, symbol, data_type):
        """
        读取合约时间戳：Redis → MySQL → None（由调用方降级到 SQLite）。

        Args:
            symbol: 品种合约代码
            data_type: 数据类型，如 futures

        Returns:
            dict 或 None
        """
        cache_key = f"symbol_timestamps:{symbol}"

        # 1. 检查 Redis 缓存
        if self.check_redis():
            try:
                raw = self.redis_client.get(cache_key)
                if raw:
                    return json.loads(raw)
            except Exception as e:
                logger.warning(f"Redis 读取失败 [{cache_key}]: {e}")

        # 2. Redis 未命中，检查 MySQL 可用性
        if not self.check_mysql():
            return None

        # 3. 从 MySQL 读取
        try:
            SessionLocal = sessionmaker(bind=self.mysql_engine)
            with SessionLocal() as db:
                record = (
                    db.query(SymbolTimestamp)
                    .filter_by(symbol=symbol, data_type=data_type)
                    .first()
                )
                if not record:
                    return None

                result = {
                    "last_refresh_at": (
                        record.last_refresh_at.isoformat()
                        if record.last_refresh_at
                        else None
                    ),
                    "refresh_count": record.refresh_count,
                }

                # 4. 回填 Redis（TTL 30 天）
                if self.check_redis():
                    try:
                        self.redis_client.setex(
                            cache_key,
                            REDIS_TTL_SECONDS,
                            json.dumps(result, ensure_ascii=False),
                        )
                    except Exception as e:
                        logger.warning(f"Redis 回填失败 [{cache_key}]: {e}")

                return result
        except Exception as e:
            logger.warning(f"MySQL 读取失败 [{cache_key}]: {e}")
            return None

    def save_market_data(self, symbol, data):
        """
        保存行情数据：先删 Redis 缓存，再写 MySQL，MySQL 成功后回填 Redis。

        Args:
            symbol: 品种合约代码
            data: 采集脚本返回的数据，包含 timeframes, current_price, timestamp, type

        Returns:
            bool: MySQL 写入是否成功
        """
        data_type = data.get("type", "futures")
        timeframes = data.get("timeframes", {})
        current_price = data.get("current_price")
        fetched_at = self._parse_fetched_at(data.get("timestamp"))

        cache_keys = [f"market_data:{symbol}:{period}" for period in timeframes.keys()]

        # 1. 删除 Redis 缓存
        if self.check_redis() and cache_keys:
            try:
                self.redis_client.delete(*cache_keys)
            except Exception as e:
                logger.warning(f"Redis 删除缓存失败 [{symbol}]: {e}")

        # 2. 写入 MySQL（事务）
        if not self.check_mysql():
            logger.warning(f"MySQL 不可用，无法写入行情数据 [{symbol}]")
            return False

        try:
            SessionLocal = sessionmaker(bind=self.mysql_engine)
            with SessionLocal() as db:
                try:
                    for period, candles in timeframes.items():
                        candles_json = json.dumps(candles, ensure_ascii=False)
                        record = (
                            db.query(MarketData)
                            .filter_by(symbol=symbol, data_type=data_type, period=period)
                            .first()
                        )
                        if record:
                            record.candles_json = candles_json
                            record.current_price = current_price
                            record.fetched_at = fetched_at
                            record.candle_count = len(candles)
                        else:
                            record = MarketData(
                                symbol=symbol,
                                data_type=data_type,
                                period=period,
                                candles_json=candles_json,
                                current_price=current_price,
                                fetched_at=fetched_at,
                                candle_count=len(candles),
                            )
                            db.add(record)
                    db.commit()
                except Exception:
                    db.rollback()
                    raise
        except Exception as e:
            logger.error(f"MySQL 写入行情数据失败 [{symbol}]: {e}")
            return False

        # 3. MySQL 成功，回填 Redis（失败仅记录日志）
        if self.check_redis():
            for period, candles in timeframes.items():
                cache_key = f"market_data:{symbol}:{period}"
                cache_value = {
                    "current_price": current_price,
                    "timestamp": fetched_at.isoformat(),
                    "candles": candles,
                }
                try:
                    self.redis_client.setex(
                        cache_key,
                        REDIS_TTL_SECONDS,
                        json.dumps(cache_value, ensure_ascii=False),
                    )
                except Exception as e:
                    logger.warning(f"Redis 更新失败 [{cache_key}]: {e}")

        return True

    def save_symbol_timestamp(self, symbol, data_type, refresh_time):
        """
        保存合约时间戳：先删 Redis 缓存，再写 MySQL，MySQL 成功后回填 Redis。

        Args:
            symbol: 品种合约代码
            data_type: 数据类型，如 futures
            refresh_time: 刷新时间（datetime）

        Returns:
            bool: MySQL 写入是否成功
        """
        cache_key = f"symbol_timestamps:{symbol}"

        # 1. 删除 Redis 缓存
        if self.check_redis():
            try:
                self.redis_client.delete(cache_key)
            except Exception as e:
                logger.warning(f"Redis 删除缓存失败 [{cache_key}]: {e}")

        # 2. 写入 MySQL（事务）
        if not self.check_mysql():
            logger.warning(f"MySQL 不可用，无法写入合约时间戳 [{symbol}]")
            return False

        try:
            SessionLocal = sessionmaker(bind=self.mysql_engine)
            with SessionLocal() as db:
                try:
                    record = (
                        db.query(SymbolTimestamp)
                        .filter_by(symbol=symbol, data_type=data_type)
                        .first()
                    )
                    if record:
                        record.last_refresh_at = refresh_time
                        record.refresh_count += 1
                    else:
                        record = SymbolTimestamp(
                            symbol=symbol,
                            data_type=data_type,
                            last_refresh_at=refresh_time,
                            refresh_count=1,
                        )
                        db.add(record)
                    db.commit()
                except Exception:
                    db.rollback()
                    raise
        except Exception as e:
            logger.error(f"MySQL 写入合约时间戳失败 [{symbol}]: {e}")
            return False

        # 3. MySQL 成功，回填 Redis（失败仅记录日志）
        if self.check_redis():
            cache_value = {
                "last_refresh_at": refresh_time.isoformat(),
                "refresh_count": record.refresh_count,
            }
            try:
                self.redis_client.setex(
                    cache_key,
                    REDIS_TTL_SECONDS,
                    json.dumps(cache_value, ensure_ascii=False),
                )
            except Exception as e:
                logger.warning(f"Redis 更新失败 [{cache_key}]: {e}")

        return True

    def save_market_data_with_timestamp(self, symbol, data, refresh_time):
        """
        在同一 MySQL 事务中原子写入行情数据与合约时间戳。

        Redis 缓存的删除/回填在事务外进行，失败仅记录日志，不影响 MySQL 写入。

        Args:
            symbol: 品种合约代码
            data: 采集脚本返回的数据，包含 timeframes, current_price, timestamp, type
            refresh_time: 刷新时间（datetime）

        Returns:
            bool: MySQL 写入是否成功
        """
        data_type = data.get("type", "futures")
        timeframes = data.get("timeframes", {})
        current_price = data.get("current_price")
        fetched_at = self._parse_fetched_at(data.get("timestamp"))

        market_cache_keys = [
            f"market_data:{symbol}:{period}" for period in timeframes.keys()
        ]
        timestamp_cache_key = f"symbol_timestamps:{symbol}"
        cache_keys = market_cache_keys + [timestamp_cache_key]

        # 1. 删除 Redis 缓存
        if self.check_redis() and cache_keys:
            try:
                self.redis_client.delete(*cache_keys)
            except Exception as e:
                logger.warning(f"Redis 删除缓存失败 [{symbol}]: {e}")

        # 2. 写入 MySQL（同一事务）
        if not self.check_mysql():
            logger.warning(f"MySQL 不可用，无法写入行情数据与时间戳 [{symbol}]")
            return False

        try:
            SessionLocal = sessionmaker(bind=self.mysql_engine)
            with SessionLocal() as db:
                try:
                    for period, candles in timeframes.items():
                        candles_json = json.dumps(candles, ensure_ascii=False)
                        record = (
                            db.query(MarketData)
                            .filter_by(symbol=symbol, data_type=data_type, period=period)
                            .first()
                        )
                        if record:
                            record.candles_json = candles_json
                            record.current_price = current_price
                            record.fetched_at = fetched_at
                            record.candle_count = len(candles)
                        else:
                            record = MarketData(
                                symbol=symbol,
                                data_type=data_type,
                                period=period,
                                candles_json=candles_json,
                                current_price=current_price,
                                fetched_at=fetched_at,
                                candle_count=len(candles),
                            )
                            db.add(record)

                    timestamp_record = (
                        db.query(SymbolTimestamp)
                        .filter_by(symbol=symbol, data_type=data_type)
                        .first()
                    )
                    if timestamp_record:
                        timestamp_record.last_refresh_at = refresh_time
                        timestamp_record.refresh_count += 1
                        refresh_count = timestamp_record.refresh_count
                    else:
                        timestamp_record = SymbolTimestamp(
                            symbol=symbol,
                            data_type=data_type,
                            last_refresh_at=refresh_time,
                            refresh_count=1,
                        )
                        db.add(timestamp_record)
                        refresh_count = 1

                    db.commit()
                except Exception:
                    db.rollback()
                    raise
        except Exception as e:
            logger.error(f"MySQL 写入行情数据与时间戳失败 [{symbol}]: {e}")
            return False

        # 3. MySQL 成功，回填 Redis（失败仅记录日志）
        if self.check_redis():
            for period, candles in timeframes.items():
                cache_key = f"market_data:{symbol}:{period}"
                cache_value = {
                    "current_price": current_price,
                    "timestamp": fetched_at.isoformat(),
                    "candles": candles,
                }
                try:
                    self.redis_client.setex(
                        cache_key,
                        REDIS_TTL_SECONDS,
                        json.dumps(cache_value, ensure_ascii=False),
                    )
                except Exception as e:
                    logger.warning(f"Redis 更新失败 [{cache_key}]: {e}")

            timestamp_cache_value = {
                "last_refresh_at": refresh_time.isoformat(),
                "refresh_count": refresh_count,
            }
            try:
                self.redis_client.setex(
                    timestamp_cache_key,
                    REDIS_TTL_SECONDS,
                    json.dumps(timestamp_cache_value, ensure_ascii=False),
                )
            except Exception as e:
                logger.warning(f"Redis 更新失败 [{timestamp_cache_key}]: {e}")

        return True

    def delete_cache(self, symbol, periods):
        """
        批量删除 Redis 行情数据缓存键。

        Args:
            symbol: 品种合约代码
            periods: 周期列表，如 ["5min", "15min"]
        """
        if not self.check_redis():
            return

        cache_keys = [f"market_data:{symbol}:{period}" for period in periods]
        if not cache_keys:
            return

        try:
            self.redis_client.delete(*cache_keys)
        except Exception as e:
            logger.warning(f"Redis 批量删除缓存失败 [{symbol}]: {e}")

    def cache_get(self, key):
        """
        通用 Redis 缓存读取。

        Args:
            key: 缓存键

        Returns:
            反序列化后的值，或 None
        """
        if not self.check_redis():
            return None

        try:
            raw = self.redis_client.get(key)
            if raw:
                return json.loads(raw)
        except Exception as e:
            logger.warning(f"Redis 读取失败 [{key}]: {e}")

        return None

    def cache_set(self, key, value, ttl=None):
        """
        通用 Redis 缓存写入。

        Args:
            key: 缓存键
            value: 缓存值（可 JSON 序列化）
            ttl: 过期时间（秒），默认 REDIS_TTL_SECONDS

        Returns:
            bool: 是否成功
        """
        if not self.check_redis():
            return False

        if ttl is None:
            ttl = REDIS_TTL_SECONDS

        try:
            self.redis_client.setex(
                key, ttl, json.dumps(value, ensure_ascii=False, default=str)
            )
            return True
        except Exception as e:
            logger.warning(f"Redis 写入失败 [{key}]: {e}")
            return False

    def cache_delete(self, key):
        """
        通用 Redis 缓存删除。

        Args:
            key: 缓存键
        """
        if not self.check_redis():
            return

        try:
            self.redis_client.delete(key)
        except Exception as e:
            logger.warning(f"Redis 删除失败 [{key}]: {e}")


_storage_manager = None


def get_storage_manager():
    """获取全局 StorageManager 单例。"""
    global _storage_manager
    if _storage_manager is None:
        _storage_manager = StorageManager()
    return _storage_manager