"""
Ch13 配套代码 4 / 4 —— 13 条优化清单 self-check

演示：
  自动检查当前 Redis 配置 vs 13 条最佳实践
  对每一条给出「✅ 通过 / ⚠ 警告 / ❌ 不通过」+ 改进建议

  注意：部分条目（如客户端是否用连接池、是否用 Pipeline）只能在客户端
  侧检查，本脚本会在这些条目上提示「需业务自检」。
"""

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)


PASS = "✅ PASS"
WARN = "⚠  WARN"
FAIL = "❌ FAIL"
SELF = "📋 SELF"


def cfg_get(key: str):
    try:
        items = r.config_get(key)
        return items.get(key)
    except redis.ResponseError:
        return None


def parse_bytes(s: str) -> int:
    if not s or s == "0":
        return 0
    try:
        return int(s)
    except ValueError:
        return 0


def check_1_pool() -> tuple:
    return (SELF, "客户端连接池", "客户端代码自检：使用 redis.ConnectionPool / Jedis Pool / Lettuce")


def check_2_pipeline() -> tuple:
    return (SELF, "批量场景用 Pipeline / MGET / HMGET", "客户端代码自检：批量操作用 pipeline.execute()")


def check_3_lua() -> tuple:
    return (SELF, "复合逻辑用 Lua 下沉", "客户端代码自检：『读-改-写』复合逻辑用 EVAL")


def check_4_ttl() -> tuple:
    info = r.info("keyspace")
    if not info:
        return (WARN, "TTL 设置", "Keyspace 为空，无法判断")
    no_ttl_dbs = []
    for db, stats in info.items():
        if isinstance(stats, dict):
            keys = stats.get("keys", 0)
            expires = stats.get("expires", 0)
            ratio = expires / keys if keys else 0
            if keys > 100 and ratio < 0.5:
                no_ttl_dbs.append((db, keys, expires, ratio))
    if no_ttl_dbs:
        msg = "； ".join(f"{db} TTL 覆盖率 {r*100:.0f}%" for db,_,_,r in no_ttl_dbs)
        return (WARN, "TTL 设置", f"以下 db 大量 Key 无 TTL：{msg}")
    return (PASS, "TTL 设置", "Key 普遍设置了 TTL")


def check_5_listpack() -> tuple:
    e = int(cfg_get("hash-max-listpack-entries") or 128)
    v = int(cfg_get("hash-max-listpack-value") or 64)
    if e <= 128 and v <= 64:
        return (PASS, "listpack 阈值合理",
                f"hash-max-listpack-entries={e}, hash-max-listpack-value={v}")
    return (WARN, "listpack 阈值偏大",
            f"entries={e}, value={v}（建议 ≤128 / ≤64，让小 Hash 用紧凑编码）")


def check_6_keylen() -> tuple:
    sample = []
    for k in r.scan_iter(count=500):
        sample.append(len(k))
        if len(sample) >= 1000:
            break
    if not sample:
        return (WARN, "Key 长度", "未发现 Key，无法采样")
    avg = sum(sample) / len(sample)
    mx = max(sample)
    if avg > 50:
        return (WARN, "Key 长度", f"平均 Key 长度 {avg:.0f}（建议 < 50），最大 {mx}")
    return (PASS, "Key 长度", f"平均 Key 长度 {avg:.0f}，最大 {mx}")


def check_7_dangerous_cmds() -> tuple:
    danger = ["KEYS", "FLUSHALL", "FLUSHDB", "DEBUG", "CONFIG"]
    enabled = []
    for cmd in danger:
        try:
            res = r.execute_command("COMMAND", "INFO", cmd)
            if res and res[0] is not None:
                enabled.append(cmd)
        except redis.ResponseError:
            continue
    if enabled:
        return (WARN, "危险命令未禁用",
                f"以下命令仍可用：{', '.join(enabled)}（生产建议 rename-command 禁用）")
    return (PASS, "危险命令已禁用", "KEYS/FLUSHALL/DEBUG 等已禁用")


def check_8_unlink_lazy() -> tuple:
    flags = {
        "lazyfree-lazy-eviction":   cfg_get("lazyfree-lazy-eviction"),
        "lazyfree-lazy-expire":     cfg_get("lazyfree-lazy-expire"),
        "lazyfree-lazy-server-del": cfg_get("lazyfree-lazy-server-del"),
        "lazyfree-lazy-user-del":   cfg_get("lazyfree-lazy-user-del"),
    }
    off = [k for k, v in flags.items() if (v or "no").lower() != "yes"]
    if off:
        return (WARN, "lazy free 未全开",
                f"未开启：{', '.join(off)}（建议全设 yes，让删除走 BIO 异步）")
    return (PASS, "lazy free 全开", "DEL/EXPIRE/淘汰 都异步释放")


def check_9_bigkey() -> tuple:
    big = []
    for k in r.scan_iter(count=500):
        try:
            usage = r.memory_usage(k)
            if usage and usage > 100 * 1024:
                big.append((k, usage))
        except redis.ResponseError:
            continue
        if len(big) >= 5:
            break
    if big:
        msg = "； ".join(f"{k}={u//1024}KB" for k, u in big[:3])
        return (FAIL, "存在大 Key", f"≥100KB 的 Key 抽样：{msg}（应拆分）")
    return (PASS, "大 Key 检查", "抽样未发现 ≥100KB 的 Key")


