import sys, os, json sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__)))) from kafka import KafkaConsumer, KafkaProducer from shared.events import BOOTSTRAP_SERVERS, TOPIC_PAYMENTS, TOPIC_NOTIFICATIONS, make_event def run(): consumer = KafkaConsumer( TOPIC_PAYMENTS, bootstrap_servers=BOOTSTRAP_SERVERS, group_id="kitchen_service", auto_offset_reset="latest", value_deserializer=lambda v: json.loads(v.decode("utf-8")) ) producer = KafkaProducer( bootstrap_servers=BOOTSTRAP_SERVERS, key_serializer=str.encode, value_serializer=lambda v: json.dumps(v).encode(), acks="all" ) print("[KITCHEN SERVICE] Service de cuisine démarré, en attente d'évènements...") for message in consumer: event = message.value if event["type"] != "payment.confirmed": continue payload = event["payload"] order_id = payload["order_id"] restaurant_id = payload["restaurant_id"] print(f"[KITCHEN SERVICE] Paiement confirmé - démarrage de la préparation de la commande {order_id}") result = make_event("order.preparation_started", { "order_id": order_id, "customer_id": payload["customer_id"], "restaurant_id": restaurant_id }) producer.send( topic=TOPIC_NOTIFICATIONS, key=str(restaurant_id), value=result ) producer.flush() print(f"[KITCHEN SERVICE] Préparation de la commande {order_id} commencée") if __name__ == "__main__": run()