cache.py

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
数据缓存层
提供基金列表和单只基金数据的内存缓存，避免重复网络请求
"""

import time
from typing import Any, Dict, List, Optional

from services.data_fetcher import FundDataFetcher
from utils.logger import logger


class DataCache:
    """数据缓存管理器 - 内存缓存，带TTL过期机制"""

    def __init__(self):
        self._fund_list: Optional[List] = None
        self._fund_list_ts: float = 0.0
        self._fund_list_ttl: float = 86400.0  # 24小时

        self._fund_details: Dict[str, Dict[str, Any]] = {}
        self._fund_details_ts: Dict[str, float] = {}
        self._fund_detail_ttl: float = 60.0  # 60秒

    def get_fund_list(self) -> List:
        """获取基金列表（带缓存）

        Returns:
            基金列表 [[code, type, name, type2, pinyin], ...]
        """
        now = time.time()
        if self._fund_list is not None and (now - self._fund_list_ts < self._fund_list_ttl):
            logger.debug(f"命中基金列表缓存，共 {len(self._fund_list)} 条")
            return self._fund_list

        try:
            self._fund_list = FundDataFetcher.get_all_funds()
            self._fund_list_ts = now
            logger.info(f"刷新基金列表缓存，共 {len(self._fund_list)} 条")
            return self._fund_list
        except Exception as e:
            logger.error(f"获取基金列表失败: {e}")
            return self._fund_list or []

    def get_fund(self, code: str) -> Optional[Dict[str, Any]]:
        """获取单只基金数据（带缓存）

        Args:
            code: 基金代码

        Returns:
            基金数据字典，失败返回 None
        """
        now = time.time()
        ts = self._fund_details_ts.get(code, 0.0)

        if code in self._fund_details and (now - ts < self._fund_detail_ttl):
            return self._fund_details[code]

        try:
            result = FundDataFetcher.get_fund(code)
            if result:
                self._fund_details[code] = result
                self._fund_details_ts[code] = now
            return result
        except Exception as e:
            logger.error(f"获取基金{code}数据失败: {e}")
            return self._fund_details.get(code)

    def invalidate_fund(self, code: str) -> None:
        """使单只基金缓存失效

        Args:
            code: 基金代码
        """
        self._fund_details.pop(code, None)
        self._fund_details_ts.pop(code, None)

    def invalidate_all(self) -> None:
        """使所有缓存失效"""
        self._fund_list = None
        self._fund_list_ts = 0.0
        self._fund_details.clear()
        self._fund_details_ts.clear()
        logger.info("已清除所有数据缓存")


# 全局单例
_cache: Optional[DataCache] = None


def get_data_cache() -> DataCache:
    """获取全局数据缓存实例

    Returns:
        DataCache 单例
    """
    global _cache
    if _cache is None:
        _cache = DataCache()
    return _cache