jeongsoo/RAG6_DockerLLM_API_Service

1# -*- coding: utf-8 -*-2"""3VITO API를 사용한 음성 인식(STT) 모듈4"""5 6import os7import logging8import requests9import json10import time11from dotenv import load_dotenv12 13# 환경 변수 로드14load_dotenv()15 16# 로거 설정17logger = logging.getLogger("VitoSTT")18# 기본 로깅 레벨 설정 (핸들러가 없으면 출력이 안될 수 있으므로 기본 핸들러 추가 고려)19if not logger.hasHandlers():20    handler = logging.StreamHandler()21    formatter = logging.Formatter('%(asctime)s - %(name)s - %(levelname)s - %(message)s')22    handler.setFormatter(formatter)23    logger.addHandler(handler)24    logger.setLevel(logging.INFO) # 기본 레벨 INFO로 설정25 26class VitoSTT:27    """VITO STT API 래퍼 클래스"""28 29    def __init__(self):30        """VITO STT 클래스 초기화"""31        self.client_id = os.getenv("VITO_CLIENT_ID")32        self.client_secret = os.getenv("VITO_CLIENT_SECRET")33 34        if not self.client_id or not self.client_secret:35            logger.warning("VITO API 인증 정보가 .env 파일에 설정되지 않았습니다.")36            logger.warning("VITO_CLIENT_ID와 VITO_CLIENT_SECRET를 확인하세요.")37            # 에러를 발생시키거나, 기능 사용 시점에 체크하도록 둘 수 있습니다.38            # 여기서는 경고만 하고 넘어갑니다.39        else:40            logger.info("VITO STT API 클라이언트 ID/Secret 로드 완료.")41 42        # API 엔드포인트43        self.token_url = "https://openapi.vito.ai/v1/authenticate"44        self.stt_url = "https://openapi.vito.ai/v1/transcribe"45 46        # 액세스 토큰47        self.access_token = None48        self._token_expires_at = 0 # 토큰 만료 시간 추적 (선택적 개선)49 50    def get_access_token(self):51        """VITO API 액세스 토큰 획득"""52        # 현재 시간을 가져와 토큰 만료 여부 확인 (선택적 개선)53        now = time.time()54        if self.access_token and now < self._token_expires_at:55            logger.debug("기존 VITO API 토큰 사용")56            return self.access_token57 58        if not self.client_id or not self.client_secret:59            logger.error("API 키가 설정되지 않아 토큰을 획득할 수 없습니다.")60            raise ValueError("VITO API 인증 정보가 설정되지 않았습니다.")61 62        logger.info("VITO API 액세스 토큰 요청 중...")63        try:64            response = requests.post(65                self.token_url,66                data={"client_id": self.client_id, "client_secret": self.client_secret},67                timeout=10 # 타임아웃 설정68            )69            response.raise_for_status() # HTTP 오류 발생 시 예외 발생70 71            result = response.json()72            self.access_token = result.get("access_token")73            expires_in = result.get("expires_in", 3600) # 만료 시간 (초), 기본값 1시간74            self._token_expires_at = time.time() + expires_in - 60 # 60초 여유75 76            if not self.access_token:77                logger.error("VITO API 응답에서 토큰을 찾을 수 없습니다.")78                raise ValueError("VITO API 토큰을 받아오지 못했습니다.")79 80            logger.info("VITO API 액세스 토큰 획득 성공")81            return self.access_token82        except requests.exceptions.Timeout:83            logger.error(f"VITO API 토큰 획득 시간 초과: {self.token_url}")84            raise TimeoutError("VITO API 토큰 획득 시간 초과")85        except requests.exceptions.RequestException as e:86            logger.error(f"VITO API 토큰 획득 실패: {e}")87            if hasattr(e, 'response') and e.response is not None:88                 logger.error(f"응답 코드: {e.response.status_code}, 내용: {e.response.text}")89            raise ConnectionError(f"VITO API 토큰 획득 실패: {e}")90 91 92    def transcribe_audio(self, audio_bytes, language="ko"):93        """94        오디오 바이트 데이터를 텍스트로 변환95 96        Args:97            audio_bytes: 오디오 파일 바이트 데이터98            language: 언어 코드 (기본값: 'ko')99 100        Returns:101            인식된 텍스트 또는 오류 메시지를 포함한 딕셔너리102            {'success': True, 'text': '인식된 텍스트'}103            {'success': False, 'error': '오류 메시지', 'details': '상세 내용'}104        """105        if not self.client_id or not self.client_secret:106            logger.error("API 키가 설정되지 않았습니다.")107            return {"success": False, "error": "API 키가 설정되지 않았습니다."}108 109        try:110            # 토큰 획득 또는 갱신111            # (선택적 개선: 만료 시간 체크 로직 추가 시 self._token_expires_at 사용)112            if not self.access_token or time.time() >= self._token_expires_at:113                logger.info("VITO API 토큰 획득/갱신 시도...")114                self.get_access_token()115 116            headers = {117                "Authorization": f"Bearer {self.access_token}"118            }119 120            files = {121                "file": ("audio_file", audio_bytes) # 파일명 튜플로 전달122            }123 124            # API 설정값 (필요에 따라 수정)125            config = {126                "use_multi_channel": False,127                "use_itn": True, # Inverse Text Normalization (숫자, 날짜 등 변환)128                "use_disfluency_filter": True, # 필러 (음, 아...) 제거129                "use_profanity_filter": False, # 비속어 필터링130                "language": language,131                # "type": "audio" # type 파라미터는 VITO 문서상 필수 아님 (자동 감지)132            }133            data = {"config": json.dumps(config)}134 135            logger.info(f"VITO STT API ({self.stt_url}) 요청 전송 중...")136            response = requests.post(137                self.stt_url,138                headers=headers,139                files=files,140                data=data,141                timeout=20 # 업로드 타임아웃142            )143            response.raise_for_status()144 145            result = response.json()146            job_id = result.get("id")147 148            if not job_id:149                logger.error("VITO API 작업 ID를 받아오지 못했습니다.")150                return {"success": False, "error": "VITO API 작업 ID를 받아오지 못했습니다."}151 152            logger.info(f"VITO STT 작업 ID: {job_id}, 결과 확인 시작...")153 154            # 결과 확인 URL155            transcript_url = f"{self.stt_url}/{job_id}"156            max_tries = 15 # 최대 시도 횟수 증가157            wait_time = 2 # 대기 시간 증가 (초)158 159            for try_count in range(max_tries):160                time.sleep(wait_time) # API 부하 감소 위해 대기161                logger.debug(f"결과 확인 시도 ({try_count + 1}/{max_tries}) - URL: {transcript_url}")162                get_response = requests.get(163                    transcript_url,164                    headers=headers,165                    timeout=10 # 결과 확인 타임아웃166                )167                get_response.raise_for_status()168 169                result = get_response.json()170                status = result.get("status")171                logger.debug(f"현재 상태: {status}")172 173                if status == "completed":174                    # 결과 추출 (utterances 구조 확인 필요)175                    utterances = result.get("results", {}).get("utterances", [])176                    if utterances:177                         # 전체 텍스트를 하나로 합침178                        transcript = " ".join([seg.get("msg", "") for seg in utterances if seg.get("msg")]).strip()179                        logger.info(f"VITO STT 인식 성공 (일부): {transcript[:50]}...")180                        return {181                            "success": True,182                            "text": transcript183                            # "raw_result": result # 필요시 전체 결과 반환184                        }185                    else:186                         logger.warning("VITO STT 완료되었으나 결과 utterances가 비어있습니다.")187                         return {"success": True, "text": ""} # 성공이지만 텍스트 없음188 189                elif status == "failed":190                    error_msg = f"VITO API 변환 실패: {result.get('message', '알 수 없는 오류')}"191                    logger.error(error_msg)192                    return {"success": False, "error": error_msg, "details": result}193 194                elif status == "transcribing":195                    logger.info(f"VITO API 처리 중... ({try_count + 1}/{max_tries})")196                else: # registered, waiting 등 다른 상태197                    logger.info(f"VITO API 상태 '{status}', 대기 중... ({try_count + 1}/{max_tries})")198 199 200            logger.error(f"VITO API 응답 타임아웃 ({max_tries * wait_time}초 초과)")201            return {"success": False, "error": "VITO API 응답 타임아웃"}202 203        except requests.exceptions.HTTPError as e:204            # 토큰 만료 오류 처리 (401 Unauthorized)205            if e.response.status_code == 401:206                logger.warning("VITO API 토큰이 만료되었거나 유효하지 않습니다. 토큰 재발급 시도...")207                self.access_token = None # 기존 토큰 무효화208                try:209                    # 재귀 호출 대신, 토큰 재발급 후 다시 시도하는 로직 구성210                    self.get_access_token()211                    logger.info("새 토큰으로 재시도합니다.")212                    # 재시도는 이 함수를 다시 호출하는 대신, 호출하는 쪽에서 처리하는 것이 더 안전할 수 있음213                    return self.transcribe_audio(audio_bytes, language) # 재귀 호출 방식214 215                except Exception as token_e:216                    logger.error(f"토큰 재획득 실패: {token_e}")217                    return {"success": False, "error": f"토큰 재획득 실패: {str(token_e)}"}218 219            else:220                # 401 외 다른 HTTP 오류221                error_body = ""222                try:223                    error_body = e.response.text224                except Exception:225                    pass226                logger.error(f"VITO API HTTP 오류: {e.response.status_code}, 응답: {error_body}")227                return {228                    "success": False,229                    "error": f"API HTTP 오류: {e.response.status_code}",230                    "details": error_body231                }232 233        except requests.exceptions.Timeout:234            logger.error("VITO API 요청 시간 초과")235            return {"success": False, "error": "API 요청 시간 초과"}236        except requests.exceptions.RequestException as e:237            logger.error(f"VITO API 요청 중 네트워크 오류 발생: {str(e)}")238            return {"success": False, "error": "API 요청 네트워크 오류", "details": str(e)}239        except Exception as e:240            logger.error(f"음성인식 처리 중 예상치 못한 오류 발생: {str(e)}", exc_info=True)241            return {242                "success": False,243                "error": "음성인식 내부 처리 실패",244                "details": str(e)245            }246