import os import json import sys from vertexai.generative_models import GenerativeModel class VertexAITranslator: def __init__(self, logger): """ VertexAITranslator 클래스 초기화 메서드. :param logger: 로깅을 위한 로거 객체. """ self.logger = logger key_path = self.get_key_path('leensoo1nt.json') # GOOGLE_APPLICATION_CREDENTIALS 환경 변수 설정 self.logger.debug(f"GOOGLE_APPLICATION_CREDENTIALS 환경 변수를 설정: {key_path}") os.environ['GOOGLE_APPLICATION_CREDENTIALS'] = key_path # GenerativeModel 객체 초기화 self.logger.debug("Vertex AI 모델 초기화 중...") self.model = GenerativeModel("gemini-1.5-flash-001") # prompt.json 파일 불러오기 self.logger.debug("prompt.json 파일을 불러옵니다.") self.prompt_data = self.load_prompt() self.allowed_chars = self.prompt_data['allowed_special_chars'] self.replacements = self.prompt_data['special_char_replacements'] def load_prompt(self): """ prompt.json 파일을 읽어와 파싱하는 메서드. :return: 파싱된 JSON 데이터. """ try: # cx_Freeze로 패키징된 경우 실행 파일의 경로로 설정 if getattr(sys, 'frozen', False): prompt_path = os.path.join(os.path.dirname(sys.executable), 'prompt.json') else: # 일반 Python 실행 환경일 경우 prompt_path = os.path.join(os.path.dirname(__file__), 'prompt.json') self.logger.debug(f"프롬프트 파일 경로: {prompt_path}") with open(prompt_path, 'r', encoding='utf-8') as file: prompt_data = json.load(file) self.logger.debug("prompt.json 파일이 성공적으로 로드되었습니다.") return prompt_data except FileNotFoundError as e: self.logger.error(f"prompt.json 파일을 찾을 수 없습니다: {e}", exc_info=True) raise e except json.JSONDecodeError as e: self.logger.error(f"prompt.json 파일 파싱 중 오류 발생: {e}", exc_info=True) raise e def get_key_path(self, key_path): # cx_Freeze로 패키징된 경우 실행 파일 경로로 설정 if getattr(sys, 'frozen', False): return os.path.join(os.path.dirname(sys.executable), key_path) else: # 일반 Python 실행 환경일 경우 return os.path.join(os.path.dirname(__file__), key_path) def clean_special_chars(self, text): """ 텍스트에서 허용되지 않는 특수 문자를 제거하고, 필요한 특수 문자를 대체하는 메서드. :param text: 입력 텍스트. :return: 정리된 텍스트. """ self.logger.debug(f"텍스트에서 특수 문자를 정리 중: {text}") cleaned_text = [] for char in text: if char in self.replacements: cleaned_text.append(self.replacements[char]) # 대체 문자 추가 self.logger.debug(f"문자 '{char}'를 대체 문자로 변경: {self.replacements[char]}") elif char not in self.allowed_chars and not char.isalnum() and not char.isspace(): # self.logger.debug(f"허용되지 않은 문자 제거: {char}") continue # 특수 문자 제거 else: cleaned_text.append(char) # 허용된 문자 추가 # self.logger.debug(f"허용된 문자 추가: {char}") cleaned_text_str = ''.join(cleaned_text) self.logger.debug(f"정리된 텍스트: {cleaned_text_str}") return cleaned_text_str # async def translate_options(self, original_data): # """ # 주어진 옵션 데이터를 Vertex AI 모델을 통해 번역하는 메서드. # :param original_data: 원본 옵션 데이터 (dict 형태). # :return: 번역된 옵션명 (파이썬의 dict 형태). # """ # self.logger.debug(f"옵션 데이터를 번역 중: {original_data}") # # 데이터 정리 # cleaned_data = {key: self.clean_special_chars(value) for key, value in original_data.items()} # self.logger.debug(f"정리된 옵션 데이터: {cleaned_data}") # # 원본 데이터를 프롬프트 템플릿에 넣는다. # option_prompt_template = self.prompt_data['option_prompt_template'] # prompt = option_prompt_template.format(options=json.dumps(cleaned_data, ensure_ascii=False)) # self.logger.debug(f"생성된 프롬프트: {prompt}") # # Vertex AI 모델에 프롬프트 전달하여 응답 받기 # self.logger.debug("Vertex AI 모델에 프롬프트를 전달하여 응답을 기다리는 중...") # response = self.model.generate_content(prompt) # self.logger.debug(f"모델 응답: {response.text}") # # 응답 데이터에서 JSON 형식 추출 # start = response.text.find('{') # end = response.text.rfind('}') + 1 # if start != -1 and end != -1: # json_text = response.text[start:end] # self.logger.debug(f"응답에서 추출된 JSON 텍스트: {json_text}") # try: # translated_data = json.loads(json_text) # self.logger.debug(f"번역된 데이터: {translated_data}") # except json.JSONDecodeError as e: # self.logger.error(f"응답 데이터를 파싱하는 중 오류 발생: {e}", exc_info=True) # raise ValueError(f"응답 데이터를 파이썬의 딕셔너리로 파싱하는 중 오류 발생: {e}", exc_info=True) # else: # self.logger.error("응답 데이터에서 유효한 JSON 형식을 찾을 수 없습니다.") # raise ValueError("응답 데이터에서 유효한 JSON 형식을 찾을 수 없습니다.") # return translated_data async def translate_options(self, original_data, product_name): """ 주어진 옵션 데이터를 Vertex AI 모델을 통해 번역하는 메서드. :param original_data: 원본 옵션 데이터 (dict 형태). :param product_name: 상품명 (str 형태). :return: 번역된 옵션명 (파이썬의 dict 형태). """ self.logger.debug(f"옵션 데이터를 번역 중: {original_data}") # 데이터 정리 cleaned_data = {key: self.clean_special_chars(value) for key, value in original_data.items()} self.logger.debug(f"정리된 옵션 데이터: {cleaned_data}") # 원본 데이터를 프롬프트 템플릿에 넣는다. option_prompt_template = self.prompt_data['option_prompt_template'] # 상품명과 옵션 데이터를 함께 전달 prompt = option_prompt_template.format(product_name=product_name, options=json.dumps(cleaned_data, ensure_ascii=False)) self.logger.debug(f"생성된 프롬프트: {prompt}") # Vertex AI 모델에 프롬프트 전달하여 응답 받기 self.logger.debug("Vertex AI 모델에 프롬프트를 전달하여 응답을 기다리는 중...") response = self.model.generate_content(prompt) # 비동기로 변경 self.logger.debug(f"모델 응답: {response.text}") # 응답 데이터에서 JSON 형식 추출 start = response.text.find('{') end = response.text.rfind('}') + 1 if start != -1 and end != -1: json_text = response.text[start:end] self.logger.debug(f"응답에서 추출된 JSON 텍스트: {json_text}") try: translated_data = json.loads(json_text) self.logger.debug(f"번역된 데이터: {translated_data}") except json.JSONDecodeError as e: self.logger.error(f"응답 데이터를 파싱하는 중 오류 발생: {e}", exc_info=True) raise ValueError(f"응답 데이터를 파이썬의 딕셔너리로 파싱하는 중 오류 발생: {e}") else: self.logger.error("응답 데이터에서 유효한 JSON 형식을 찾을 수 없습니다.") raise ValueError("응답 데이터에서 유효한 JSON 형식을 찾을 수 없습니다.") return translated_data