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