Spaces:
Running
Running
| import asyncio | |
| import json | |
| import logging | |
| import os | |
| import signal | |
| from confluent_kafka import Consumer, KafkaError | |
| from src.api.routers.triage import _run_triage | |
| from src.db.repository import TicketRepository, TicketRecord | |
| # Configure logging | |
| logging.basicConfig(level=logging.INFO, format="%(asctime)s [%(levelname)s] %(name)s: %(message)s") | |
| logger = logging.getLogger("customercore.triage_consumer") | |
| BROKER = os.environ.get("REDPANDA_BROKER", "localhost:9092") | |
| TOPIC = os.environ.get("REDPANDA_TICKET_TOPIC", "support-tickets") | |
| GROUP_ID = os.environ.get("REDPANDA_TRIAL_GROUP", "triage-consumer-group") | |
| running = True | |
| def signal_handler(sig, frame): | |
| global running | |
| logger.info("Graceful shutdown triggered...") | |
| running = False | |
| # Listen to terminate signals | |
| signal.signal(signal.SIGINT, signal_handler) | |
| signal.signal(signal.SIGTERM, signal_handler) | |
| async def consume_loop(): | |
| consumer = Consumer({ | |
| "bootstrap.servers": BROKER, | |
| "group.id": GROUP_ID, | |
| "auto.offset.reset": "earliest", | |
| "enable.auto.commit": True, | |
| }) | |
| consumer.subscribe([TOPIC]) | |
| logger.info(f"Subscribed to topic '{TOPIC}' on broker '{BROKER}' (group: '{GROUP_ID}')") | |
| try: | |
| while running: | |
| # Poll for new messages using an executor to avoid blocking the asyncio event loop | |
| msg = await asyncio.to_thread(consumer.poll, 1.0) | |
| if msg is None: | |
| continue | |
| if msg.error(): | |
| if msg.error().code() != KafkaError._PARTITION_EOF: | |
| logger.error(f"Kafka error: {msg.error()}") | |
| continue | |
| try: | |
| payload = msg.value().decode("utf-8") | |
| event = json.loads(payload) | |
| logger.info(f"Received ticket event key={msg.key().decode('utf-8') if msg.key() else 'None'}") | |
| # Extract fields from the event JSON | |
| tenant_id = event.get("tenant_id") | |
| ticket_id = event.get("ticket_id") or event.get("event_id") | |
| body = event.get("body") or event.get("subject") or "" | |
| customer_id = event.get("customer_id") or "unknown" | |
| customer_tier = event.get("customer_tier") or "free" | |
| channel = event.get("channel") or "web" | |
| if not tenant_id or not ticket_id: | |
| logger.warning(f"Skipping event missing critical fields: tenant_id={tenant_id}, ticket_id={ticket_id}") | |
| continue | |
| # Ensure the ticket exists in the Supabase db (or the in-memory store in test environment) | |
| repo = TicketRepository(tenant_id) | |
| existing = await repo.get(ticket_id) | |
| if not existing: | |
| logger.info(f"Creating new ticket {ticket_id} in database...") | |
| record = TicketRecord( | |
| id=ticket_id, | |
| tenant_id=tenant_id, | |
| customer_id=customer_id, | |
| channel=channel, | |
| raw_text=body, | |
| status="pending", | |
| ) | |
| await repo.create(record) | |
| await repo.write_audit( | |
| actor="system", | |
| action="ticket.created", | |
| ticket_id=ticket_id, | |
| details={"channel": channel, "customer_tier": customer_tier, "via": "streaming_triage_consumer"}, | |
| ) | |
| # Run the triage agent asynchronously to avoid blocking the message consumer thread | |
| asyncio.create_task( | |
| _run_triage( | |
| ticket_id=ticket_id, | |
| text=body, | |
| customer_id=customer_id, | |
| tenant_id=tenant_id, | |
| customer_tier=customer_tier, | |
| channel=channel, | |
| ) | |
| ) | |
| logger.info(f"Dispatched async triage worker task for ticket {ticket_id}") | |
| except Exception as exc: | |
| logger.exception(f"Error processing Redpanda message: {exc}") | |
| finally: | |
| consumer.close() | |
| logger.info("Redpanda consumer closed gracefully.") | |
| def main(): | |
| logger.info("=" * 60) | |
| logger.info("CustomerCore Real-Time Streaming Triage Consumer") | |
| logger.info("Press Ctrl+C to terminate...") | |
| logger.info("=" * 60) | |
| try: | |
| asyncio.run(consume_loop()) | |
| except KeyboardInterrupt: | |
| logger.info("Shutdown complete.") | |
| if __name__ == "__main__": | |
| main() | |