def check_10_isolation() -> tuple:
    info = r.info("keyspace")
    db_count = len([k for k in info if k.startswith("db")])
    if db_count > 4:
        return (WARN, "业务隔离",
                f"使用了 {db_count} 个 DB，混用业务的可能性高（建议拆实例而非用多 DB）")
    return (SELF, "业务隔离",
            "需架构层面确认：不同业务是否使用了独立 Redis 实例")


def check_11_rdb_save() -> tuple:
    save = cfg_get("save")
    if save and save.strip():
        return (WARN, "RDB 自动保存",
                f"save = '{save}'。生产环境建议关闭自动 RDB，改 cron 错峰触发 BGSAVE")
    return (PASS, "RDB 自动保存", "已禁用，可走 cron 错峰触发")


def check_12_maxmemory() -> tuple:
    mm = parse_bytes(cfg_get("maxmemory") or "0")
    info = r.info("memory")
    used = int(info.get("used_memory", 0))
    if mm == 0:
        return (FAIL, "maxmemory 未限制",
                "maxmemory=0 危险！请设置上限（建议留 30% 内存给 fork）")
    if used / mm > 0.7:
        return (WARN, "maxmemory 利用率高",
                f"已用 {used/mm*100:.0f}% / {mm//1024//1024}MB（>70% 进入告警区）")
    return (PASS, "maxmemory 设置合理",
            f"已用 {used/mm*100:.0f}% / {mm//1024//1024}MB")


def check_13_linux() -> tuple:
    notes = []
    try:
        with open("/proc/sys/vm/overcommit_memory") as f:
            v = f.read().strip()
            if v != "1":
                notes.append(f"overcommit_memory={v}（建议 1）")
    except OSError:
        notes.append("无法读取 /proc/sys/vm/overcommit_memory")
    try:
        with open("/sys/kernel/mm/transparent_hugepage/enabled") as f:
            v = f.read().strip()
            if "[never]" not in v:
                notes.append(f"THP={v}（建议 never）")
    except OSError:
        notes.append("无法读取 transparent_hugepage")
    try:
        with open("/proc/sys/vm/swappiness") as f:
            v = int(f.read().strip())
            if v > 10:
                notes.append(f"swappiness={v}（建议 ≤10）")
    except OSError:
        notes.append("无法读取 swappiness")

    if notes:
        return (WARN, "Linux 内核参数", "； ".join(notes))
    return (PASS, "Linux 内核参数", "overcommit/THP/swappiness 配置 OK")


CHECKS = [
    ("1.  连接池",            check_1_pool),
    ("2.  Pipeline / MGET",   check_2_pipeline),
    ("3.  Lua 下沉",          check_3_lua),
    ("4.  TTL 设置",          check_4_ttl),
    ("5.  listpack 阈值",     check_5_listpack),
    ("6.  Key 长度",          check_6_keylen),
    ("7.  禁用危险命令",      check_7_dangerous_cmds),
    ("8.  UNLINK / lazy free", check_8_unlink_lazy),
    ("9.  大 Key 检查",       check_9_bigkey),
    ("10. 业务隔离",          check_10_isolation),
    ("11. 错峰主从同步",      check_11_rdb_save),
    ("12. maxmemory 留余量",  check_12_maxmemory),
    ("13. Linux 内核调优",    check_13_linux),
]


def main() -> None:
    try:
        section("Redis 13 条优化清单 self-check")
        results = []
        for name, fn in CHECKS:
            try:
                status, _, msg = fn()
            except Exception as e:
                status, msg = FAIL, f"check error: {e}"
            results.append((name, status, msg))
            print(f"\n  [{status}] {name}")
            print(f"          {msg}")

        section("最终诊断报告")
        n_pass = sum(1 for _, s, _ in results if s == PASS)
        n_warn = sum(1 for _, s, _ in results if s == WARN)
        n_fail = sum(1 for _, s, _ in results if s == FAIL)
        n_self = sum(1 for _, s, _ in results if s == SELF)
        print(f"  ✅ 通过：{n_pass}  ⚠ 警告：{n_warn}  ❌ 不通过：{n_fail}  📋 需自检：{n_self}")

        score = (n_pass + n_self * 0.5) / len(results) * 100
        print(f"  自动评分：{score:.0f} / 100\n")

        critical = [n for n, s, _ in results if s == FAIL]
        if critical:
            print("  🚨 必须立即处理：")
            for n in critical:
                print(f"     - {n}")
        warn_items = [n for n, s, _ in results if s == WARN]
        if warn_items:
            print("\n  ⚠ 建议改进：")
            for n in warn_items:
                print(f"     - {n}")
        if not critical and not warn_items:
            print("  🎉 全部通过，配置堪称完美！")
    except redis.ConnectionError as e:
        print(f"❌ Redis 连接失败：{e}")


if __name__ == "__main__":
    main()