배경음악 기능 추가

feature-BGM
김성경 2026-05-12 15:18:35 +09:00
parent 2c8c16c288
commit ec3e0159e8
7 changed files with 169 additions and 77 deletions

View File

@ -356,18 +356,26 @@ async def generate_lyric(
step4_start = time.perf_counter() step4_start = time.perf_counter()
logger.debug(f"[generate_lyric] Step 4: 백그라운드 태스크 스케줄링...") logger.debug(f"[generate_lyric] Step 4: 백그라운드 태스크 스케줄링...")
orientation = request_body.orientation orientation = request_body.orientation
background_tasks.add_task(
generate_lyric_background, if request_body.instrumental:
task_id=task_id, # BGM 모드: ChatGPT 가사 생성 없이 Lyric을 즉시 completed로 마무리
prompt=lyric_prompt, lyric.status = "completed"
lyric_input_data=lyric_input_data, lyric.lyric_result = ""
lyric_id=lyric.id, await session.commit()
) logger.info(f"[generate_lyric] BGM 모드 - 가사 생성 스킵, lyric_id: {lyric.id}")
else:
background_tasks.add_task(
generate_lyric_background,
task_id=task_id,
prompt=lyric_prompt,
lyric_input_data=lyric_input_data,
lyric_id=lyric.id,
)
background_tasks.add_task( background_tasks.add_task(
generate_subtitle_background, generate_subtitle_background,
orientation = orientation, orientation=orientation,
task_id=task_id task_id=task_id,
) )
step4_elapsed = (time.perf_counter() - step4_start) * 1000 step4_elapsed = (time.perf_counter() - step4_start) * 1000

View File

@ -76,6 +76,7 @@ class GenerateLyricRequest(BaseModel):
description="영상 방향 (horizontal: 가로형, vertical: 세로형)", description="영상 방향 (horizontal: 가로형, vertical: 세로형)",
), ),
m_id : Optional[int] = Field(None, description="마케팅 인텔리전스 ID 값") m_id : Optional[int] = Field(None, description="마케팅 인텔리전스 ID 값")
instrumental: bool = Field(default=False, description="BGM 전용 모드 (가사 생성 안 함)")
class GenerateLyricResponse(BaseModel): class GenerateLyricResponse(BaseModel):

View File

