hgzero/ai-python/app/services/eventhub_service.py

"""Azure Event Hub 서비스 - STT 텍스트 수신"""
import asyncio
import logging
import json
from azure.eventhub.aio import EventHubConsumerClient
from azure.eventhub.extensions.checkpointstoreblobaio import BlobCheckpointStore

from app.config import get_settings
from app.services.redis_service import RedisService

logger = logging.getLogger(__name__)
settings = get_settings()


class EventHubService:
    """Event Hub 리스너 - STT 텍스트 실시간 수신"""

    def __init__(self):
        self.client = None
        self.redis_service = RedisService()

    async def start(self):
        """Event Hub 리스닝 시작"""
        if not settings.eventhub_connection_string:
            logger.warning("Event Hub 연결 문자열이 설정되지 않음 - Event Hub 리스너 비활성화")
            return

        logger.info("Event Hub 리스너 시작")

        try:
            # Redis 연결
            await self.redis_service.connect()

            # Event Hub 클라이언트 생성
            self.client = EventHubConsumerClient.from_connection_string(
                conn_str=settings.eventhub_connection_string,
                consumer_group=settings.eventhub_consumer_group,
                eventhub_name=settings.eventhub_name,
            )

            # 이벤트 수신 시작
            async with self.client:
                await self.client.receive(
                    on_event=self.on_event,
                    on_error=self.on_error,
                    starting_position="-1",  # 최신 이벤트부터
                )

        except Exception as e:
            logger.error(f"Event Hub 리스너 오류: {e}")
        finally:
            await self.redis_service.disconnect()

    async def on_event(self, partition_context, event):
        """
        이벤트 수신 핸들러

        이벤트 형식 (STT Service에서 발행):
        {
            "eventType": "TranscriptSegmentReady",
            "meetingId": "meeting-123",
            "text": "변환된 텍스트",
            "timestamp": 1234567890000
        }
        """
        try:
            # 이벤트 데이터 파싱
            event_data = json.loads(event.body_as_str())

            event_type = event_data.get("eventType")
            meeting_id = event_data.get("meetingId")
            text = event_data.get("text")
            timestamp = event_data.get("timestamp")

            if event_type == "TranscriptSegmentReady" and meeting_id and text:
                logger.info(
                    f"STT 텍스트 수신 - meetingId: {meeting_id}, "
                    f"텍스트 길이: {len(text)}"
                )

                # Redis에 텍스트 축적 (슬라이딩 윈도우)
                await self.redis_service.add_transcript_segment(
                    meeting_id=meeting_id,
                    text=text,
                    timestamp=timestamp
                )

                logger.debug(f"Redis 저장 완료 - meetingId: {meeting_id}")

            # 체크포인트 업데이트
            await partition_context.update_checkpoint(event)

        except Exception as e:
            logger.error(f"이벤트 처리 오류: {e}", exc_info=True)

    async def on_error(self, partition_context, error):
        """에러 핸들러"""
        logger.error(
            f"Event Hub 에러 - Partition: {partition_context.partition_id}, "
            f"Error: {error}"
        )

    async def stop(self):
        """Event Hub 리스너 종료"""
        if self.client:
            await self.client.close()
            logger.info("Event Hub 리스너 종료")


# 백그라운드 태스크로 실행할 함수
async def start_eventhub_listener():
    """Event Hub 리스너 백그라운드 실행"""
    service = EventHubService()
    await service.start()