110 lines
4.3 KiB
Python
110 lines
4.3 KiB
Python
import copy
|
|
import re
|
|
import time
|
|
import json
|
|
from typing import Literal, Any
|
|
|
|
import httpx
|
|
|
|
from app.utils.logger import get_logger
|
|
from app.utils.prompts.chatgpt_prompt import ChatgptService
|
|
from app.utils.prompts.schemas import *
|
|
from app.utils.prompts.prompts import *
|
|
|
|
logger = get_logger("subtitle")
|
|
|
|
# 비한국어 출력에서 번역 누락(한글 잔존)을 탐지하기 위한 패턴
|
|
_HANGUL_RE = re.compile(r"[가-힣]")
|
|
|
|
# 한글 잔존 시 GPT 재호출 최대 횟수 (최초 호출 포함)
|
|
_MAX_LANGUAGE_ATTEMPTS = 3
|
|
|
|
|
|
class SubtitleContentsGenerator():
|
|
def __init__(self):
|
|
self.chatgpt_service = ChatgptService(timeout=60.0)
|
|
|
|
@staticmethod
|
|
def _find_hangul_leftovers(output_data: SubtitlePromptOutput) -> list[str]:
|
|
"""비한국어 출력에서 한글이 남아 있는 pitching_tag 목록을 반환합니다.
|
|
|
|
프롬프트가 '한국어로 생성 후 {language}로 번역'하는 2단계 구조라서,
|
|
항목 수가 많으면 일부가 번역되지 않은 채(한국어/혼합 문장) 돌아오는
|
|
사례가 있어 코드 레벨에서 검증한다.
|
|
"""
|
|
return [
|
|
result.pitching_tag
|
|
for result in output_data.pitching_results
|
|
if _HANGUL_RE.search(result.pitching_data)
|
|
]
|
|
|
|
async def generate_subtitle_contents(self, marketing_intelligence : dict[str, Any], pitching_label_list : list[Any], customer_name : str, detail_region_info : str, language : str = "Korean", industry: str = "") -> SubtitlePromptOutput:
|
|
start = time.perf_counter()
|
|
logger.info(
|
|
f"[SubtitleContentsGenerator] START - customer: {customer_name}, "
|
|
f"pitching_count: {len(pitching_label_list)}, "
|
|
f"labels: {pitching_label_list}, "
|
|
f"language: {language}, "
|
|
f"industry: {industry}"
|
|
)
|
|
|
|
dynamic_subtitle_prompt = create_dynamic_subtitle_prompt(len(pitching_label_list), industry)
|
|
pitching_label_string = "\n".join(pitching_label_list)
|
|
marketing_intel_string = json.dumps(marketing_intelligence, ensure_ascii=False)
|
|
input_data = {
|
|
"marketing_intelligence" : marketing_intel_string,
|
|
"pitching_tag_list_string" : pitching_label_string,
|
|
"customer_name" : customer_name,
|
|
"detail_region_info" : detail_region_info,
|
|
"language" : language,
|
|
"industry": industry, # 가사/영상 파이프라인에서 전달된 업종 enum
|
|
}
|
|
|
|
logger.info(
|
|
f"[SubtitleContentsGenerator] GPT 호출 시작 - model: {dynamic_subtitle_prompt.prompt_model}"
|
|
)
|
|
|
|
# 비한국어 언어는 번역 누락(한글 잔존) 검증 후 필요 시 재호출.
|
|
# 전부 실패하면 잔존 건수가 가장 적은 시도를 채택한다 (영상 생성 자체는 진행).
|
|
output_data = None
|
|
best_output = None
|
|
best_leftover_count: int | None = None
|
|
for lang_attempt in range(1, _MAX_LANGUAGE_ATTEMPTS + 1):
|
|
candidate = await self.chatgpt_service.generate_structured_output(dynamic_subtitle_prompt, input_data)
|
|
|
|
if language == "Korean":
|
|
output_data = candidate
|
|
break
|
|
|
|
leftovers = self._find_hangul_leftovers(candidate)
|
|
if not leftovers:
|
|
output_data = candidate
|
|
break
|
|
|
|
logger.warning(
|
|
f"[SubtitleContentsGenerator] 번역 누락(한글 잔존) {len(leftovers)}건 "
|
|
f"(attempt {lang_attempt}/{_MAX_LANGUAGE_ATTEMPTS}) - language: {language}, "
|
|
f"tags: {leftovers}"
|
|
)
|
|
if best_leftover_count is None or len(leftovers) < best_leftover_count:
|
|
best_output = candidate
|
|
best_leftover_count = len(leftovers)
|
|
|
|
if output_data is None:
|
|
logger.error(
|
|
f"[SubtitleContentsGenerator] 모든 시도에서 한글 잔존 - "
|
|
f"최소 잔존 {best_leftover_count}건 결과 채택 - language: {language}"
|
|
)
|
|
output_data = best_output
|
|
|
|
elapsed = (time.perf_counter() - start) * 1000
|
|
logger.info(
|
|
f"[SubtitleContentsGenerator] DONE - 소요시간: {elapsed:.0f}ms, "
|
|
f"결과: {[r.pitching_tag for r in output_data.pitching_results]}"
|
|
)
|
|
return output_data
|
|
|
|
|
|
|
|
|