"""
第 3 章 - 最小可用 Consumer
==================================

30 行写完一个消费者，演示：
- group.id / auto.offset.reset / enable.auto.commit 配置
- subscribe + poll 循环模型
- 处理 None / 错误消息 / EOF
- finally 中调 close() 触发同步 commit + LeaveGroup

运行：
    bash ../init.sh
    # 启动两份会自动分摊分区，关掉一份会触发 Rebalance
    python simple_consumer.py
    python simple_consumer.py
"""

from __future__ import annotations

import signal
import sys
import time

from confluent_kafka import Consumer, KafkaError

BOOTSTRAP = sys.argv[1] if len(sys.argv) > 1 else "127.0.0.1:9092"
TOPIC = "learn.03.hello"
GROUP = "ch3-simple-group"

consumer = Consumer(
    {
        "bootstrap.servers": BOOTSTRAP,
        "group.id": GROUP,
        "client.id": f"ch3-simple-consumer-{int(time.time())}",
        "auto.offset.reset": "earliest",
        "enable.auto.commit": True,
        "auto.commit.interval.ms": 5000,
        "session.timeout.ms": 10000,
        "max.poll.interval.ms": 60000,
    }
)


def on_assign(c, partitions):
    print(f"👉 被分配到分区: {[(p.topic, p.partition) for p in partitions]}")


def on_revoke(c, partitions):
    print(f"👋 被收回分区: {[(p.topic, p.partition) for p in partitions]}")


running = True


def stop(*_):
    global running
    print("\n⏹  收到退出信号，正在关闭...")
    running = False


signal.signal(signal.SIGINT, stop)
signal.signal(signal.SIGTERM, stop)


def main() -> None:
    print(f"📥 订阅 {TOPIC}（group={GROUP}）")
    consumer.subscribe([TOPIC], on_assign=on_assign, on_revoke=on_revoke)
    count = 0
    try:
        while running:
            msg = consumer.poll(timeout=1.0)
            if msg is None:
                continue
            if msg.error():
                if msg.error().code() == KafkaError._PARTITION_EOF:
                    continue
                print(f"⚠️  消费错误: {msg.error()}")
                continue
            count += 1
            print(
                f"[{count:>3}] {msg.topic()}-{msg.partition()}@{msg.offset()} "
                f"key={msg.key().decode() if msg.key() else None} "
                f"value={msg.value().decode()}"
            )
    finally:
        consumer.close()
        print(f"\n📊 共消费 {count} 条消息")


if __name__ == "__main__":
    main()