dtm_python/app/services/cache_service.py

"""缓存服务 - 统一管理Redis缓存和内存缓存"""
import redis
import pickle
import hashlib
import urllib.parse
import os
import pandas as pd
from flask import jsonify
from typing import Dict, Any, Optional

# Redis配置
REDIS_HOST = os.getenv('REDIS_HOST', 'localhost')
REDIS_PORT = int(os.getenv('REDIS_PORT', 6379))
REDIS_DB = int(os.getenv('REDIS_DB', 0))
REDIS_PASSWORD = os.getenv('REDIS_PASSWORD', None)
CACHE_PREFIX = os.getenv('CACHE_PREFIX', 'sku_analysis')
CACHE_EXPIRE_SECONDS = 86400  # 24小时
CACHE_VERSION = '2.0'

# 全局内存缓存
_global_cache: Dict[str, Any] = {
    'latest_df': None,
    'latest_analysis_results': None,
    'latest_spu_analysis_results': None,
    'latest_hotproduct_results': None,
    'latest_spu_hotproduct_results': None,
    'latest_filename': None
}

def get_redis_connection():
    """创建Redis连接"""
    try:
        redis_client = redis.Redis(
            host=REDIS_HOST,
            port=REDIS_PORT,
            db=REDIS_DB,
            password=REDIS_PASSWORD,
            decode_responses=False,
            socket_connect_timeout=5,
            socket_timeout=5
        )
        # 测试连接
        redis_client.ping()
        return redis_client
    except Exception as e:
        print(f"无法连接到Redis服务器: {e}")
        return None

def _ensure_list_type(keys_result):
    """确保keys_result是列表类型"""
    # 处理可能的异步结果或其他非列表类型
    if keys_result is None:
        return []
    
    # 如果是字节字符串，转换为列表
    if isinstance(keys_result, bytes):
        return [keys_result]
    
    # 如果已经是列表或元组，直接转换为列表
    if isinstance(keys_result, (list, tuple)):
        return list(keys_result)
    
    # 尝试转换为列表
    try:
        return list(keys_result)
    except (TypeError, ValueError):
        # 如果无法转换，返回空列表
        return []

def _ensure_int_type(value):
    """确保值是整数类型"""
    if value is None:
        return 0
    if isinstance(value, int):
        return value
    try:
        return int(value)
    except (TypeError, ValueError):
        return 0

def get_cache_info():
    """获取缓存信息"""
    try:
        # 初始化Redis连接
        redis_client = get_redis_connection()
        if not redis_client:
            return jsonify({'message': 'Redis连接失败'}), 500
        
        # 获取所有以sku_analysis开头的键（包括旧版本和新版本的缓存）
        keys_result = redis_client.keys(f"{CACHE_PREFIX}:*")
        
        # 确保keys_result是列表类型
        keys = _ensure_list_type(keys_result)
        
        cache_details = []
        
        if keys:
            for key in keys:
                try:
                    key_str = key.decode('utf-8') if isinstance(key, bytes) else key
                    # 获取缓存数据大小
                    size = redis_client.memory_usage(key)
                    
                    # 获取过期时间
                    ttl = redis_client.ttl(key)
                    # 确保ttl是整数类型
                    ttl_int = _ensure_int_type(ttl)
                    
                    # 从键中提取文件名信息
                    filename = "未知文件"
                    if ":" in key_str:
                        parts = key_str.split(":")
                        if len(parts) >= 3:
                            filename = parts[2]
                    
                    cache_info = {
                        'key': key_str,
                        'filename': filename,
                        'size': f"{size} bytes" if size else "未知",
                        'ttl': f"{ttl_int} 秒" if ttl_int > 0 else "永久"
                    }
                    
                    cache_details.append(cache_info)
                except Exception as e:
                    print(f"获取缓存键详细信息失败: {e}")
        
        return jsonify({
            'message': '缓存信息获取成功',
            'keyCount': len(cache_details),
            'cacheDetails': cache_details
        })
    except Exception as e:
        return jsonify({'error': f'获取缓存信息失败: {str(e)}'}), 500

