# -*- coding: utf-8 -*- """ VITO API를 사용한 음성 인식(STT) 모듈 """ import os import logging import requests import json import time # time import 추가 from dotenv import load_dotenv # 환경 변수 로드 load_dotenv() # 로거 설정 (app.py와 공유하거나 독립적으로 설정 가능) # 여기서는 독립적인 로거를 사용합니다. 필요시 app.py의 로거를 사용하도록 수정할 수 있습니다. logger = logging.getLogger("VitoSTT") # 기본 로깅 레벨 설정 (핸들러가 없으면 출력이 안될 수 있으므로 기본 핸들러 추가 고려) if not logger.hasHandlers(): handler = logging.StreamHandler() formatter = logging.Formatter('%(asctime)s - %(name)s - %(levelname)s - %(message)s') handler.setFormatter(formatter) logger.addHandler(handler) logger.setLevel(logging.INFO) # 기본 레벨 INFO로 설정 class VitoSTT: """VITO STT API 래퍼 클래스""" def __init__(self): """VITO STT 클래스 초기화""" self.client_id = os.getenv("VITO_CLIENT_ID") self.client_secret = os.getenv("VITO_CLIENT_SECRET") if not self.client_id or not self.client_secret: logger.warning("VITO API 인증 정보가 .env 파일에 설정되지 않았습니다.") logger.warning("VITO_CLIENT_ID와 VITO_CLIENT_SECRET를 확인하세요.") # 에러를 발생시키거나, 기능 사용 시점에 체크하도록 둘 수 있습니다. # 여기서는 경고만 하고 넘어갑니다. else: logger.info("VITO STT API 클라이언트 ID/Secret 로드 완료.") # API 엔드포인트 self.token_url = "https://openapi.vito.ai/v1/authenticate" self.stt_url = "https://openapi.vito.ai/v1/transcribe" # 액세스 토큰 self.access_token = None self._token_expires_at = 0 # 토큰 만료 시간 추적 (선택적 개선) def get_access_token(self): """VITO API 액세스 토큰 획득""" # 현재 시간을 가져와 토큰 만료 여부 확인 (선택적 개선) # now = time.time() # if self.access_token and now < self._token_expires_at: # logger.debug("기존 VITO API 토큰 사용") # return self.access_token if not self.client_id or not self.client_secret: logger.error("API 키가 설정되지 않아 토큰을 획득할 수 없습니다.") raise ValueError("VITO API 인증 정보가 설정되지 않았습니다.") logger.info("VITO API 액세스 토큰 요청 중...") try: response = requests.post( self.token_url, data={"client_id": self.client_id, "client_secret": self.client_secret}, timeout=10 # 타임아웃 설정 ) response.raise_for_status() # HTTP 오류 발생 시 예외 발생 result = response.json() self.access_token = result.get("access_token") expires_in = result.get("expires_in", 3600) # 만료 시간 (초), 기본값 1시간 self._token_expires_at = time.time() + expires_in - 60 # 60초 여유 if not self.access_token: logger.error("VITO API 응답에서 토큰을 찾을 수 없습니다.") raise ValueError("VITO API 토큰을 받아오지 못했습니다.") logger.info("VITO API 액세스 토큰 획득 성공") return self.access_token except requests.exceptions.Timeout: logger.error(f"VITO API 토큰 획득 시간 초과: {self.token_url}") raise TimeoutError("VITO API 토큰 획득 시간 초과") except requests.exceptions.RequestException as e: logger.error(f"VITO API 토큰 획득 실패: {e}") if hasattr(e, 'response') and e.response is not None: logger.error(f"응답 코드: {e.response.status_code}, 내용: {e.response.text}") raise ConnectionError(f"VITO API 토큰 획득 실패: {e}") def transcribe_audio(self, audio_bytes, language="ko"): """ 오디오 바이트 데이터를 텍스트로 변환 Args: audio_bytes: 오디오 파일 바이트 데이터 language: 언어 코드 (기본값: 'ko') Returns: 인식된 텍스트 또는 오류 메시지를 포함한 딕셔너리 {'success': True, 'text': '인식된 텍스트'} {'success': False, 'error': '오류 메시지', 'details': '상세 내용'} """ if not self.client_id or not self.client_secret: logger.error("API 키가 설정되지 않았습니다.") return {"success": False, "error": "API 키가 설정되지 않았습니다."} try: # 토큰 획득 또는 갱신 # (선택적 개선: 만료 시간 체크 로직 추가 시 self._token_expires_at 사용) if not self.access_token: # or time.time() >= self._token_expires_at: logger.info("VITO API 토큰 획득/갱신 시도...") self.get_access_token() headers = { "Authorization": f"Bearer {self.access_token}" } files = { "file": ("audio_file", audio_bytes) # 파일명 튜플로 전달 } # API 설정값 (필요에 따라 수정) config = { "use_multi_channel": False, "use_itn": True, # Inverse Text Normalization (숫자, 날짜 등 변환) "use_disfluency_filter": True, # 필러 (음, 아...) 제거 "use_profanity_filter": False, # 비속어 필터링 "language": language, # "type": "audio" # type 파라미터는 VITO 문서상 필수 아님 (자동 감지) } data = {"config": json.dumps(config)} logger.info(f"VITO STT API ({self.stt_url}) 요청 전송 중...") response = requests.post( self.stt_url, headers=headers, files=files, data=data, timeout=20 # 업로드 타임아웃 ) response.raise_for_status() result = response.json() job_id = result.get("id") if not job_id: logger.error("VITO API 작업 ID를 받아오지 못했습니다.") return {"success": False, "error": "VITO API 작업 ID를 받아오지 못했습니다."} logger.info(f"VITO STT 작업 ID: {job_id}, 결과 확인 시작...") # 결과 확인 URL transcript_url = f"{self.stt_url}/{job_id}" max_tries = 15 # 최대 시도 횟수 증가 wait_time = 2 # 대기 시간 증가 (초) for try_count in range(max_tries): time.sleep(wait_time) # API 부하 감소 위해 대기 logger.debug(f"결과 확인 시도 ({try_count + 1}/{max_tries}) - URL: {transcript_url}") get_response = requests.get( transcript_url, headers=headers, timeout=10 # 결과 확인 타임아웃 ) get_response.raise_for_status() result = get_response.json() status = result.get("status") logger.debug(f"현재 상태: {status}") if status == "completed": # 결과 추출 (utterances 구조 확인 필요) utterances = result.get("results", {}).get("utterances", []) if utterances: # 전체 텍스트를 하나로 합침 transcript = " ".join([seg.get("msg", "") for seg in utterances if seg.get("msg")]).strip() logger.info(f"VITO STT 인식 성공 (일부): {transcript[:50]}...") return { "success": True, "text": transcript # "raw_result": result # 필요시 전체 결과 반환 } else: logger.warning("VITO STT 완료되었으나 결과 utterances가 비어있습니다.") return {"success": True, "text": ""} # 성공이지만 텍스트 없음 elif status == "failed": error_msg = f"VITO API 변환 실패: {result.get('message', '알 수 없는 오류')}" logger.error(error_msg) return {"success": False, "error": error_msg, "details": result} elif status == "transcribing": logger.info(f"VITO API 처리 중... ({try_count + 1}/{max_tries})") else: # registered, waiting 등 다른 상태 logger.info(f"VITO API 상태 '{status}', 대기 중... ({try_count + 1}/{max_tries})") logger.error(f"VITO API 응답 타임아웃 ({max_tries * wait_time}초 초과)") return {"success": False, "error": "VITO API 응답 타임아웃"} except requests.exceptions.HTTPError as e: # 토큰 만료 오류 처리 (401 Unauthorized) if e.response.status_code == 401: logger.warning("VITO API 토큰이 만료되었거나 유효하지 않습니다. 토큰 재발급 시도...") self.access_token = None # 기존 토큰 무효화 try: # 재귀 호출 대신, 토큰 재발급 후 다시 시도하는 로직 구성 self.get_access_token() logger.info("새 토큰으로 재시도합니다.") # 재시도는 이 함수를 다시 호출하는 대신, 호출하는 쪽에서 처리하는 것이 더 안전할 수 있음 # 여기서는 한 번 더 시도하는 로직 추가 (무한 루프 방지 필요) # return self.transcribe_audio(audio_bytes, language) # 재귀 호출 방식 # --- 비재귀 방식 --- headers["Authorization"] = f"Bearer {self.access_token}" # 헤더 업데이트 # POST 요청부터 다시 시작 (코드 중복 발생 가능성 있음) # ... (POST 요청 및 결과 폴링 로직 반복) ... # 간단하게는 그냥 실패 처리하고 상위에서 재시도 유도 return {"success": False, "error": "토큰 만료 후 재시도 필요", "details": "토큰 재발급 성공"} except Exception as token_e: logger.error(f"토큰 재획득 실패: {token_e}") return {"success": False, "error": f"토큰 재획득 실패: {str(token_e)}"} else: # 401 외 다른 HTTP 오류 error_body = "" try: error_body = e.response.text except Exception: pass logger.error(f"VITO API HTTP 오류: {e.response.status_code}, 응답: {error_body}") return { "success": False, "error": f"API HTTP 오류: {e.response.status_code}", "details": error_body } except requests.exceptions.Timeout: logger.error("VITO API 요청 시간 초과") return {"success": False, "error": "API 요청 시간 초과"} except requests.exceptions.RequestException as e: logger.error(f"VITO API 요청 중 네트워크 오류 발생: {str(e)}") return {"success": False, "error": "API 요청 네트워크 오류", "details": str(e)} except Exception as e: logger.error(f"음성인식 처리 중 예상치 못한 오류 발생: {str(e)}", exc_info=True) return { "success": False, "error": "음성인식 내부 처리 실패", "details": str(e) }