#!/usr/bin/env python3
"""
Avro Producer 示例。

依赖：
    pip install confluent-kafka[avro]  # 或 pip install confluent-kafka fastavro

启动前置条件：
    Kafka:   localhost:9092
    Schema Registry: http://localhost:8081

运行方式：
    python avro_producer.py
"""

import json
import os
import time
from pathlib import Path

from confluent_kafka import Producer
from confluent_kafka.serialization import SerializationContext, MessageField, StringSerializer
from confluent_kafka.schema_registry import SchemaRegistryClient
from confluent_kafka.schema_registry.avro import AvroSerializer

BOOTSTRAP = os.getenv("BOOTSTRAP", "localhost:9092")
SR_URL = os.getenv("SR_URL", "http://localhost:8081")
TOPIC = "learn.18.users"

SCHEMA_DIR = Path(__file__).parent / "schemas"


def load_schema(name: str) -> str:
    """读取 .avsc 文件原文（保持 JSON 字符串）。"""
    return (SCHEMA_DIR / name).read_text(encoding="utf-8")


def user_to_dict(user: dict, ctx: SerializationContext) -> dict:
    """对象 → dict（这里直接传 dict，所以原样返回）。"""
    return user


def delivery_report(err, msg):
    if err is not None:
        print(f"[ERROR] {err}")
    else:
        print(
            f"[OK] topic={msg.topic()} partition={msg.partition()} "
            f"offset={msg.offset()} key={msg.key()}"
        )


def main():
    sr = SchemaRegistryClient({"url": SR_URL})

    schema_str = load_schema("user_v2.avsc")
    avro_serializer = AvroSerializer(
        schema_registry_client=sr,
        schema_str=schema_str,
        to_dict=user_to_dict,
        conf={"auto.register.schemas": True},
    )
    key_serializer = StringSerializer("utf_8")

    producer = Producer({
        "bootstrap.servers": BOOTSTRAP,
        "linger.ms": 20,
        "compression.type": "zstd",
        "enable.idempotence": True,
        "acks": "all",
    })

    sample_users = [
        {"id": 1001, "name": "Alice",  "email": "alice@example.com",  "age": 28},
        {"id": 1002, "name": "Bob",    "email": "bob@example.com",    "age": None},
        {"id": 1003, "name": "Carol",  "email": "unknown@example.com", "age": 35},
    ]

    for u in sample_users:
        key = str(u["id"])
        value_bytes = avro_serializer(
            u, SerializationContext(TOPIC, MessageField.VALUE)
        )
        producer.produce(
            topic=TOPIC,
            key=key_serializer(key, SerializationContext(TOPIC, MessageField.KEY)),
            value=value_bytes,
            on_delivery=delivery_report,
        )
        print(f"  → 序列化字节长度 = {len(value_bytes)} (含 5 字节 magic+id 头)")

    producer.flush(10)
    print("done.")


if __name__ == "__main__":
    main()