@ -185,9 +185,10 @@ async def generate_song(
) )
# Song 테이블에 초기 데이터 저장 # Song 테이블에 초기 데이터 저장
song_prompt = ( if request_body.instrumental:
f"[Lyrics]\n{request_body.lyrics}\n\n[Genre]\n{request_body.genre}" song_prompt = f"[Instrumental]\n[Genre]\n{request_body.genre}"
) else:
song_prompt = f"[Lyrics]\n{request_body.lyrics}\n\n[Genre]\n{request_body.genre}"
logger.debug( logger.debug(
f"[generate_song] Lyrics comparison - task_id: {task_id}\n" f"[generate_song] Lyrics comparison - task_id: {task_id}\n"
f"{'=' * 60}\n" f"{'=' * 60}\n"
@ -249,6 +250,7 @@ async def generate_song(
suno_task_id = await suno_service.generate( suno_task_id = await suno_service.generate(
prompt=request_body.lyrics, prompt=request_body.lyrics,
genre=request_body.genre, genre=request_body.genre,
instrumental=request_body.instrumental,
) )
stage2_time = time.perf_counter() stage2_time = time.perf_counter()
@ -452,14 +454,8 @@ async def get_song_status(
) )
suno_audio_id = first_clip.get("id") suno_audio_id = first_clip.get("id")
word_data = await suno_service.get_lyric_timestamp(
suno_task_id, suno_audio_id # BGM 모드(lyric_result가 비어 있음)에서는 타임스탬프 생성 스킵
)
logger.debug(
f"[get_song_status] word_data from get_lyric_timestamp - "
f"suno_task_id: {suno_task_id}, suno_audio_id: {suno_audio_id}, "
f"word_data: {word_data}"
)
lyric_result = await session.execute( lyric_result = await session.execute(
select(Lyric) select(Lyric)
.where(Lyric.task_id == song.task_id) .where(Lyric.task_id == song.task_id)
@ -467,51 +463,60 @@ async def get_song_status(
.limit(1) .limit(1)
) )
lyric = lyric_result.scalar_one_or_none() lyric = lyric_result.scalar_one_or_none()
gt_lyric = lyric.lyric_result gt_lyric = lyric.lyric_result if lyric else None
lyric_line_list = gt_lyric.split("\n")
sentences = [
lyric_line.strip(",. ")
for lyric_line in lyric_line_list
if lyric_line and lyric_line != "---"
]
logger.debug(
f"[get_song_status] sentences from lyric - "
f"sentences: {sentences}"
)
timestamped_lyrics = suno_service.align_lyrics( if gt_lyric:
word_data, sentences word_data = await suno_service.get_lyric_timestamp(
) suno_task_id, suno_audio_id
logger.debug( )
f"[get_song_status] sentences from lyric - " logger.debug(
f"sentences: {sentences}" f"[get_song_status] word_data from get_lyric_timestamp - "
) f"suno_task_id: {suno_task_id}, suno_audio_id: {suno_audio_id}, "
f"word_data: {word_data}"
# TODO : DB upload timestamped_lyrics )
for order_idx, timestamped_lyric in enumerate( lyric_line_list = gt_lyric.split("\n")
timestamped_lyrics sentences = [
): lyric_line.strip(",. ")
# start_sec 또는 end_sec가 None인 경우 건너뛰기 for lyric_line in lyric_line_list
if ( if lyric_line and lyric_line != "---"
timestamped_lyric["start_sec"] is None ]
or timestamped_lyric["end_sec"] is None logger.debug(
): f"[get_song_status] sentences from lyric - "
logger.warning( f"sentences: {sentences}"
f"[get_song_status] Skipping timestamp - " )
f"lyric_line: {timestamped_lyric['text']}, "
f"start_sec: {timestamped_lyric['start_sec']}, " timestamped_lyrics = suno_service.align_lyrics(
f"end_sec: {timestamped_lyric['end_sec']}" word_data, sentences
) )
continue
for order_idx, timestamped_lyric in enumerate(
song_timestamp = SongTimestamp( timestamped_lyrics
suno_audio_id=suno_audio_id, ):
order_idx=order_idx, if (
lyric_line=timestamped_lyric["text"], timestamped_lyric["start_sec"] is None
start_time=timestamped_lyric["start_sec"], or timestamped_lyric["end_sec"] is None
end_time=timestamped_lyric["end_sec"], ):
logger.warning(
f"[get_song_status] Skipping timestamp - "
f"lyric_line: {timestamped_lyric['text']}, "
f"start_sec: {timestamped_lyric['start_sec']}, "
f"end_sec: {timestamped_lyric['end_sec']}"
)
continue
song_timestamp = SongTimestamp(
suno_audio_id=suno_audio_id,
order_idx=order_idx,
lyric_line=timestamped_lyric["text"],
start_time=timestamped_lyric["start_sec"],
end_time=timestamped_lyric["end_sec"],
)
session.add(song_timestamp)
else:
logger.info(
f"[get_song_status] BGM 모드 - 타임스탬프 생성 스킵, "
f"suno_task_id: {suno_task_id}"
) )
session.add(song_timestamp)
await session.commit() await session.commit()
parsed_response.status = "processing" parsed_response.status = "processing"

View File

@ -1,6 +1,6 @@
from typing import Optional from typing import Optional
from pydantic import BaseModel, Field from pydantic import BaseModel, Field, model_validator
# ============================================================================= # =============================================================================
@ -33,7 +33,7 @@ class GenerateSongRequest(BaseModel):
} }
} }
lyrics: str = Field(..., description="노래에 사용할 가사") lyrics: Optional[str] = Field(None, description="노래에 사용할 가사 (instrumental=True이면 생략 가능)")
genre: str = Field( genre: str = Field(
..., ...,
description="음악 장르 (K-Pop, Pop, R&B, Hip-Hop, Ballad, EDM, Rock, Jazz 등)", description="음악 장르 (K-Pop, Pop, R&B, Hip-Hop, Ballad, EDM, Rock, Jazz 등)",
@ -42,6 +42,15 @@ class GenerateSongRequest(BaseModel):
default="Korean", default="Korean",
description="노래 언어 (Korean, English, Chinese, Japanese, Thai, Vietnamese)", description="노래 언어 (Korean, English, Chinese, Japanese, Thai, Vietnamese)",
) )
instrumental: bool = Field(default=False, description="BGM 전용 모드 (가사 없이 음악만 생성)")
@model_validator(mode="after")
def validate_lyrics_required(self) -> "GenerateSongRequest":
if not self.instrumental and not self.lyrics:
raise ValueError("instrumental=False일 때 lyrics는 필수입니다.")
if self.instrumental:
self.lyrics = None
return self
class GenerateSongResponse(BaseModel): class GenerateSongResponse(BaseModel):

