#!/usr/bin/env python3
"""
inspect_metadata.py — 用 AdminClient 探测 Kafka 集群（KRaft 模式）的元数据

打印内容：
    1) cluster_id（整个集群的全局 UUID，KRaft 时代由 kafka-storage.sh 生成）
    2) Active Controller 节点 ID（即 KRaft Quorum 的 Raft Leader）
    3) 所有 Broker 列表（host:port + rack）
    4) 所有 Topic + 分区数 + 副本布局
    5) 默认配置的几个关键值（动态查询 broker config）

用法：
    pip install confluent-kafka
    python inspect_metadata.py
    python inspect_metadata.py --bootstrap localhost:9092
"""

from __future__ import annotations

import argparse
import sys

from confluent_kafka.admin import (
    AdminClient,
    ConfigResource,
    ResourceType,
)


def parse_args():
    p = argparse.ArgumentParser()
    p.add_argument("--bootstrap", default="localhost:9092",
                   help="bootstrap.servers")
    p.add_argument("--with-topics", action="store_true",
                   help="额外打印每个 Topic 的分区/副本布局")
    p.add_argument("--with-broker-configs", action="store_true",
                   help="额外打印每个 Broker 的 KRaft 关键配置")
    return p.parse_args()


def section(title):
    print()
    print("=" * 78)
    print(f"  {title}")
    print("=" * 78)


def main():
    args = parse_args()
    admin = AdminClient({"bootstrap.servers": args.bootstrap})

    md = admin.list_topics(timeout=15)

    section("CLUSTER")
    print(f"cluster_id          : {md.cluster_id}")
    print(f"controller broker_id: {md.controller_id}    "
          f"← 在 KRaft 集群里这是 Active Controller 在 Quorum 内的 node_id")
    print(f"alive broker count  : {len(md.brokers)}")

    section("BROKERS")
    print(f"{'id':<6}{'host':<30}{'port':<8}{'rack':<10}")
    print("-" * 60)
    for bid in sorted(md.brokers.keys()):
        b = md.brokers[bid]
        print(f"{bid:<6}{b.host:<30}{b.port:<8}{(b.rack or '-'):<10}")

    if args.with_topics:
        section("TOPICS")
        for tname in sorted(md.topics.keys()):
            t = md.topics[tname]
            if tname.startswith("__"):
                marker = "  [internal]"
            else:
                marker = ""
            print(f"\n● {tname}{marker}")
            for pid in sorted(t.partitions.keys()):
                p = t.partitions[pid]
                print(f"    P{pid}  Leader={p.leader}  "
                      f"Replicas={list(p.replicas)}  ISR={list(p.isrs)}")

    if args.with_broker_configs:
        section("BROKER CONFIGS (KRaft 关键项)")
        keys_of_interest = [
            "process.roles",
            "node.id",
            "controller.quorum.voters",
            "controller.listener.names",
            "inter.broker.listener.name",
            "metadata.log.segment.bytes",
            "min.insync.replicas",
        ]
        for bid in sorted(md.brokers.keys()):
            res = ConfigResource(ResourceType.BROKER, str(bid))
            fut = admin.describe_configs([res])[res]
            try:
                conf = fut.result(timeout=10)
            except Exception as e:
                print(f"\n● Broker {bid}: 拉取配置失败：{e}")
                continue
            print(f"\n● Broker {bid}")
            for k in keys_of_interest:
                if k in conf:
                    v = conf[k]
                    src = v.source.name if hasattr(v.source, 'name') else v.source
                    print(f"    {k:<35} = {v.value}  ({src})")

    section("DONE")
    print("提示：")
    print("  * 在 KRaft 模式下，controller_id 是 Active Controller 的 node_id；")
    print("    要看完整 Quorum 状态用 kafka-metadata-quorum.sh describe --status。")
    print("  * Topic 名以 __ 开头是 Kafka 内部 Topic，包括：")
    print("      __consumer_offsets    （消费组 offset）")
    print("      __transaction_state   （事务协调）")
    print("      __cluster_metadata    （KRaft 元数据日志，仅 Quorum 节点能看到分区数据）")


if __name__ == "__main__":
    try:
        main()
    except KeyboardInterrupt:
        sys.exit(0)