"""
第 3 章 - AdminClient 全家桶演示
==================================

本脚本依次演示：
1. 用 AdminClient 创建 Topic（带 Topic 级配置）
2. 列出当前所有 Topic + 集群元数据
3. describe_configs 看 Topic 的全量配置
4. 灌入一些消息，再用 list_consumer_group_offsets 看 Lag
5. alter_consumer_group_offsets 重置 Offset 到 earliest
6. 删除 Topic 收尾

依赖：
    pip install confluent-kafka==2.4.0

运行：
    python admin_demo.py [bootstrap]
    默认 bootstrap = 127.0.0.1:9092
"""

from __future__ import annotations

import sys
import time
import uuid
from typing import Iterable

from confluent_kafka import Consumer, Producer, TopicPartition
from confluent_kafka.admin import (
    AdminClient,
    ConfigResource,
    ConsumerGroupTopicPartitions,
    NewTopic,
)

BOOTSTRAP = sys.argv[1] if len(sys.argv) > 1 else "127.0.0.1:9092"
TOPIC = "learn.03.admin_demo"
GROUP = f"admin-demo-grp-{uuid.uuid4().hex[:6]}"


def banner(title: str) -> None:
    print("\n" + "=" * 60)
    print(f"  {title}")
    print("=" * 60)


def safe_run(name: str, futures: dict) -> None:
    for k, f in futures.items():
        try:
            f.result()
            print(f"  ✅ {name}: {k}")
        except Exception as e:
            print(f"  ❌ {name}: {k} -> {e}")


def main() -> None:
    admin = AdminClient({"bootstrap.servers": BOOTSTRAP})

    banner("1) create_topics")
    new_topic = NewTopic(
        topic=TOPIC,
        num_partitions=3,
        replication_factor=1,
        config={
            "retention.ms": "3600000",
            "segment.bytes": "33554432",
            "cleanup.policy": "delete",
        },
    )
    safe_run("create", admin.create_topics([new_topic], request_timeout=10))
    time.sleep(1)

    banner("2) list_topics + 集群元数据")
    md = admin.list_topics(timeout=10)
    print(f"  Cluster ID         : {md.cluster_id}")
    print(f"  Controller Broker  : {md.controller_id}")
    print(f"  Brokers            :")
    for b in md.brokers.values():
        print(f"    - id={b.id} host={b.host}:{b.port}")
    print(f"  Topics ({len(md.topics)}):")
    for t in sorted(md.topics.keys()):
        if t.startswith("__"):
            continue
        partitions = md.topics[t].partitions
        print(f"    - {t} (partitions={len(partitions)})")

    banner("3) describe_configs（看 Topic 全量配置）")
    res = ConfigResource(ConfigResource.Type.TOPIC, TOPIC)
    fs = admin.describe_configs([res])
    cfgs = list(fs.values())[0].result()
    interesting = [
        "cleanup.policy",
        "retention.ms",
        "segment.bytes",
        "min.insync.replicas",
        "compression.type",
        "max.message.bytes",
    ]
    for k in interesting:
        if k in cfgs:
            v = cfgs[k]
            print(f"  {k:24s} = {v.value}  (source={v.source.name})")

    banner(f"4) 灌入 30 条消息 + 用消费组 {GROUP} 消费一半")
    p = Producer({"bootstrap.servers": BOOTSTRAP, "linger.ms": 5})
    for i in range(30):
        p.produce(TOPIC, key=f"k{i % 5}".encode(), value=f"msg-{i}".encode())
    p.flush(10)
    print("  ✅ 30 条消息已发出")

    c = Consumer(
        {
            "bootstrap.servers": BOOTSTRAP,
            "group.id": GROUP,
            "auto.offset.reset": "earliest",
            "enable.auto.commit": False,
        }
    )
    c.subscribe([TOPIC])
    consumed = 0
    deadline = time.time() + 10
    while consumed < 15 and time.time() < deadline:
        msg = c.poll(1.0)
        if msg is None or msg.error():
            continue
        consumed += 1
    c.commit(asynchronous=False)
    print(f"  ✅ 已消费并提交 {consumed} 条")

    banner("5) list_consumer_group_offsets（查 Lag）")
    req = ConsumerGroupTopicPartitions(GROUP)
    fs = admin.list_consumer_group_offsets([req])
    res = list(fs.values())[0].result()

    md = admin.list_topics(TOPIC, timeout=5)
    end_offsets = {}
    for part_id in md.topics[TOPIC].partitions.keys():
        low, high = c.get_watermark_offsets(TopicPartition(TOPIC, part_id), timeout=5)
        end_offsets[part_id] = high

    print(f"  Group = {GROUP}")
    print(f"  {'Partition':<10} {'Committed':<12} {'End':<12} {'Lag':<6}")
    total_lag = 0
    for tp in res.topic_partitions:
        committed = tp.offset
        end = end_offsets.get(tp.partition, 0)
        lag = max(0, end - committed)
        total_lag += lag
        print(f"  {tp.partition:<10} {committed:<12} {end:<12} {lag:<6}")
    print(f"  TOTAL LAG = {total_lag}")

    c.close()

    banner("6) alter_consumer_group_offsets（重置到 earliest）")
    new_tps = [TopicPartition(TOPIC, p, 0) for p in md.topics[TOPIC].partitions.keys()]
    req = ConsumerGroupTopicPartitions(GROUP, new_tps)
    fs = admin.alter_consumer_group_offsets([req])
    safe_run("reset", {GROUP: list(fs.values())[0]})

    banner("7) 删除 Topic 收尾")
    safe_run("delete", admin.delete_topics([TOPIC], operation_timeout=10))

    print("\n🎉 admin_demo 演示完成！")


if __name__ == "__main__":
    main()