"""
Ch13 配套代码 3 / 4 —— 内存综合分析器

演示：
  1. MEMORY STATS  —— 全局内存视图
  2. MEMORY USAGE  —— 抽样 Top N 大 Key
  3. OBJECT IDLETIME —— 找出长时间未访问的「冷 Key」
  4. 给出整体诊断（碎片率、Top 大 Key、冷 Key 占比）
"""

import random
import redis

r = redis.Redis(host="127.0.0.1", port=6379, decode_responses=True)


def section(title: str) -> None:
    print("\n" + "=" * 64)
    print(title)
    print("=" * 64)


def fmt_bytes(n) -> str:
    if n is None:
        return "-"
    n = float(n)
    for unit in ["B", "KB", "MB", "GB", "TB"]:
        if n < 1024:
            return f"{n:.2f}{unit}"
        n /= 1024
    return f"{n:.2f}PB"


def prepare_demo_data() -> None:
    section("Step 0: 准备一批演示数据")
    print("  - 100 个普通 String / 5 个大 Hash / 1 个超大 List")

    pipe = r.pipeline(transaction=False)
    for i in range(100):
        pipe.set(f"demo:mem:str:{i}", "x" * random.randint(50, 200))
    for i in range(5):
        mapping = {f"f{j}": "y" * 100 for j in range(200)}
        pipe.hset(f"demo:mem:hash:{i}", mapping=mapping)
    pipe.rpush("demo:mem:biglist", *[f"item{i}" for i in range(5000)])
    pipe.execute()
    print("  ✓ 已写入约 100+5+1 个 demo Key（前缀 demo:mem:）")


def cleanup_demo_data() -> None:
    keys = list(r.scan_iter(match="demo:mem:*", count=500))
    if keys:
        r.unlink(*keys)


def show_memory_stats() -> None:
    section("Step 1: MEMORY STATS —— 全局内存视图")
    stats = r.execute_command("MEMORY", "STATS")

    pairs = {}
    it = iter(stats)
    for k in it:
        v = next(it)
        pairs[k] = v

    interesting = [
        "peak.allocated",
        "total.allocated",
        "startup.allocated",
        "replication.backlog",
        "clients.slaves",
        "clients.normal",
        "aof.buffer",
        "lua.caches",
        "overhead.total",
        "keys.count",
        "keys.bytes-per-key",
        "dataset.bytes",
        "dataset.percentage",
        "allocator.allocated",
        "allocator.active",
        "allocator.resident",
        "allocator-fragmentation.ratio",
        "allocator-fragmentation.bytes",
        "allocator-rss.ratio",
        "rss-overhead.ratio",
        "fragmentation",
    ]
    print(f"  {'指标':<32} 值")
    print("  " + "-" * 60)
    for k in interesting:
        if k in pairs:
            v = pairs[k]
            if isinstance(v, (int, float)) and "ratio" not in k and "percentage" not in k and "count" not in k and "per-key" not in k:
                v = fmt_bytes(v)
            elif isinstance(v, float):
                v = f"{v:.4f}"
            print(f"  {k:<32} {v}")


def show_top_keys(top_n: int = 10) -> None:
    section(f"Step 2: MEMORY USAGE 抽样找 Top {top_n} 大 Key")

    sampled = []
    for k in r.scan_iter(match="demo:mem:*", count=500):
        try:
            usage = r.memory_usage(k)
            if usage is not None:
                sampled.append((k, usage))
        except redis.ResponseError:
            continue

    if not sampled:
        print("  （未发现 demo Key，先跑 prepare_demo_data）")
        return

    sampled.sort(key=lambda x: -x[1])
    print(f"  共扫描 {len(sampled)} 个 Key\n")
    print(f"  {'Key':<40} {'类型':<10} {'内存':<12} {'IdleTime(s)':<12}")
    print("  " + "-" * 80)
    for k, usage in sampled[:top_n]:
        try:
            t = r.type(k)
            idle = r.object("idletime", k) or 0
        except redis.ResponseError:
            t = idle = "-"
        print(f"  {k:<40} {t:<10} {fmt_bytes(usage):<12} {idle:<12}")

    total = sum(u for _, u in sampled)
    top_total = sum(u for _, u in sampled[:top_n])
    print(f"\n  Top {top_n} 占比：{top_total/total*100:.1f}%（{fmt_bytes(top_total)} / {fmt_bytes(total)}）")
    if top_total / total > 0.5:
        print("  ⚠ Top 10 个 Key 占总内存超过 50%，存在大 Key 风险！")


def show_idle_keys(threshold_sec: int = 0) -> None:
    section(f"Step 3: OBJECT IDLETIME 找冷 Key（阈值 = {threshold_sec}s）")

    cold = []
    for k in r.scan_iter(match="demo:mem:*", count=500):
        try:
            idle = r.object("idletime", k) or 0
            if idle >= threshold_sec:
                cold.append((k, idle))
        except redis.ResponseError:
            continue

    cold.sort(key=lambda x: -x[1])
    if not cold:
        print("  （没有冷 Key）")
        return
    print(f"  共发现冷 Key {len(cold)} 个，前 10 个：\n")
    for k, idle in cold[:10]:
        print(f"    {k:<40} idle={idle}s")
    print("\n  💡 冷 Key 处理策略：")
    print("     ① 设置 TTL 让其自动过期")
    print("     ② 用 UNLINK 异步删除")
    print("     ③ 持久化策略改为 LRU/LFU 让其自动淘汰")


def diagnosis() -> None:
    section("Step 4: 诊断结论")
    info = r.info("memory")
    used  = int(info.get("used_memory", 0))
    rss   = int(info.get("used_memory_rss", 0))
    frag  = float(info.get("mem_fragmentation_ratio", 1.0))

    print(f"  used_memory    = {fmt_bytes(used)}")
    print(f"  rss            = {fmt_bytes(rss)}")
    print(f"  碎片率          = {frag:.2f}")

    print()
    if frag > 1.5:
        print("  ❌ 碎片率 > 1.5：建议开启 activedefrag 或重启从节点")
    elif frag < 1.0:
        print("  🚨 碎片率 < 1.0：可能触发 SWAP！立即 free -m 检查内存")
    else:
        print("  ✅ 碎片率正常")

    if int(info.get("evicted_keys", 0)) > 0:
        print(f"  ⚠ 累计淘汰 {info['evicted_keys']} 个 Key：maxmemory 设置过小")


def main() -> None:
    try:
        prepare_demo_data()
        show_memory_stats()
        show_top_keys(top_n=10)
        show_idle_keys(threshold_sec=0)
        diagnosis()
    except redis.ConnectionError as e:
        print(f"❌ Redis 连接失败：{e}")
    finally:
        cleanup_demo_data()


if __name__ == "__main__":
    main()