710 lines
26 KiB
Python
710 lines
26 KiB
Python
# -*- coding: utf-8 -*-
|
|
"""
|
|
OpenRouter Translation API Python Client
|
|
- OpenRouter API를 통한 다양한 LLM 모델 번역 지원
|
|
- gemma_client.py와 동일한 인터페이스 제공
|
|
- OpenRouter 모델 ID를 직접 사용
|
|
|
|
사용 예:
|
|
from openrouter_client import OpenRouterTranslator
|
|
|
|
# 모델 ID 직접 지정
|
|
ort = OpenRouterTranslator(
|
|
api_key="sk-or-v1-xxx",
|
|
model_id="xiaomi/mimo-v2-flash:free"
|
|
)
|
|
|
|
# OCR 결과 번역
|
|
ko_list = ort.translate_ocr_texts(
|
|
product_name="휴대용 선풍기",
|
|
category="가전/계절가전",
|
|
ocr_results=[{"text":"强力送风"}, {"text":"USB-C 快速充电"}]
|
|
)
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import os
|
|
import time
|
|
import json
|
|
import random
|
|
import logging
|
|
from typing import List, Dict, Any, Optional
|
|
|
|
import requests
|
|
|
|
|
|
_JSON = Dict[str, Any]
|
|
|
|
|
|
class OpenRouterTranslatorError(RuntimeError):
|
|
"""OpenRouter 번역 클라이언트 예외"""
|
|
pass
|
|
|
|
|
|
class OpenRouterTranslator:
|
|
"""
|
|
OpenRouter API를 통한 번역 클라이언트.
|
|
|
|
GemmaTranslator와 동일한 인터페이스를 제공하여 호환성 유지.
|
|
|
|
Params
|
|
------
|
|
api_key : str
|
|
OpenRouter API 키 (sk-or-v1-xxx)
|
|
model_id : str
|
|
OpenRouter 모델 ID (예: "xiaomi/mimo-v2-flash:free", "deepseek/deepseek-r1-0528:free")
|
|
base_url : str
|
|
OpenRouter API 베이스 URL
|
|
timeout : int
|
|
요청 타임아웃(초)
|
|
max_retries : int
|
|
요청 재시도 횟수
|
|
backoff : float
|
|
재시도 backoff base (지수)
|
|
session : requests.Session | None
|
|
세션 주입 가능
|
|
logger : logging.Logger | None
|
|
로거 주입 가능
|
|
site_url : str | None
|
|
OpenRouter 대시보드에 표시될 사이트 URL (선택)
|
|
site_name : str | None
|
|
OpenRouter 대시보드에 표시될 사이트 이름 (선택)
|
|
"""
|
|
|
|
def __init__(
|
|
self,
|
|
api_key: Optional[str] = None,
|
|
model_id: str = "xiaomi/mimo-v2-flash:free",
|
|
base_url: str = "https://openrouter.ai/api/v1",
|
|
timeout: int = 120,
|
|
max_retries: int = 2,
|
|
backoff: float = 0.6,
|
|
session: Optional[requests.Session] = None,
|
|
logger: Optional[logging.Logger] = None,
|
|
site_url: Optional[str] = None,
|
|
site_name: Optional[str] = None,
|
|
) -> None:
|
|
self.api_key = api_key or os.getenv("OPENROUTER_API_KEY")
|
|
if not self.api_key:
|
|
raise OpenRouterTranslatorError("OpenRouter API 키가 필요합니다. api_key 파라미터 또는 OPENROUTER_API_KEY 환경변수를 설정하세요.")
|
|
|
|
self.base_url = base_url.rstrip("/")
|
|
self.timeout = timeout
|
|
self.max_retries = max_retries
|
|
self.backoff = backoff
|
|
self.sess = session or requests.Session()
|
|
self.log = logger or logging.getLogger(__name__)
|
|
self.site_url = site_url
|
|
self.site_name = site_name
|
|
|
|
# 모델 설정
|
|
self.model_id = model_id
|
|
self.log.info(f"[OpenRouterTranslator] 모델 설정: {self.model_id}")
|
|
|
|
# -----------------------------
|
|
# 모델 관리
|
|
# -----------------------------
|
|
def set_model(self, model_id: str) -> None:
|
|
"""
|
|
사용할 모델 변경
|
|
|
|
Args:
|
|
model_id: OpenRouter 모델 ID (예: "deepseek/deepseek-r1-0528:free")
|
|
"""
|
|
self.model_id = model_id
|
|
self.log.info(f"[OpenRouterTranslator] 모델 변경: {self.model_id}")
|
|
|
|
def get_current_model(self) -> Dict[str, Any]:
|
|
"""현재 설정된 모델 정보 반환"""
|
|
return {
|
|
"id": self.model_id,
|
|
"name": self.model_id,
|
|
}
|
|
|
|
# -----------------------------
|
|
# 내부 HTTP 헬퍼
|
|
# -----------------------------
|
|
def _build_headers(self) -> Dict[str, str]:
|
|
"""API 요청 헤더 생성"""
|
|
headers = {
|
|
"Authorization": f"Bearer {self.api_key}",
|
|
"Content-Type": "application/json",
|
|
}
|
|
if self.site_url:
|
|
headers["HTTP-Referer"] = self.site_url
|
|
if self.site_name:
|
|
headers["X-Title"] = self.site_name
|
|
return headers
|
|
|
|
def _post_chat(self, messages: List[Dict[str, str]], temperature: float = 0.3) -> str:
|
|
"""
|
|
OpenRouter Chat Completions API 호출
|
|
|
|
Returns
|
|
-------
|
|
str
|
|
모델 응답 텍스트
|
|
"""
|
|
url = f"{self.base_url}/chat/completions"
|
|
payload = {
|
|
"model": self.model_id,
|
|
"messages": messages,
|
|
"temperature": temperature,
|
|
}
|
|
|
|
last_err: Optional[Exception] = None
|
|
for attempt in range(self.max_retries + 1):
|
|
try:
|
|
r = self.sess.post(
|
|
url,
|
|
headers=self._build_headers(),
|
|
json=payload,
|
|
timeout=self.timeout
|
|
)
|
|
r.raise_for_status()
|
|
data = r.json()
|
|
|
|
# 응답 파싱
|
|
choices = data.get("choices", [])
|
|
if not choices:
|
|
raise OpenRouterTranslatorError("API 응답에 choices가 없습니다.")
|
|
|
|
content = choices[0].get("message", {}).get("content", "")
|
|
return content.strip()
|
|
|
|
except requests.exceptions.HTTPError as e:
|
|
last_err = e
|
|
# 429 Rate Limit, 5xx 서버 에러 시 재시도
|
|
if e.response is not None and e.response.status_code in (429, 500, 502, 503, 504):
|
|
if attempt < self.max_retries:
|
|
sleep_s = (self.backoff ** attempt) + random.uniform(0, 0.3)
|
|
self.log.warning(f"[OpenRouterTranslator] POST {url} 실패({e}), 재시도 {attempt+1}/{self.max_retries} 대기 {sleep_s:.2f}s")
|
|
time.sleep(sleep_s)
|
|
continue
|
|
raise OpenRouterTranslatorError(f"HTTP 에러: {e}")
|
|
|
|
except Exception as e:
|
|
last_err = e
|
|
if attempt < self.max_retries:
|
|
sleep_s = (self.backoff ** attempt) + random.uniform(0, 0.2)
|
|
self.log.warning(f"[OpenRouterTranslator] POST {url} 실패({e}), 재시도 {attempt+1}/{self.max_retries} 대기 {sleep_s:.2f}s")
|
|
time.sleep(sleep_s)
|
|
else:
|
|
break
|
|
|
|
raise OpenRouterTranslatorError(f"POST {url} 실패: {last_err}")
|
|
|
|
# -----------------------------
|
|
# 번역 프롬프트 생성
|
|
# -----------------------------
|
|
def _build_translation_prompt(
|
|
self,
|
|
product_name: str,
|
|
category: str,
|
|
items: List[Dict[str, Any]],
|
|
steps: int = 1
|
|
) -> str:
|
|
"""번역 요청 프롬프트 생성"""
|
|
|
|
items_json = json.dumps(items, ensure_ascii=False, indent=2)
|
|
|
|
if steps == 1:
|
|
# 직역 모드
|
|
prompt = f"""당신은 중국어-한국어 전문 온라인 쇼핑마케팅 번역가입니다.
|
|
|
|
상품 정보:
|
|
- 상품명: {product_name}
|
|
- 카테고리: {category}
|
|
|
|
아래 JSON 배열의 각 항목에서 "source" 필드의 중국어 텍스트를 한국어로 번역해주세요.
|
|
|
|
번역 규칙:
|
|
1. 정확하고 자연스러운 한국어로 번역
|
|
2. 상품 문맥에 맞게 번역
|
|
3. 브랜드명, 고유명사는 그대로 유지
|
|
4. 숫자와 단위는 유지
|
|
|
|
입력:
|
|
{items_json}
|
|
|
|
출력 형식 (JSON 배열만 출력, 다른 텍스트 없이):
|
|
[
|
|
{{"id": 1, "translation": "번역된 텍스트"}},
|
|
{{"id": 2, "translation": "번역된 텍스트"}}
|
|
]"""
|
|
else:
|
|
# 마케팅 톤 변환 모드 (steps == 2)
|
|
# Step 2는 이미 한국어로 번역된 텍스트를 입력으로 받음
|
|
prompt = f"""당신은 마케팅 카피라이터입니다.
|
|
|
|
상품 정보:
|
|
- 상품명: {product_name}
|
|
- 카테고리: {category}
|
|
|
|
아래 JSON 배열의 각 항목에서 "source" 필드의 한국어 텍스트를
|
|
한국 소비자에게 매력적으로 느껴지도록 마케팅 톤으로 다듬어주세요.
|
|
|
|
변환 규칙:
|
|
1. 원본의 의미는 정확히 유지
|
|
2. 한국 소비자에게 친숙하고 자연스러운 표현으로 변경
|
|
3. 상품의 장점을 부각하는 표현 사용
|
|
4. 브랜드명, 고유명사는 그대로 유지
|
|
5. 숫자와 단위는 유지
|
|
6. 너무 과장되지 않게, 자연스럽게
|
|
|
|
입력:
|
|
{items_json}
|
|
|
|
출력 형식 (JSON 배열만 출력, 다른 텍스트 없이):
|
|
[
|
|
{{"id": 1, "translation": "마케팅톤으로 다듬은 텍스트"}},
|
|
{{"id": 2, "translation": "마케팅톤으로 다듬은 텍스트"}}
|
|
]"""
|
|
|
|
return prompt
|
|
|
|
def _build_combined_translation_prompt(
|
|
self,
|
|
product_name: str,
|
|
category: str,
|
|
items: List[Dict[str, Any]]
|
|
) -> str:
|
|
"""
|
|
1단계(직역)와 2단계(마케팅톤 변환)를 합친 프롬프트 생성
|
|
|
|
Args:
|
|
product_name: 상품명
|
|
category: 카테고리
|
|
items: 번역할 항목 리스트 [{"id": 1, "source": "중국어 텍스트"}, ...]
|
|
|
|
Returns:
|
|
통합 프롬프트 문자열
|
|
"""
|
|
items_json = json.dumps(items, ensure_ascii=False, indent=2)
|
|
|
|
prompt = f"""당신은 중국어-한국어 전문 번역가이자 마케팅 카피라이터입니다.
|
|
|
|
상품 정보:
|
|
- 상품명: {product_name}
|
|
- 카테고리: {category}
|
|
|
|
아래 JSON 배열의 각 항목에서 "source" 필드의 중국어 텍스트를 다음 두 단계로 처리해주세요:
|
|
|
|
1단계: 정확하고 자연스러운 한국어로 번역
|
|
2단계: 한국 소비자에게 매력적으로 느껴지도록 마케팅 톤으로 다듬기
|
|
|
|
처리 규칙:
|
|
1. 정확한 의미 전달 (1단계)
|
|
2. 상품 문맥에 맞게 번역 (1단계)
|
|
3. 브랜드명, 고유명사는 그대로 유지
|
|
4. 숫자와 단위는 유지
|
|
5. 한국 소비자에게 친숙하고 자연스러운 표현 사용 (2단계)
|
|
6. 상품의 장점을 부각하는 표현 (2단계)
|
|
7. 너무 과장되지 않게, 자연스럽게 (2단계)
|
|
|
|
입력:
|
|
{items_json}
|
|
|
|
출력 형식 (JSON 배열만 출력, 다른 텍스트 없이):
|
|
[
|
|
{{"id": 1, "translation": "마케팅톤으로 다듬은 한국어 번역"}},
|
|
{{"id": 2, "translation": "마케팅톤으로 다듬은 한국어 번역"}}
|
|
]"""
|
|
|
|
return prompt
|
|
|
|
# -----------------------------
|
|
# 응답 파싱
|
|
# -----------------------------
|
|
def _parse_translation_response(self, response_text: str) -> List[Dict[str, Any]]:
|
|
"""
|
|
모델 응답에서 번역 결과 JSON 파싱
|
|
|
|
Returns
|
|
-------
|
|
List[Dict]
|
|
[{"id": 1, "translation": "..."}, ...]
|
|
"""
|
|
clean_str = response_text.strip()
|
|
|
|
# Markdown 코드 블록 제거
|
|
if clean_str.startswith("```"):
|
|
# 첫 줄 제거 (```json 등)
|
|
parts = clean_str.split('\n', 1)
|
|
if len(parts) > 1:
|
|
clean_str = parts[1]
|
|
else:
|
|
clean_str = clean_str.replace("```", "")
|
|
|
|
if clean_str.endswith("```"):
|
|
clean_str = clean_str[:-3]
|
|
|
|
clean_str = clean_str.strip()
|
|
|
|
try:
|
|
result = json.loads(clean_str)
|
|
if isinstance(result, list):
|
|
return result
|
|
elif isinstance(result, dict):
|
|
return [result]
|
|
else:
|
|
raise OpenRouterTranslatorError(f"예상치 못한 응답 타입: {type(result)}")
|
|
except json.JSONDecodeError as e:
|
|
self.log.error(f"[OpenRouterTranslator] JSON 파싱 실패: {e}\n응답: {response_text[:500]}...")
|
|
raise OpenRouterTranslatorError(f"JSON 파싱 실패: {e}")
|
|
|
|
# -----------------------------
|
|
# 공개 API (GemmaTranslator 호환)
|
|
# -----------------------------
|
|
def translate_ocr_texts(
|
|
self,
|
|
product_name: str,
|
|
category: str,
|
|
ocr_results: List[Dict[str, Any]],
|
|
batch_size: int = 16,
|
|
) -> List[str]:
|
|
"""
|
|
OCR 결과 번역 (GemmaTranslator.translate_ocr_texts 호환)
|
|
|
|
입력: OCR 결과 리스트(각 항목에 최소 'text' 키 필요)
|
|
출력: 원래 순서 유지한 ko 문자열 리스트
|
|
"""
|
|
if not ocr_results:
|
|
return []
|
|
|
|
# 유효성 검증
|
|
if not product_name or len(product_name.strip()) < 1:
|
|
raise OpenRouterTranslatorError("product_name은 1자 이상이어야 합니다.")
|
|
if not category or len(category.strip()) < 1:
|
|
raise OpenRouterTranslatorError("category는 1자 이상이어야 합니다.")
|
|
|
|
# id 부여 및 source 필터링
|
|
items = []
|
|
source_to_orig_idx = {}
|
|
for i, d in enumerate(ocr_results):
|
|
source = (d.get("text") or "").strip()
|
|
if len(source) >= 1:
|
|
item_id = len(items) + 1
|
|
items.append({"id": item_id, "source": source})
|
|
source_to_orig_idx[item_id] = i
|
|
|
|
if not items:
|
|
return [""] * len(ocr_results)
|
|
|
|
# 배치 처리
|
|
out_ko = [""] * len(ocr_results)
|
|
for i in range(0, len(items), batch_size):
|
|
chunk = items[i : i + batch_size]
|
|
|
|
prompt = self._build_translation_prompt(product_name, category, chunk, steps=1)
|
|
messages = [{"role": "user", "content": prompt}]
|
|
|
|
try:
|
|
response_text = self._post_chat(messages)
|
|
translated_items = self._parse_translation_response(response_text)
|
|
|
|
for item in translated_items:
|
|
item_id = int(item.get("id", 0))
|
|
translation = item.get("translation", "")
|
|
|
|
orig_idx = source_to_orig_idx.get(item_id)
|
|
if orig_idx is not None:
|
|
out_ko[orig_idx] = str(translation).strip() if translation else ""
|
|
|
|
except Exception as e:
|
|
self.log.error(f"[OpenRouterTranslator] 배치 번역 실패: {e}")
|
|
# 실패한 배치는 빈 문자열로 유지
|
|
continue
|
|
|
|
return out_ko
|
|
|
|
def run_llm_translation(
|
|
self,
|
|
product_name: str,
|
|
category: str,
|
|
ocr_results: List[Dict[str, Any]],
|
|
job_type: str = "ocr_translator_step1",
|
|
prompt_name: str = "ocr_translator_step1",
|
|
retry_count: int = 3,
|
|
steps: int = 1
|
|
) -> List[str]:
|
|
"""
|
|
LLM 번역 (GemmaTranslator.run_llm_translation 호환)
|
|
|
|
Args:
|
|
product_name: 상품명
|
|
category: 카테고리
|
|
ocr_results: OCR 결과 리스트
|
|
job_type: 작업 타입 (호환성용, 사용 안함)
|
|
prompt_name: 프롬프트 이름 (호환성용, 사용 안함)
|
|
retry_count: 재시도 횟수
|
|
steps: 번역 단계
|
|
1=직역만 (1회 API 호출)
|
|
2=직역 후 마케팅톤 변환 (2회 API 호출: step1 직역 → step2 마케팅톤 변환)
|
|
"""
|
|
if not ocr_results:
|
|
return []
|
|
|
|
# items 구성
|
|
items = []
|
|
source_to_orig_idx = {}
|
|
for i, res in enumerate(ocr_results):
|
|
text = (res.get("text") or "").strip()
|
|
if text:
|
|
item_id = len(items) + 1
|
|
items.append({"id": item_id, "source": text})
|
|
source_to_orig_idx[item_id] = i
|
|
|
|
if not items:
|
|
return [""] * len(ocr_results)
|
|
|
|
# steps 검증
|
|
steps = max(1, min(2, steps))
|
|
|
|
# 임시로 max_retries 조정
|
|
original_retries = self.max_retries
|
|
self.max_retries = max(retry_count - 1, 0)
|
|
|
|
try:
|
|
# Step 1: 직역
|
|
prompt_step1 = self._build_translation_prompt(product_name, category, items, steps=1)
|
|
messages_step1 = [{"role": "user", "content": prompt_step1}]
|
|
|
|
self.log.info(f"[OpenRouterTranslator] Step 1: 직역 시작")
|
|
response_text_step1 = self._post_chat(messages_step1)
|
|
translated_items_step1 = self._parse_translation_response(response_text_step1)
|
|
|
|
# Step 1 결과를 리스트로 변환
|
|
step1_results = [""] * len(ocr_results)
|
|
for item in translated_items_step1:
|
|
item_id = int(item.get("id", 0))
|
|
|
|
# 결과 필드 찾기 (translation > result > translated)
|
|
res_text = ""
|
|
if "translation" in item:
|
|
res_text = item["translation"]
|
|
elif "result" in item:
|
|
res_text = item["result"]
|
|
elif "translated" in item:
|
|
res_text = item["translated"]
|
|
|
|
if res_text is None:
|
|
res_text = ""
|
|
|
|
orig_idx = source_to_orig_idx.get(item_id)
|
|
if orig_idx is not None:
|
|
step1_results[orig_idx] = str(res_text).strip()
|
|
|
|
# Step 1만 필요한 경우
|
|
if steps == 1:
|
|
return step1_results
|
|
|
|
# Step 2: 마케팅톤 변환 (Step 1 결과를 입력으로 사용)
|
|
self.log.info(f"[OpenRouterTranslator] Step 2: 마케팅톤 변환 시작")
|
|
|
|
# Step 1 결과를 items 형태로 변환
|
|
items_step2 = []
|
|
for i, translated_text in enumerate(step1_results, 1):
|
|
if translated_text: # 빈 문자열이 아닌 경우만
|
|
items_step2.append({"id": i, "source": translated_text})
|
|
|
|
if not items_step2:
|
|
return step1_results
|
|
|
|
prompt_step2 = self._build_translation_prompt(product_name, category, items_step2, steps=2)
|
|
messages_step2 = [{"role": "user", "content": prompt_step2}]
|
|
|
|
response_text_step2 = self._post_chat(messages_step2)
|
|
translated_items_step2 = self._parse_translation_response(response_text_step2)
|
|
|
|
# Step 2 결과를 최종 결과로 변환
|
|
out_ko = [""] * len(ocr_results)
|
|
for item in translated_items_step2:
|
|
item_id = int(item.get("id", 0))
|
|
|
|
# 결과 필드 찾기 (translation > result > translated)
|
|
res_text = ""
|
|
if "translation" in item:
|
|
res_text = item["translation"]
|
|
elif "result" in item:
|
|
res_text = item["result"]
|
|
elif "translated" in item:
|
|
res_text = item["translated"]
|
|
|
|
if res_text is None:
|
|
res_text = ""
|
|
|
|
# item_id는 step2의 items 인덱스이므로, 원본 인덱스로 매핑
|
|
# step2 items는 step1 결과의 순서를 유지하므로, item_id - 1이 원본 인덱스
|
|
if 1 <= item_id <= len(step1_results):
|
|
orig_idx = item_id - 1
|
|
# 원본에서 실제로 번역된 항목인지 확인
|
|
if orig_idx < len(ocr_results):
|
|
out_ko[orig_idx] = str(res_text).strip()
|
|
|
|
return out_ko
|
|
|
|
except Exception as e:
|
|
self.log.error(f"[OpenRouterTranslator] run_llm_translation 실패: {e}")
|
|
raise e
|
|
finally:
|
|
self.max_retries = original_retries
|
|
|
|
def run_combined_llm_translation(
|
|
self,
|
|
product_name: str,
|
|
category: str,
|
|
ocr_results: List[Dict[str, Any]],
|
|
retry_count: int = 3
|
|
) -> List[str]:
|
|
"""
|
|
통합 프롬프트를 사용한 LLM 번역 (직역 + 마케팅톤 변환을 한 번에 수행)
|
|
|
|
Args:
|
|
product_name: 상품명
|
|
category: 카테고리
|
|
ocr_results: OCR 결과 리스트
|
|
retry_count: 재시도 횟수
|
|
|
|
Returns:
|
|
번역된 문자열 리스트 (원래 순서 유지)
|
|
|
|
Note:
|
|
이 메서드는 1단계(직역)와 2단계(마케팅톤 변환)를 한 번의 API 호출로 수행합니다.
|
|
run_llm_translation(steps=2)와 달리 2회의 API 호출이 아닌 1회만 호출합니다.
|
|
"""
|
|
if not ocr_results:
|
|
return []
|
|
|
|
# items 구성
|
|
items = []
|
|
source_to_orig_idx = {}
|
|
for i, res in enumerate(ocr_results):
|
|
text = (res.get("text") or "").strip()
|
|
if text:
|
|
item_id = len(items) + 1
|
|
items.append({"id": item_id, "source": text})
|
|
source_to_orig_idx[item_id] = i
|
|
|
|
if not items:
|
|
return [""] * len(ocr_results)
|
|
|
|
# 임시로 max_retries 조정
|
|
original_retries = self.max_retries
|
|
self.max_retries = max(retry_count - 1, 0)
|
|
|
|
try:
|
|
# 통합 프롬프트 사용 (직역 + 마케팅톤 변환)
|
|
prompt = self._build_combined_translation_prompt(product_name, category, items)
|
|
messages = [{"role": "user", "content": prompt}]
|
|
|
|
self.log.info(f"[OpenRouterTranslator] 통합 번역 시작 (직역 + 마케팅톤 변환)")
|
|
response_text = self._post_chat(messages)
|
|
translated_items = self._parse_translation_response(response_text)
|
|
|
|
# 결과를 리스트로 변환
|
|
out_ko = [""] * len(ocr_results)
|
|
for item in translated_items:
|
|
item_id = int(item.get("id", 0))
|
|
|
|
# 결과 필드 찾기 (translation > result > translated)
|
|
res_text = ""
|
|
if "translation" in item:
|
|
res_text = item["translation"]
|
|
elif "result" in item:
|
|
res_text = item["result"]
|
|
elif "translated" in item:
|
|
res_text = item["translated"]
|
|
|
|
if res_text is None:
|
|
res_text = ""
|
|
|
|
orig_idx = source_to_orig_idx.get(item_id)
|
|
if orig_idx is not None:
|
|
out_ko[orig_idx] = str(res_text).strip()
|
|
|
|
return out_ko
|
|
|
|
except Exception as e:
|
|
self.log.error(f"[OpenRouterTranslator] run_combined_llm_translation 실패: {e}")
|
|
raise e
|
|
finally:
|
|
self.max_retries = original_retries
|
|
|
|
def batch_translate_texts(
|
|
self,
|
|
product_name: str,
|
|
category: str,
|
|
text_list: List[str],
|
|
delimiter: str = " / ",
|
|
batch_size: int = 8,
|
|
) -> List[str]:
|
|
"""
|
|
순수 텍스트 리스트 번역 (GemmaTranslator.batch_translate_texts 호환)
|
|
"""
|
|
if not text_list:
|
|
return []
|
|
|
|
# text_list를 ocr_results 형태로 변환
|
|
ocr_results = [{"text": t} for t in text_list]
|
|
return self.translate_ocr_texts(product_name, category, ocr_results, batch_size)
|
|
|
|
def translate_option_groups(
|
|
self,
|
|
product_name: str,
|
|
category: str,
|
|
option_groups: List[Dict[str, Any]],
|
|
batch_size: int = 8,
|
|
) -> List[Dict[str, Any]]:
|
|
"""
|
|
옵션 그룹 번역 (GemmaTranslator.translate_option_groups 호환)
|
|
|
|
option_groups 예:
|
|
[{"id": 1, "source": ["红色","蓝色"]}, {"id": 2, "source": ["小号","大号"]}]
|
|
반환:
|
|
[{"id": 1, "translations": ["빨강","파랑"]}, {"id": 2, "translations": ["소형","대형"]}]
|
|
"""
|
|
if not option_groups:
|
|
return []
|
|
|
|
results = []
|
|
for group in option_groups:
|
|
group_id = group.get("id")
|
|
sources = group.get("source", [])
|
|
|
|
if not sources:
|
|
results.append({"id": group_id, "translations": []})
|
|
continue
|
|
|
|
# 각 옵션을 번역
|
|
ocr_results = [{"text": s} for s in sources]
|
|
translations = self.translate_ocr_texts(product_name, category, ocr_results, batch_size)
|
|
|
|
results.append({"id": group_id, "translations": translations})
|
|
|
|
return results
|
|
|
|
# -----------------------------
|
|
# 유틸리티
|
|
# -----------------------------
|
|
def health(self) -> _JSON:
|
|
"""API 상태 확인 (간단한 테스트 요청)"""
|
|
try:
|
|
messages = [{"role": "user", "content": "Hello"}]
|
|
self._post_chat(messages)
|
|
return {"status": "ok", "model": self.model_id}
|
|
except Exception as e:
|
|
return {"status": "error", "error": str(e)}
|
|
|
|
def get_credits(self) -> _JSON:
|
|
"""OpenRouter 크레딧 잔액 조회"""
|
|
url = f"{self.base_url}/credits"
|
|
try:
|
|
r = self.sess.get(url, headers=self._build_headers(), timeout=10)
|
|
r.raise_for_status()
|
|
return r.json()
|
|
except Exception as e:
|
|
self.log.error(f"[OpenRouterTranslator] 크레딧 조회 실패: {e}")
|
|
return {"error": str(e)}
|
|
|