Spaces:
Running
Running
File size: 4,781 Bytes
5553f71 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 | 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()
|