View File

@ -31,7 +31,7 @@ async def autotag_images(image_url_list : list[str]) -> list[dict]: #tag_list
image_result_tasks = [chatgpt.generate_structured_output(image_autotag_prompt, image_input_data, image_input_data['img_url'], False, silent = True) for image_input_data in image_input_data_list] image_result_tasks = [chatgpt.generate_structured_output(image_autotag_prompt, image_input_data, image_input_data['img_url'], False, silent = True) for image_input_data in image_input_data_list]
image_result_list: list[BaseModel | BaseException] = await asyncio.gather(*image_result_tasks, return_exceptions=True) image_result_list: list[BaseModel | BaseException] = await asyncio.gather(*image_result_tasks, return_exceptions=True)
MAX_RETRY = 3 # 하드코딩, 어떻게 처리할지는 나중에 MAX_RETRY = 2 # 하드코딩, 어떻게 처리할지는 나중에
for _ in range(MAX_RETRY): for _ in range(MAX_RETRY):
failed_idx = [i for i, r in enumerate(image_result_list) if isinstance(r, Exception)] failed_idx = [i for i, r in enumerate(image_result_list) if isinstance(r, Exception)]
print("Failed", failed_idx) print("Failed", failed_idx)

63
app/utils/bgm_lyrics.py Normal file
View File

@ -0,0 +1,63 @@
"""
BGM 모드용 더미 가사 템플릿
instrumental=True 호출 Suno가 가사 길이/구조를 참고해 60초짜리 BGM을 생성하도록
placeholder 가사를 제공합니다. 실제 보컬은 생성되지 않습니다.
3가지 버전 모두 섹션 태그 없이 한국어 9줄로 통일.
분위기(밝음/감성/에너지) 가사 텍스트로 차별화합니다.
"""
import random
_BGM_DUMMY_LYRICS_LIST = [
# 버전 1 — 밝고 경쾌한 분위기
(
"햇살 가득한 아침이 시작되고\n"
"따스한 바람이 살며시 불어와\n"
"거리마다 웃음꽃이 피어나고\n"
"오늘도 설레는 하루가 열려\n"
"가볍게 발걸음을 내딛으며\n"
"환한 빛 속으로 걸어가는 길\n"
"두근두근 설레는 이 순간을\n"
"온 마음 가득 담아 느껴봐\n"
"오늘 하루도 빛나는 하루야\n"
"환한 미소로 하루를 마무리해\n"
),
# 버전 2 — 잔잔하고 감성적인 분위기
(
"저녁 노을이 물드는 창가에서\n"
"조용히 흘러가는 시간 속에\n"
"잔잔한 바람이 마음을 적시고\n"
"기억 속 풍경이 스쳐 지나가\n"
"부드럽게 감기는 이 느낌처럼\n"
"천천히 숨을 고르며 머물러\n"
"마음 깊은 곳에 스며드는 온기\n"
"조용히 눈을 감고 느껴봐\n"
"이 순간 여기 머무는 것만으로도 충분해\n"
"고요한 밤이 나를 감싸 안아줘\n"
),
# 버전 3 — 강렬하고 에너지 넘치는 분위기
(
"밤거리에 불빛이 타오르고\n"
"심장이 두근두근 뛰기 시작해\n"
"온몸에 퍼지는 뜨거운 열기\n"
"멈출 수 없는 이 흐름 속으로\n"
"있는 힘껏 달려가는 이 순간\n"
"모든 걸 내려놓고 느껴봐\n"
"짜릿하게 타오르는 지금 이 밤\n"
"온 세상이 하나로 움직여\n"
"끝까지 불태워 이 에너지를\n"
"새벽빛이 밝아올 때까지 달려\n"
),
]
def get_random_bgm_lyrics() -> tuple[str, int]:
"""BGM 더미 가사 3종 중 하나를 랜덤으로 반환합니다.
Returns:
(lyrics, version): 선택된 가사 텍스트와 버전 번호 (1~3)
"""
index = random.randrange(len(_BGM_DUMMY_LYRICS_LIST))
return _BGM_DUMMY_LYRICS_LIST[index], index + 1

