#!/usr/bin/env python3
  """
  缓存状态巡检脚本
  
  按「缓存类型」维度（embedding / translation / anchors）查看：
  - 估算 key 数量
  - TTL 分布（采样）
  - 近期活跃 key（按 IDLETIME 近似）
  - 若干样本 key 及 value 概览
  
  使用示例：
  
    # 默认：检查已知三类缓存，使用 env_config 中的 Redis 配置
    python scripts/redis/redis_cache_health_check.py
  
    # 只看某一类缓存
    python scripts/redis/redis_cache_health_check.py --type embedding
    python scripts/redis/redis_cache_health_check.py --type translation anchors
  
    # 自定义前缀（pattern），不限定缓存类型
    python scripts/redis/redis_cache_health_check.py --pattern "mycache:*"
  
    # 调整采样规模
    python scripts/redis/redis_cache_health_check.py --sample-size 100 --max-scan 50000
  """
  
  from __future__ import annotations
  
  import argparse
  import json

  import sys
  from collections import defaultdict
  from dataclasses import dataclass
  from datetime import datetime
  from pathlib import Path
  from typing import Dict, Iterable, List, Optional, Tuple
  
  import redis
  import numpy as np
  
  # 让脚本可以直接使用 config/env_config 与 services_config
  PROJECT_ROOT = Path(__file__).parent.parent.parent
  sys.path.insert(0, str(PROJECT_ROOT))
  
  from config.env_config import REDIS_CONFIG  # type: ignore
  from config.services_config import get_translation_cache_config  # type: ignore

  from embeddings.bf16 import decode_embedding_from_redis  # type: ignore

  
  
  @dataclass
  class CacheTypeConfig:
      name: str
      pattern: str
      description: str
  
  
  def _load_known_cache_types() -> Dict[str, CacheTypeConfig]:
      """根据当前配置装配三种已知缓存类型及其前缀 pattern。"""
      cache_types: Dict[str, CacheTypeConfig] = {}
  

      # embedding 缓存：prefix 来自 REDIS_CONFIG['embedding_cache_prefix']（默认 embedding）
      embedding_prefix = REDIS_CONFIG.get("embedding_cache_prefix", "embedding")

      cache_types["embedding"] = CacheTypeConfig(
          name="embedding",

          pattern=f"{embedding_prefix}:*",

          description="文本向量缓存（embeddings/text_encoder.py）",
      )
  
      # translation 缓存：prefix 来自 services.translation.cache.key_prefix
      cache_cfg = get_translation_cache_config()
      trans_prefix = cache_cfg.get("key_prefix", "trans:v2")
      cache_types["translation"] = CacheTypeConfig(
          name="translation",
          pattern=f"{trans_prefix}:*",
          description="翻译结果缓存（query/qwen_mt_translate.Translator）",
      )
  
      # anchors 缓存：prefix 来自 REDIS_CONFIG['anchor_cache_prefix']（若存在），否则 product_anchors
      anchor_prefix = REDIS_CONFIG.get("anchor_cache_prefix", "product_anchors")
      cache_types["anchors"] = CacheTypeConfig(
          name="anchors",
          pattern=f"{anchor_prefix}:*",
          description="商品内容理解缓存（indexer/product_enrich.py，anchors/语义属性/tags）",
      )
  
      return cache_types
  
  
  def get_redis_client(db: int = 0) -> redis.Redis:
      return redis.Redis(
          host=REDIS_CONFIG.get("host", "localhost"),
          port=REDIS_CONFIG.get("port", 6479),
          password=REDIS_CONFIG.get("password"),
          db=db,
          decode_responses=False,  # 原始 bytes，方便区分 pickle / str
          socket_timeout=10,
          socket_connect_timeout=10,
      )
  
  
  def scan_keys(
      client: redis.Redis, pattern: str, max_scan: int, scan_count: int = 1000
  ) -> List[bytes]:
      """使用 SCAN 扫描匹配 pattern 的 key，最多扫描 max_scan 个结果。"""
      keys: List[bytes] = []
      cursor: int = 0
      scanned = 0
      while True:
          cursor, batch = client.scan(cursor=cursor, match=pattern, count=scan_count)
          for k in batch:
              keys.append(k)
              scanned += 1
              if scanned >= max_scan:
                  return keys
          if cursor == 0:
              break
      return keys
  
  
  def ttl_bucket(ttl: int) -> str:
      """将 TTL（秒）归类到简短区间标签中。"""
      if ttl < 0:
          # -1: 永不过期；-2: 不存在
          return "no-expire-or-expired"
      if ttl <= 3600:
          return "0-1h"
      if ttl <= 86400:
          return "1h-1d"
      if ttl <= 30 * 86400:
          return "1d-30d"
      return ">30d"
  
  
  def format_seconds(sec: int) -> str:
      if sec < 0:
          return str(sec)
      if sec < 60:
          return f"{sec}s"
      if sec < 3600:
          return f"{sec // 60}m{sec % 60}s"
      if sec < 86400:
          h = sec // 3600
          m = (sec % 3600) // 60
          return f"{h}h{m}m"
      d = sec // 86400
      h = (sec % 86400) // 3600
      return f"{d}d{h}h"
  
  
  def decode_value_preview(
      cache_type: str, key: bytes, raw_value: Optional[bytes]
  ) -> str:
      """根据缓存类型生成简短的 value 概览字符串。"""
      if raw_value is None:
          return "<nil>"
  

      # embedding: BF16 bytes

      if cache_type == "embedding":
          try:

              arr = decode_embedding_from_redis(raw_value)

              if isinstance(arr, np.ndarray):

                  dim = int(arr.size)
                  return f"bf16 bytes={len(raw_value)} dim={dim} restored_dtype={arr.dtype}"
              return f"bf16 decode type={type(arr).__name__}"

          except Exception:
              return f"<binary {len(raw_value)} bytes>"
  
      # anchors: JSON dict
      if cache_type == "anchors":
          try:
              text = raw_value.decode("utf-8", errors="replace")
              obj = json.loads(text)
              if isinstance(obj, dict):
                  brief = {
                      k: obj.get(k)
                      for k in ["id", "lang", "title_input", "title", "category_path", "anchor_text"]
                      if k in obj
                  }
                  return "json " + json.dumps(brief, ensure_ascii=False)[:200]
              # 其他情况简单截断
              return "json " + text[:200]
          except Exception:
              return raw_value.decode("utf-8", errors="replace")[:200]
  
      # translation: 纯字符串
      if cache_type == "translation":
          try:
              text = raw_value.decode("utf-8", errors="replace")
              return text[:200]
          except Exception:
              return f"<binary {len(raw_value)} bytes>"
  
      # 兜底：尝试解码为 UTF-8
      try:
          text = raw_value.decode("utf-8", errors="replace")
          return text[:200]
      except Exception:
          return f"<binary {len(raw_value)} bytes>"
  
  
  def analyze_cache_type(
      client: redis.Redis,
      cache_type: str,
      cfg: CacheTypeConfig,
      sample_size: int,
      max_scan: int,
  ) -> None:
      """对单个缓存类型做统计与样本展示。"""
      print("=" * 80)
      print(f"Cache type: {cache_type}  pattern={cfg.pattern}")
      print(f"Description: {cfg.description}")
      print("=" * 80)
  
      keys = scan_keys(client, cfg.pattern, max_scan=max_scan)
      total_scanned = len(keys)
      print(f"Scanned up to {max_scan} keys, matched {total_scanned} keys.")
      if total_scanned == 0:
          print("No keys found for this cache type.\n")
          return
  
      # 采样
      if total_scanned <= sample_size:
          sampled_keys = keys
      else:
          # 简单取前 sample_size 个，scan 本身已是渐进遍历，足够近似随机
          sampled_keys = keys[:sample_size]
  
      ttl_hist: Dict[str, int] = defaultdict(int)
      recent_keys: List[Tuple[bytes, int, int]] = []  # (key, ttl, idletime)
      samples: List[Tuple[bytes, int, int, str]] = []  # (key, ttl, idletime, preview)
  
      for k in sampled_keys:
          try:
              ttl = client.ttl(k)
          except Exception:
              ttl = -3  # 表示 TTL 查询失败
  
          # TTL 分布
          ttl_hist[ttl_bucket(ttl)] += 1
  
          # 近期活跃判断（idletime 越小越“新”）
          idletime = -1
          try:
              # OBJECT IDLETIME 返回秒数（整数）
              idletime = client.object("idletime", k)  # type: ignore[arg-type]
          except Exception:
              pass
  
          # 记录近期活跃样本
          if idletime >= 0 and idletime <= 600:
              recent_keys.append((k, ttl, idletime))
  
          # 收集样本 value 预览（控制数量）
          if len(samples) < 5:
              raw_val = None
              try:
                  raw_val = client.get(k)
              except Exception:
                  pass
              preview = decode_value_preview(cache_type, k, raw_val)
              samples.append((k, ttl, idletime, preview))
  
      # TTL 分布输出
      print("\nTTL distribution (sampled):")
      total_sampled = len(sampled_keys)
      for bucket in ["no-expire-or-expired", "0-1h", "1h-1d", "1d-30d", ">30d"]:
          cnt = ttl_hist.get(bucket, 0)
          pct = (cnt / total_sampled * 100.0) if total_sampled else 0.0
          print(f"  {bucket:<18}: {cnt:>4} ({pct:>5.1f}%)")
  
      # 近期活跃 key
      recent_keys = sorted(recent_keys, key=lambda x: x[2])[:5]
      print("\nRecent active keys (idletime <= 600s, from sampled set):")
      if not recent_keys:
          print("  (none in sampled set)")
      else:
          for k, ttl, idle in recent_keys:
              try:
                  k_str = k.decode("utf-8", errors="replace")
              except Exception:
                  k_str = repr(k)
              if len(k_str) > 80:
                  k_str = k_str[:77] + "..."
              print(
                  f"  key={k_str}  ttl={ttl} ({format_seconds(ttl)})  "
                  f"idletime={idle} ({format_seconds(idle)})"
              )
  
      # 样本 value 概览
      print("\nSample keys & value preview:")
      if not samples:
          print("  (no samples)")
      else:
          for k, ttl, idle, preview in samples:
              try:
                  k_str = k.decode("utf-8", errors="replace")
              except Exception:
                  k_str = repr(k)
              if len(k_str) > 80:
                  k_str = k_str[:77] + "..."
              print(f"  key={k_str}")
              print(f"    ttl={ttl} ({format_seconds(ttl)})  idletime={idle} ({format_seconds(idle)})")
              print(f"    value: {preview}")
  
      print()  # 结尾空行
  
  
  def main() -> None:
      parser = argparse.ArgumentParser(description="Redis 缓存状态巡检（按缓存类型）")
      parser.add_argument(
          "--type",
          dest="types",
          nargs="+",
          choices=["embedding", "translation", "anchors"],
          help="指定要检查的缓存类型（默认：三种全部）",
      )
      parser.add_argument(
          "--pattern",
          type=str,
          help="自定义 key pattern（如 'mycache:*'）。设置后将忽略 --type，仅按 pattern 进行一次检查。",
      )
      parser.add_argument(
          "--db",
          type=int,
          default=0,
          help="Redis 数据库编号（默认 0）",
      )
      parser.add_argument(
          "--sample-size",
          type=int,
          default=50,
          help="每种缓存类型采样的 key 数量（默认 50）",
      )
      parser.add_argument(
          "--max-scan",
          type=int,
          default=20000,
          help="每种缓存类型最多 SCAN 的 key 数量上限（默认 20000）",
      )
  
      args = parser.parse_args()
  
      print("Redis 缓存状态巡检")
      print("=" * 80)
      print(f"Checked at: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}")
      print(f"Redis host={REDIS_CONFIG.get('host', 'localhost')} port={REDIS_CONFIG.get('port', 6479)} db={args.db}")
      print()
  
      try:
          client = get_redis_client(db=args.db)
          client.ping()
          print("✅ Redis 连接成功\n")
      except Exception as exc:
          print(f"❌ Redis 连接失败: {exc}")
          print(f"  Host: {REDIS_CONFIG.get('host', 'localhost')}")
          print(f"  Port: {REDIS_CONFIG.get('port', 6479)}")
          print(f"  Password: {'已配置' if REDIS_CONFIG.get('password') else '未配置'}")
          return
  
      # 如果指定了自定义 pattern，则只做一次“匿名类型”的巡检
      if args.pattern:
          anon_cfg = CacheTypeConfig(
              name=f"pattern:{args.pattern}",
              pattern=args.pattern,
              description="自定义 pattern 检查",
          )
          analyze_cache_type(
              client=client,
              cache_type="custom",
              cfg=anon_cfg,
              sample_size=args.sample_size,
              max_scan=args.max_scan,
          )
          print("巡检完成。")
          return
  
      # 否则根据已知缓存类型巡检
      known_types = _load_known_cache_types()
      types_to_check: Iterable[str]
      if args.types:
          types_to_check = args.types
      else:
          types_to_check = known_types.keys()
  
      for t in types_to_check:
          cfg = known_types.get(t)
          if not cfg:
              print(f"⚠️ 未知缓存类型: {t}，跳过")
              continue
          analyze_cache_type(
              client=client,
              cache_type=t,
              cfg=cfg,
              sample_size=args.sample_size,
              max_scan=args.max_scan,
          )
  
      print("巡检完成。")
  
  
  if __name__ == "__main__":
      main()