""" Lyric Background Tasks 가사 생성 관련 백그라운드 태스크를 정의합니다. """ import traceback from sqlalchemy import select from sqlalchemy.exc import SQLAlchemyError from app.database.session import BackgroundSessionLocal from app.lyric.models import Lyric from app.utils.chatgpt_prompt import ChatgptService, ChatGPTResponseError from app.utils.prompts.prompts import Prompt from app.utils.logger import get_logger # 로거 설정 logger = get_logger("lyric") async def _update_lyric_status( task_id: str, status: str, result: str | None = None, lyric_id: int | None = None, ) -> bool: """Lyric 테이블의 상태를 업데이트합니다. Args: task_id: 프로젝트 task_id status: 변경할 상태 ("processing", "completed", "failed") result: 가사 결과 또는 에러 메시지 lyric_id: 특정 Lyric 레코드 ID (재생성 시 정확한 레코드 식별용) Returns: bool: 업데이트 성공 여부 """ try: async with BackgroundSessionLocal() as session: if lyric_id: # lyric_id로 특정 레코드 조회 (재생성 시에도 정확한 레코드 업데이트) query_result = await session.execute( select(Lyric).where(Lyric.id == lyric_id) ) else: # 기존 방식: task_id로 최신 레코드 조회 query_result = await session.execute( select(Lyric) .where(Lyric.task_id == task_id) .order_by(Lyric.created_at.desc()) .limit(1) ) lyric = query_result.scalar_one_or_none() if lyric: lyric.status = status if result is not None: lyric.lyric_result = result await session.commit() logger.info(f"[Lyric] Status updated - task_id: {task_id}, lyric_id: {lyric_id}, status: {status}") return True else: logger.warning(f"[Lyric] NOT FOUND in DB - task_id: {task_id}, lyric_id: {lyric_id}") return False except SQLAlchemyError as e: logger.error(f"[Lyric] DB Error while updating status - task_id: {task_id}, lyric_id: {lyric_id}, error: {e}") return False except Exception as e: logger.error(f"[Lyric] Unexpected error while updating status - task_id: {task_id}, lyric_id: {lyric_id}, error: {e}") return False async def generate_lyric_background( task_id: str, prompt: Prompt, lyric_input_data: dict, # 프롬프트 메타데이터에서 정의된 Input lyric_id: int | None = None, ) -> None: """백그라운드에서 ChatGPT를 통해 가사를 생성하고 Lyric 테이블을 업데이트합니다. Args: task_id: 프로젝트 task_id prompt: ChatGPT에 전달할 프롬프트 lyric_input_data: 프롬프트 입력 데이터 lyric_id: 특정 Lyric 레코드 ID (재생성 시 정확한 레코드 식별용) """ import time task_start = time.perf_counter() logger.info(f"[generate_lyric_background] START - task_id: {task_id}") logger.debug(f"[generate_lyric_background] ========== START ==========") logger.debug(f"[generate_lyric_background] task_id: {task_id}") logger.debug(f"[generate_lyric_background] language: {lyric_input_data['language']}") #logger.debug(f"[generate_lyric_background] prompt length: {len(prompt)}자") try: # ========== Step 1: ChatGPT 서비스 초기화 ========== step1_start = time.perf_counter() logger.debug(f"[generate_lyric_background] Step 1: ChatGPT 서비스 초기화...") # service = ChatgptService( # customer_name="", # 프롬프트가 이미 생성되었으므로 빈 값 # region="", # detail_region_info="", # language=language, # ) chatgpt = ChatgptService() step1_elapsed = (time.perf_counter() - step1_start) * 1000 logger.debug(f"[generate_lyric_background] Step 1 완료 ({step1_elapsed:.1f}ms)") # ========== Step 2: ChatGPT API 호출 (가사 생성) ========== step2_start = time.perf_counter() logger.info(f"[generate_lyric_background] Step 2: ChatGPT API 호출 시작 - task_id: {task_id}") logger.debug(f"[generate_lyric_background] Step 2: ChatGPT API 호출 시작...") #result = await service.generate(prompt=prompt) result_response = await chatgpt.generate_structured_output(prompt, lyric_input_data) result = result_response.lyric step2_elapsed = (time.perf_counter() - step2_start) * 1000 logger.info(f"[generate_lyric_background] Step 2 완료 - 응답 {len(result)}자 ({step2_elapsed:.1f}ms)") # ========== Step 3: DB 상태 업데이트 ========== step3_start = time.perf_counter() logger.debug(f"[generate_lyric_background] Step 3: DB 상태 업데이트...") await _update_lyric_status(task_id, "completed", result, lyric_id) step3_elapsed = (time.perf_counter() - step3_start) * 1000 logger.debug(f"[generate_lyric_background] Step 3 완료 ({step3_elapsed:.1f}ms)") # ========== 완료 ========== total_elapsed = (time.perf_counter() - task_start) * 1000 logger.info(f"[generate_lyric_background] SUCCESS - task_id: {task_id}, 총 소요시간: {total_elapsed:.1f}ms") logger.debug(f"[generate_lyric_background] ========== SUCCESS ==========") logger.debug(f"[generate_lyric_background] 총 소요시간: {total_elapsed:.1f}ms") logger.debug(f"[generate_lyric_background] - Step 1 (서비스 초기화): {step1_elapsed:.1f}ms") logger.debug(f"[generate_lyric_background] - Step 2 (GPT API 호출): {step2_elapsed:.1f}ms") logger.debug(f"[generate_lyric_background] - Step 3 (DB 업데이트): {step3_elapsed:.1f}ms") except ChatGPTResponseError as e: elapsed = (time.perf_counter() - task_start) * 1000 logger.error( f"[generate_lyric_background] ChatGPT ERROR - task_id: {task_id}, " f"status: {e.status}, code: {e.error_code}, message: {e.error_message} ({elapsed:.1f}ms)" ) await _update_lyric_status(task_id, "failed", f"ChatGPT Error: {e.error_message}", lyric_id) except SQLAlchemyError as e: elapsed = (time.perf_counter() - task_start) * 1000 logger.error(f"[generate_lyric_background] DB ERROR - task_id: {task_id}, error: {e} ({elapsed:.1f}ms)", exc_info=True) await _update_lyric_status(task_id, "failed", f"Database Error: {str(e)}", lyric_id) except Exception as e: elapsed = (time.perf_counter() - task_start) * 1000 logger.error(f"[generate_lyric_background] EXCEPTION - task_id: {task_id}, error: {e} ({elapsed:.1f}ms)", exc_info=True) await _update_lyric_status(task_id, "failed", f"Error: {str(e)}", lyric_id)