vito_stt.py

# -*- 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)
            }