View File

@ -55,11 +55,12 @@ generate() 호출 시 callback_url 파라미터를 전달하면 생성 완료
- 동일 task_id에 대해 여러 콜백 수신 가능 (멱등성 처리 필요) - 동일 task_id에 대해 여러 콜백 수신 가능 (멱등성 처리 필요)
""" """
from typing import Any, List, Optional from typing import Any
import httpx import httpx
from app.song.schemas.song_schema import PollingSongResponse, SongClipData from app.song.schemas.song_schema import PollingSongResponse, SongClipData
from app.utils.bgm_lyrics import get_random_bgm_lyrics
from app.utils.logger import get_logger from app.utils.logger import get_logger
from config import apikey_settings, recovery_settings from config import apikey_settings, recovery_settings
@ -102,9 +103,10 @@ class SunoService:
async def generate( async def generate(
self, self,
prompt: str, prompt: str | None = None,
genre: str | None = None, genre: str | None = None,
callback_url: str | None = None, callback_url: str | None = None,
instrumental: bool = False,
) -> str: ) -> str:
""" """
음악 생성 요청 음악 생성 요청
@ -115,6 +117,7 @@ class SunoService:
genre: 음악 장르 (: "K-Pop", "Pop", "R&B", "Hip-Hop", "Ballad", "EDM") genre: 음악 장르 (: "K-Pop", "Pop", "R&B", "Hip-Hop", "Ballad", "EDM")
None일 경우 style 파라미터를 전송하지 않음 None일 경우 style 파라미터를 전송하지 않음
callback_url: 생성 완료 알림 받을 URL (None일 경우 config에서 기본값 사용) callback_url: 생성 완료 알림 받을 URL (None일 경우 config에서 기본값 사용)
instrumental: True이면 BGM 전용 더미 가사로 60 길이를 유도하고 보컬 없이 생성
Returns: Returns:
task_id: 작업 추적용 ID task_id: 작업 추적용 ID
@ -124,23 +127,26 @@ class SunoService:
- 다운로드 URL: 2-3 생성 - 다운로드 URL: 2-3 생성
- 생성되는 노래는 1 이내의 길이 - 생성되는 노래는 1 이내의 길이
""" """
# 정확히 1분 길이의 노래 생성을 위한 프롬프트 조건 추가
formatted_prompt = f"[Song Duration: Exactly 1 minute - Must be precisely 60 seconds]\n{prompt}"
# callback_url이 없으면 config에서 기본값 사용 (Suno API 필수 파라미터)
actual_callback_url = callback_url or apikey_settings.SUNO_CALLBACK_URL actual_callback_url = callback_url or apikey_settings.SUNO_CALLBACK_URL
if instrumental:
bgm_lyrics, bgm_version = get_random_bgm_lyrics()
formatted_prompt = f"[Song Duration: Around 1 minute - Must be around 60 seconds]\n{bgm_lyrics}"
logger.info(f"[Suno] BGM 더미 가사 버전 {bgm_version} 선택됨")
else:
formatted_prompt = (
f"[Song Duration: Exactly 1 minute - Must be precisely 60 seconds]\n{prompt}"
)
payload: dict[str, Any] = { payload: dict[str, Any] = {
"model": "V5", "model": "V5",
"customMode": True, "customMode": True,
"instrumental": False, "instrumental": instrumental,
"prompt": formatted_prompt, "prompt": formatted_prompt,
"callBackUrl": actual_callback_url, "callBackUrl": actual_callback_url,
} }
# genre가 있을 때만 style 추가
if genre: if genre:
payload["style"] = genre payload["style"] = f"{genre}, around 60 seconds" if instrumental else genre
last_error: Exception | None = None last_error: Exception | None = None