""" Song Background Tasks 노래 생성 관련 백그라운드 태스크를 정의합니다. """ from datetime import date from pathlib import Path import aiofiles import httpx from sqlalchemy import select from app.database.session import AsyncSessionLocal from app.song.models import Song from app.utils.common import generate_task_id from app.utils.upload_blob_as_request import AzureBlobUploader from config import prj_settings async def download_and_save_song( task_id: str, audio_url: str, store_name: str, ) -> None: """백그라운드에서 노래를 다운로드하고 Song 테이블을 업데이트합니다. Args: task_id: 프로젝트 task_id audio_url: 다운로드할 오디오 URL store_name: 저장할 파일명에 사용할 업체명 """ print(f"[download_and_save_song] START - task_id: {task_id}, store_name: {store_name}") try: # 저장 경로 생성: media/song/{날짜}/{uuid7}/{store_name}.mp3 today = date.today().strftime("%Y-%m-%d") unique_id = await generate_task_id() # 파일명에 사용할 수 없는 문자 제거 safe_store_name = "".join( c for c in store_name if c.isalnum() or c in (" ", "_", "-") ).strip() safe_store_name = safe_store_name or "song" file_name = f"{safe_store_name}.mp3" # 절대 경로 생성 media_dir = Path("media") / "song" / today / unique_id media_dir.mkdir(parents=True, exist_ok=True) file_path = media_dir / file_name print(f"[download_and_save_song] Directory created - path: {file_path}") # 오디오 파일 다운로드 print(f"[download_and_save_song] Downloading audio - task_id: {task_id}, url: {audio_url}") async with httpx.AsyncClient() as client: response = await client.get(audio_url, timeout=60.0) response.raise_for_status() async with aiofiles.open(str(file_path), "wb") as f: await f.write(response.content) print(f"[download_and_save_song] File saved - task_id: {task_id}, path: {file_path}") # 프론트엔드에서 접근 가능한 URL 생성 relative_path = f"/media/song/{today}/{unique_id}/{file_name}" base_url = f"http://{prj_settings.PROJECT_DOMAIN}" file_url = f"{base_url}{relative_path}" print(f"[download_and_save_song] URL generated - task_id: {task_id}, url: {file_url}") # Song 테이블 업데이트 (새 세션 사용) async with AsyncSessionLocal() as session: # 여러 개 있을 경우 가장 최근 것 선택 result = await session.execute( select(Song) .where(Song.task_id == task_id) .order_by(Song.created_at.desc()) .limit(1) ) song = result.scalar_one_or_none() if song: song.status = "completed" song.song_result_url = file_url await session.commit() print(f"[download_and_save_song] SUCCESS - task_id: {task_id}, status: completed") else: print(f"[download_and_save_song] Song NOT FOUND in DB - task_id: {task_id}") except Exception as e: print(f"[download_and_save_song] EXCEPTION - task_id: {task_id}, error: {e}") # 실패 시 Song 테이블 업데이트 async with AsyncSessionLocal() as session: # 여러 개 있을 경우 가장 최근 것 선택 result = await session.execute( select(Song) .where(Song.task_id == task_id) .order_by(Song.created_at.desc()) .limit(1) ) song = result.scalar_one_or_none() if song: song.status = "failed" await session.commit() print(f"[download_and_save_song] FAILED - task_id: {task_id}, status updated to failed") async def download_and_upload_song_to_blob( task_id: str, audio_url: str, store_name: str, ) -> None: """백그라운드에서 노래를 다운로드하고 Azure Blob Storage에 업로드한 뒤 Song 테이블을 업데이트합니다. Args: task_id: 프로젝트 task_id audio_url: 다운로드할 오디오 URL store_name: 저장할 파일명에 사용할 업체명 """ print(f"[download_and_upload_song_to_blob] START - task_id: {task_id}, store_name: {store_name}") temp_file_path: Path | None = None try: # 파일명에 사용할 수 없는 문자 제거 safe_store_name = "".join( c for c in store_name if c.isalnum() or c in (" ", "_", "-") ).strip() safe_store_name = safe_store_name or "song" file_name = f"{safe_store_name}.mp3" # 임시 저장 경로 생성 temp_dir = Path("media") / "temp" / task_id temp_dir.mkdir(parents=True, exist_ok=True) temp_file_path = temp_dir / file_name print(f"[download_and_upload_song_to_blob] Temp directory created - path: {temp_file_path}") # 오디오 파일 다운로드 print(f"[download_and_upload_song_to_blob] Downloading audio - task_id: {task_id}, url: {audio_url}") async with httpx.AsyncClient() as client: response = await client.get(audio_url, timeout=60.0) response.raise_for_status() async with aiofiles.open(str(temp_file_path), "wb") as f: await f.write(response.content) print(f"[download_and_upload_song_to_blob] File downloaded - task_id: {task_id}, path: {temp_file_path}") # Azure Blob Storage에 업로드 uploader = AzureBlobUploader(task_id=task_id) upload_success = await uploader.upload_music(file_path=str(temp_file_path)) if not upload_success: raise Exception("Azure Blob Storage 업로드 실패") # SAS 토큰이 제외된 public_url 사용 blob_url = uploader.public_url print(f"[download_and_upload_song_to_blob] Uploaded to Blob - task_id: {task_id}, url: {blob_url}") # Song 테이블 업데이트 (새 세션 사용) async with AsyncSessionLocal() as session: # 여러 개 있을 경우 가장 최근 것 선택 result = await session.execute( select(Song) .where(Song.task_id == task_id) .order_by(Song.created_at.desc()) .limit(1) ) song = result.scalar_one_or_none() if song: song.status = "completed" song.song_result_url = blob_url await session.commit() print(f"[download_and_upload_song_to_blob] SUCCESS - task_id: {task_id}, status: completed") else: print(f"[download_and_upload_song_to_blob] Song NOT FOUND in DB - task_id: {task_id}") except Exception as e: print(f"[download_and_upload_song_to_blob] EXCEPTION - task_id: {task_id}, error: {e}") # 실패 시 Song 테이블 업데이트 async with AsyncSessionLocal() as session: result = await session.execute( select(Song) .where(Song.task_id == task_id) .order_by(Song.created_at.desc()) .limit(1) ) song = result.scalar_one_or_none() if song: song.status = "failed" await session.commit() print(f"[download_and_upload_song_to_blob] FAILED - task_id: {task_id}, status updated to failed") finally: # 임시 파일 삭제 if temp_file_path and temp_file_path.exists(): try: temp_file_path.unlink() print(f"[download_and_upload_song_to_blob] Temp file deleted - path: {temp_file_path}") except Exception as e: print(f"[download_and_upload_song_to_blob] Failed to delete temp file: {e}") # 임시 디렉토리 삭제 시도 temp_dir = Path("media") / "temp" / task_id if temp_dir.exists(): try: temp_dir.rmdir() except Exception: pass # 디렉토리가 비어있지 않으면 무시 async def download_and_upload_song_by_suno_task_id( suno_task_id: str, audio_url: str, store_name: str, duration: float | None = None, ) -> None: """suno_task_id로 Song을 조회하여 노래를 다운로드하고 Azure Blob Storage에 업로드한 뒤 Song 테이블을 업데이트합니다. Args: suno_task_id: Suno API 작업 ID audio_url: 다운로드할 오디오 URL store_name: 저장할 파일명에 사용할 업체명 duration: 노래 재생 시간 (초) """ print(f"[download_and_upload_song_by_suno_task_id] START - suno_task_id: {suno_task_id}, store_name: {store_name}, duration: {duration}") temp_file_path: Path | None = None task_id: str | None = None try: # suno_task_id로 Song 조회하여 task_id 가져오기 async with AsyncSessionLocal() as session: result = await session.execute( select(Song) .where(Song.suno_task_id == suno_task_id) .order_by(Song.created_at.desc()) .limit(1) ) song = result.scalar_one_or_none() if not song: print(f"[download_and_upload_song_by_suno_task_id] Song NOT FOUND - suno_task_id: {suno_task_id}") return task_id = song.task_id print(f"[download_and_upload_song_by_suno_task_id] Song found - suno_task_id: {suno_task_id}, task_id: {task_id}") # 파일명에 사용할 수 없는 문자 제거 safe_store_name = "".join( c for c in store_name if c.isalnum() or c in (" ", "_", "-") ).strip() safe_store_name = safe_store_name or "song" file_name = f"{safe_store_name}.mp3" # 임시 저장 경로 생성 temp_dir = Path("media") / "temp" / task_id temp_dir.mkdir(parents=True, exist_ok=True) temp_file_path = temp_dir / file_name print(f"[download_and_upload_song_by_suno_task_id] Temp directory created - path: {temp_file_path}") # 오디오 파일 다운로드 print(f"[download_and_upload_song_by_suno_task_id] Downloading audio - suno_task_id: {suno_task_id}, url: {audio_url}") async with httpx.AsyncClient() as client: response = await client.get(audio_url, timeout=60.0) response.raise_for_status() async with aiofiles.open(str(temp_file_path), "wb") as f: await f.write(response.content) print(f"[download_and_upload_song_by_suno_task_id] File downloaded - suno_task_id: {suno_task_id}, path: {temp_file_path}") # Azure Blob Storage에 업로드 uploader = AzureBlobUploader(task_id=task_id) upload_success = await uploader.upload_music(file_path=str(temp_file_path)) if not upload_success: raise Exception("Azure Blob Storage 업로드 실패") # SAS 토큰이 제외된 public_url 사용 blob_url = uploader.public_url print(f"[download_and_upload_song_by_suno_task_id] Uploaded to Blob - suno_task_id: {suno_task_id}, url: {blob_url}") # Song 테이블 업데이트 (새 세션 사용) async with AsyncSessionLocal() as session: result = await session.execute( select(Song) .where(Song.suno_task_id == suno_task_id) .order_by(Song.created_at.desc()) .limit(1) ) song = result.scalar_one_or_none() if song: song.status = "completed" song.song_result_url = blob_url if duration is not None: song.duration = duration await session.commit() print(f"[download_and_upload_song_by_suno_task_id] SUCCESS - suno_task_id: {suno_task_id}, status: completed, duration: {duration}") else: print(f"[download_and_upload_song_by_suno_task_id] Song NOT FOUND in DB - suno_task_id: {suno_task_id}") except Exception as e: print(f"[download_and_upload_song_by_suno_task_id] EXCEPTION - suno_task_id: {suno_task_id}, error: {e}") # 실패 시 Song 테이블 업데이트 if task_id: async with AsyncSessionLocal() as session: result = await session.execute( select(Song) .where(Song.suno_task_id == suno_task_id) .order_by(Song.created_at.desc()) .limit(1) ) song = result.scalar_one_or_none() if song: song.status = "failed" await session.commit() print(f"[download_and_upload_song_by_suno_task_id] FAILED - suno_task_id: {suno_task_id}, status updated to failed") finally: # 임시 파일 삭제 if temp_file_path and temp_file_path.exists(): try: temp_file_path.unlink() print(f"[download_and_upload_song_by_suno_task_id] Temp file deleted - path: {temp_file_path}") except Exception as e: print(f"[download_and_upload_song_by_suno_task_id] Failed to delete temp file: {e}") # 임시 디렉토리 삭제 시도 if task_id: temp_dir = Path("media") / "temp" / task_id if temp_dir.exists(): try: temp_dir.rmdir() except Exception: pass # 디렉토리가 비어있지 않으면 무시