def clear_cache():
    """清空缓存"""
    try:
        # 初始化Redis连接
        redis_client = get_redis_connection()
        if not redis_client:
            return jsonify({'message': 'Redis连接失败'}), 500
        
        # 清空所有以sku_analysis开头的键（包括所有版本的缓存）
        keys_result = redis_client.keys(f"{CACHE_PREFIX}:*")
        
        # 确保 keys_result 是列表类型
        keys = _ensure_list_type(keys_result)
        
        deleted_count = 0
        if keys:
            deleted_count = redis_client.delete(*keys)
        
        return jsonify({'message': '缓存清空成功', 'deletedCount': deleted_count})
    except Exception as e:
        return jsonify({'error': f'清空缓存失败: {str(e)}'}), 500


# ==================== Redis缓存管理函数 ====================

def generate_cache_key(df_hash: str, filename: Optional[str] = None) -> str:
    """生成缓存键（包含版本号）"""
    return generate_cache_key_with_prefix(df_hash, filename, CACHE_PREFIX)


def generate_cache_key_with_prefix(
    df_hash: str,
    filename: Optional[str] = None,
    prefix: Optional[str] = None
) -> str:
    """生成缓存键，允许自定义前缀"""
    actual_prefix = prefix or CACHE_PREFIX
    if filename:
        encoded_filename = urllib.parse.quote(filename, safe='')
        key_str = f"{actual_prefix}:v{CACHE_VERSION}:{df_hash}:{encoded_filename}"
    else:
        key_str = f"{actual_prefix}:v{CACHE_VERSION}:{df_hash}"
    return key_str


def save_to_cache(redis_client, cache_key: str, data: Any) -> bool:
    """将数据保存到Redis缓存"""
    if redis_client is None:
        return False
    try:
        serialized_data = pickle.dumps(data)
        redis_client.setex(cache_key, CACHE_EXPIRE_SECONDS, serialized_data)
        return True
    except Exception as e:
        print(f"保存到缓存失败: {e}")
        return False


def load_from_cache(redis_client, cache_key: str) -> Optional[Any]:
    """仏Redis缓存加载数据"""
    if redis_client is None:
        return None
    try:
        cached_data = redis_client.get(cache_key)
        if cached_data:
            return pickle.loads(cached_data)
        return None
    except Exception as e:
        print(f"从缓存加载失败: {e}")
        return None


def validate_analysis_results(results: Any) -> bool:
    """验证分析结果的数据完整性"""
    if not results or not isinstance(results, dict):
        return False
    
    sku_keys = [k for k in results.keys() if k != '_analysis_summary_']
    if not sku_keys:
        return False
    
    first_sku = sku_keys[0]
    sku_data = results.get(first_sku, {})
    
    required_fields = [
        'revenue_peak_idx', 'quantity_peak_idx',
        'peak_revenue_date', 'peak_quantity_date',
        'stages_map', 'stage_boundaries'
    ]
    
    for field in required_fields:
        if field not in sku_data:
            print(f"⚠️ 缓存数据缺少字段: {field}，将重新计算")
            return False
    
    boundaries = sku_data.get('stage_boundaries', [])
    if boundaries:
        for boundary in boundaries:
            if 'index' not in boundary or 'date' not in boundary:
                print(f"⚠️ 缓存数据的 stage_boundaries 格式不完整，将重新计算")
                return False
    
    return True


# ==================== 全局内存缓存管理 ====================

def set_global_cache(key: str, value: Any) -> None:
    """设置全局内存缓存"""
    print(f"💾 [CACHE] 设置缓存 key={key}, value_type={type(value).__name__}")
    _global_cache[key] = value
    print(f"💾 [CACHE] 缓存设置完成")


def get_global_cache(key: str) -> Any:
    """获取全局内存缓存"""
    value = _global_cache.get(key)
    print(f"🔍 [CACHE] 获取缓存 key={key}, value_type={type(value).__name__}, is_none={value is None}")
    return value


def clear_global_cache() -> None:
    """清空全局内存缓存"""
    _global_cache['latest_df'] = None
    _global_cache['latest_analysis_results'] = None
    _global_cache['latest_spu_analysis_results'] = None
    _global_cache['latest_hotproduct_results'] = None
    _global_cache['latest_spu_hotproduct_results'] = None
    _global_cache['latest_filename'] = None