from kafka import KafkaProducer, KafkaConsumer
from kafka.admin import KafkaAdminClient, NewTopic
from typing import List, Dict, Any
import json

class KafkaConfig:
    def __init__(
        self,
        bootstrap_servers: str = 'localhost:9092',
        client_id: str = 'python-kafka'
    ):
        self.bootstrap_servers = bootstrap_servers
        self.client_id = client_id
    
    def create_admin_client(self) -> KafkaAdminClient:
        """Admin 클라이언트 생성"""
        return KafkaAdminClient(
            bootstrap_servers=self.bootstrap_servers,
            client_id=self.client_id
        )
    
    def create_topic(self, topic_name: str, num_partitions: int = 1):
        """토픽 생성"""
        admin_client = self.create_admin_client()
        topic = NewTopic(
            name=topic_name,
            num_partitions=num_partitions,
            replication_factor=1
        )
        admin_client.create_topics([topic])

class KafkaMessageProducer:
    def __init__(self, config: KafkaConfig):
        self.producer = KafkaProducer(
            bootstrap_servers=config.bootstrap_servers,
            value_serializer=lambda v: json.dumps(v).encode('utf-8'),
            key_serializer=lambda k: k.encode('utf-8') if k else None
        )
    
    def send_message(
        self,
        topic: str,
        value: Dict,
        key: str = None,
        partition: int = None
    ):
        """메시지 전송"""
        future = self.producer.send(
            topic=topic,
            value=value,
            key=key,
            partition=partition
        )
        return future.get(timeout=10)

class KafkaMessageConsumer:
    def __init__(
        self,
        config: KafkaConfig,
        topics: List[str],
        group_id: str
    ):
        self.consumer = KafkaConsumer(
            *topics,
            bootstrap_servers=config.bootstrap_servers,
            group_id=group_id,
            auto_offset_reset='earliest',
            enable_auto_commit=True,
            value_deserializer=lambda x: json.loads(x.decode('utf-8'))
        )
    
    def consume_messages(self, handler):
        """메시지 소비"""
        try:
            for message in self.consumer:
                handler(message)
        finally:
            self.close()

from typing import List, Callable
import time

class KafkaBatchProcessor:
    def __init__(
        self,
        producer: KafkaMessageProducer,
        batch_size: int = 100,
        flush_interval: int = 30
    ):
        self.producer = producer
        self.batch_size = batch_size
        self.flush_interval = flush_interval
        self.messages = []
        self.last_flush = time.time()
    
    def add_message(self, topic: str, message: Dict):
        """메시지 배치에 추가"""
        self.messages.append((topic, message))
        
        if len(self.messages) >= self.batch_size:
            self.flush_batch()

class KafkaStreamProcessor:
    def __init__(
        self,
        input_topic: str,
        output_topic: str,
        config: KafkaConfig,
        group_id: str
    ):
        self.producer = KafkaMessageProducer(config)
        self.consumer = KafkaMessageConsumer(
            config,
            [input_topic],
            group_id
        )
    
    def process_stream(self, transform_func):
        """스트림 처리"""
        def message_handler(message):
            transformed_value = transform_func(message.value)
            self.producer.send_message(
                self.output_topic,
                transformed_value
            )
        
        self.consumer.consume_messages(message_handler)