Merge branch 'main' into scraper-poc

app/song/api/routers/v1/song.py

@@ -24,7 +24,8 @@ from app.dependencies.pagination import (
 )
 from app.home.models import Project
 from app.lyric.models import Lyric
-from app.song.models import Song
+from app.song.models import Song, SongTimestamp
+
 from app.song.schemas.song_schema import (
     DownloadSongResponse,
     GenerateSongRequest,
@@ -343,13 +344,14 @@ async def get_song_status(
     Azure Blob Storage에 업로드한 뒤 Song 테이블의 status를 completed로,
     song_result_url을 Blob URL로 업데이트합니다.
     """
-    logger.info(f"[get_song_status] START - song_id: {song_id}")
+    suno_task_id = song_id # 임시방편 / 외부 suno 노출 방지
+    logger.info(f"[get_song_status] START - song_id: {suno_task_id}")
     try:
         suno_service = SunoService()
-        result = await suno_service.get_task_status(song_id)
+        result = await suno_service.get_task_status(suno_task_id)
-        logger.debug(f"[get_song_status] Suno API raw response - song_id: {song_id}, result: {result}")
+        logger.debug(f"[get_song_status] Suno API raw response - song_id: {suno_task_id}, result: {result}")
         parsed_response = suno_service.parse_status_response(result)
-        logger.info(f"[get_song_status] Suno API response - song_id: {song_id}, status: {parsed_response.status}")
+        logger.info(f"[get_song_status] Suno API response - song_id: {suno_task_id}, status: {parsed_response.status}")
 
         # SUCCESS 상태인 경우 백그라운드에서 MP3 다운로드 및 Blob 업로드 진행
         if parsed_response.status == "SUCCESS" and result:
@@ -369,7 +371,7 @@ async def get_song_status(
                     # song_id로 Song 조회
                     song_result = await session.execute(
                         select(Song)
-                        .where(Song.suno_task_id == song_id)
+                        .where(Song.suno_task_id == suno_task_id)
                         .order_by(Song.created_at.desc())
                         .limit(1)
                     )
@@ -388,24 +390,51 @@ async def get_song_status(
                         song.status = "uploading"
                         song.suno_audio_id = first_clip.get('id')
                         await session.commit()
-                        logger.info(f"[get_song_status] Song status changed to uploading - song_id: {song_id}")
+                        logger.info(f"[get_song_status] Song status changed to uploading - song_id: {suno_task_id}")
 
                         # 백그라운드 태스크로 MP3 다운로드 및 Blob 업로드 실행
                         background_tasks.add_task(
                             download_and_upload_song_by_suno_task_id,
-                            suno_task_id=song_id,
+                            suno_task_id=suno_task_id,
                             audio_url=audio_url,
                             store_name=store_name,
                             duration=clip_duration,
                         )
-                        logger.info(f"[get_song_status] Background task scheduled - song_id: {song_id}, store_name: {store_name}")
+                        logger.info(f"[get_song_status] Background task scheduled - song_id: {suno_task_id}, store_name: {store_name}")
+
+                        suno_audio_id = first_clip.get('id')
+                        word_data = await suno_service.get_lyric_timestamp(suno_task_id, suno_audio_id)
+                        lyric_result = await session.execute(
+                            select(Lyric)
+                            .where(Lyric.task_id == song.task_id)
+                            .order_by(Lyric.created_at.desc())
+                            .limit(1)
+                        )
+                        lyric = lyric_result.scalar_one_or_none()
+                        gt_lyric = lyric.lyric_result    
+                        lyric_line_list = gt_lyric.split("\n")
+                        sentences = [lyric_line.strip(',. ') for lyric_line in lyric_line_list if lyric_line and lyric_line != "---"]  
+
+                        timestamped_lyrics = suno_service.align_lyrics(word_data, sentences)
+                        # TODO : DB upload timestamped_lyrics
+                        for order_idx, timestamped_lyric in enumerate(timestamped_lyrics):
+                            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)
+                        
+                        await session.commit()
 
                     elif song and song.status == "uploading":
-                        logger.info(f"[get_song_status] SKIPPED - Song is already uploading, song_id: {song_id}")
+                        logger.info(f"[get_song_status] SKIPPED - Song is already uploading, song_id: {suno_task_id}")
                     elif song and song.status == "completed":
-                        logger.info(f"[get_song_status] SKIPPED - Song already completed, song_id: {song_id}")
+                        logger.info(f"[get_song_status] SKIPPED - Song already completed, song_id: {suno_task_id}")
 
-        logger.info(f"[get_song_status] SUCCESS - song_id: {song_id}")
+        logger.info(f"[get_song_status] SUCCESS - song_id: {suno_task_id}")
         return parsed_response
 
     except Exception as e:

app/utils/creatomate.py

@@ -58,6 +58,77 @@ CACHE_TTL_SECONDS = 300
 # 모듈 레벨 공유 HTTP 클라이언트 (커넥션 풀 재사용)
 _shared_client: httpx.AsyncClient | None = None
 
+text_template_v_1 = {
+    "type": "composition",
+    "track": 3,
+    "elements": [
+        {
+            "type": "text",
+            "time": 0,
+            "y": "87.9086%",
+            "width": "100%",
+            "height": "40%",
+            "x_alignment": "50%",
+            "y_alignment": "50%",
+            "font_family": "Noto Sans",
+            "font_weight": "700",
+            "font_size": "8 vmin",
+            "background_color": "rgba(216,216,216,0)",
+            "background_x_padding": "33%",
+            "background_y_padding": "7%",
+            "background_border_radius": "28%",
+            "fill_color": "#ffffff",
+            "stroke_color": "rgba(51,51,51,1)",
+            "stroke_width": "0.6 vmin",
+        }
+    ]
+}
+text_template_v_2 = {
+    "type": "composition",
+    "track": 3,
+    "elements": [
+        {
+            "type": "text",
+            "time": 0,
+            "x": "7.7233%",
+            "y": "82.9852%",
+            "width": "84.5534%",
+            "height": "5.7015%",
+            "x_anchor": "0%",
+            "y_anchor": "0%",
+            "x_alignment": "50%",
+            "y_alignment": "100%",
+            "font_family": "Noto Sans",
+            "font_weight": "700",
+            "font_size": "6.9999 vmin",
+            "fill_color": "#ffffff"
+        }
+    ]
+}
+
+text_template_h_1 =     {
+    "type": "composition",
+    "track": 3,
+    "elements": [
+        {
+            "type": "text",
+            "time": 0,
+            "x": "10%",
+            "y": "80%",
+            "width": "80%",
+            "height": "15%",
+            "x_anchor": "0%",
+            "y_anchor": "0%",
+            "x_alignment": "50%",
+            "font_family": "Noto Sans",
+            "font_weight": "700",
+            "font_size": "5.9998 vmin",
+            "fill_color": "#ffffff",
+            "stroke_color": "#333333",
+            "stroke_width": "0.2 vmin"
+        }
+    ]
+}
 
 async def get_shared_client() -> httpx.AsyncClient:
     """공유 HTTP 클라이언트를 반환합니다. 없으면 생성합니다."""
@@ -470,3 +541,22 @@ class CreatomateService:
                 )
 
         return new_template
+
+    def lining_lyric(self, text_template: dict, lyric_index: int, lyric_text: str, start_sec: float, end_sec: float) -> dict:
+        duration = end_sec - start_sec
+        text_scene = copy.deepcopy(text_template)
+        text_scene["name"] = f"Caption-{lyric_index}"
+        text_scene["duration"] = duration
+        text_scene["time"] = start_sec
+        text_scene["elements"][0]["name"] = f"lyric-{lyric_index}"
+        text_scene["elements"][0]["text"] = lyric_text
+        return text_scene
+
+
+    def get_text_template(self):
+        match self.orientation:
+            case "vertical":
+                return text_template_v_2
+            case "horizontal":
+                return text_template_h_1
+

app/utils/suno.py

@@ -180,6 +180,37 @@ class SunoService:
 
             return data
         
+    async def get_lyric_timestamp(self, task_id: str, audio_id: str) -> dict[str, Any]:
+        """
+        음악 타임스탬프 정보 추출
+
+        Args:
+            task_id: generate()에서 반환된 작업 ID
+            audio_id: 사용할 audio id
+
+        Returns:
+            data.alignedWords: 수노 가사 input - startS endS 시간 데이터 매핑 
+        """
+
+        payload = {
+            "task_id" : task_id,
+            "audio_id" : audio_id
+        }
+        async with httpx.AsyncClient() as client:
+            response = await client.post(
+                f"{self.BASE_URL}/generate/get-timestamped-lyrics",
+                headers=self.headers,
+                json=payload,
+                timeout=30.0,
+            )
+            response.raise_for_status()
+            data = response.json()
+
+            if not data or not data['data']:
+                raise ValueError("Suno API returned empty response for task status")
+
+            return data['data']['alignedWords']
+
     def parse_status_response(self, result: dict | None) -> PollingSongResponse:
         """Suno API 상태 응답을 파싱하여 PollingSongResponse로 변환합니다.
 
@@ -253,3 +284,83 @@ class SunoService:
             message=status_messages.get(status, f"상태: {status}"),
             error_message=None,
         )
+    
+    def align_lyrics(self, word_data: list[dict], sentences: list[str]) -> list[dict]:
+        """
+        word의 시작/끝 포지션만 저장하고, 시간은 word에서 참조
+        """
+        
+        # Step 1: 전체 텍스트 + word별 포지션 범위 저장
+        full_text = ""
+        word_ranges = []  # [(start_pos, end_pos, entry), ...]
+        
+        for entry in word_data:
+            word = entry['word']
+            start_pos = len(full_text)
+            full_text += word
+            end_pos = len(full_text) - 1
+            
+            word_ranges.append((start_pos, end_pos, entry))
+    
+        # Step 2: 메타데이터 제거 + 포지션 재매핑
+        meta_ranges = []
+        i = 0
+        while i < len(full_text):
+            if full_text[i] == '[':
+                start = i
+                while i < len(full_text) and full_text[i] != ']':
+                    i += 1
+                meta_ranges.append((start, i + 1))
+            i += 1
+    
+        clean_text = ""
+        new_to_old = {}  # 클린 포지션 -> 원본 포지션
+        
+        for old_pos, char in enumerate(full_text):
+            in_meta = any(s <= old_pos < e for s, e in meta_ranges)
+            if not in_meta:
+                new_to_old[len(clean_text)] = old_pos
+                clean_text += char
+    
+        # Step 3: 포지션으로 word 찾기
+        def get_word_at(old_pos: int):
+            for start, end, entry in word_ranges:
+                if start <= old_pos <= end:
+                    return entry
+            return None
+        
+        # Step 4: 문장 매칭
+        def normalize(text):
+            return ''.join(c for c in text if c not in ' \n\t-')
+        
+        norm_clean = normalize(clean_text)
+        norm_to_clean = [i for i, c in enumerate(clean_text) if c not in ' \n\t-']
+        
+        results = []
+        search_pos = 0
+    
+        for sentence in sentences:
+            norm_sentence = normalize(sentence)
+            found_pos = norm_clean.find(norm_sentence, search_pos)
+    
+            if found_pos != -1:
+                clean_start = norm_to_clean[found_pos]
+                clean_end = norm_to_clean[found_pos + len(norm_sentence) - 1]
+                
+                old_start = new_to_old[clean_start]
+                old_end = new_to_old[clean_end]
+                
+                word_start = get_word_at(old_start)
+                word_end = get_word_at(old_end)
+                
+                results.append({
+                    'text': sentence,
+                    'start_sec': word_start['startS'],
+                    'end_sec': word_end['endS'],
+                })
+                
+                search_pos = found_pos + len(norm_sentence)
+            else:
+                results.append({'text': sentence, 'start_sec': None, 'end_sec': None})
+        
+        return results

app/video/api/routers/v1/video.py

@@ -27,7 +27,7 @@ from app.dependencies.pagination import (
 )
 from app.home.models import Image, Project
 from app.lyric.models import Lyric
-from app.song.models import Song
+from app.song.models import Song, SongTimestamp
 from app.video.models import Video
 from app.video.schemas.video_schema import (
     DownloadVideoResponse,
@@ -304,6 +304,19 @@ async def generate_video(
         )
         logger.debug(f"[generate_video] Duration extended to {creatomate_service.target_duration}s - task_id: {task_id}")
 
+        # 이런거 추가해야하는데 AI가 자꾸 번호 달면 제가 번호를 다 밀어야 하나요?
+
+        song_timestamp_result = await session.execute(
+            select(SongTimestamp)
+            .where(SongTimestamp.suno_audio_id == song.suno_audio_id)
+        )
+        song_timestamp_list = song_timestamp_result.scalars().all()
+
+        text_template = creatomate_service.get_text_template()
+        for idx, aligned in enumerate(song_timestamp_list):
+            caption = creatomate_service.lining_lyric(text_template, idx, aligned.lyric_line, aligned.start_time, aligned.end_time )
+            final_template['source']['elements'].append(caption)
+
         # 6-5. 커스텀 렌더링 요청 (비동기)
         render_response = await creatomate_service.make_creatomate_custom_call_async(
             final_template["source"],

poc/timestamp_lyric/lyric_timestamp_mapper.py

@@ -1,42 +0,0 @@
-from openai import OpenAI
-from difflib import SequenceMatcher
-from dataclasses import dataclass
-from typing import List, Tuple
-import aiohttp, json
-
-
-@dataclass
-class TimestampedLyric:
-    text: str
-    start: float
-    end: float
-
-SUNO_BASE_URL="https://api.sunoapi.org"
-SUNO_TIMESTAMP_ROUTE = "/api/v1/generate/get-timestamped-lyrics"
-SUNO_DETAIL_ROUTE = "/api/v1/generate/record-info"
-
-class LyricTimestampMapper:
-    suno_api_key : str
-    def __init__(self, suno_api_key):
-        self.suno_api_key = suno_api_key
-    
-    async def get_suno_audio_id_from_task_id(self, suno_task_id): # expire if db save audio id
-        url = f"{SUNO_BASE_URL}{SUNO_DETAIL_ROUTE}"
-        headers = {"Authorization": f"Bearer {self.suno_api_key}"}
-        async with aiohttp.ClientSession() as session:
-            async with session.get(url, headers=headers, params={"taskId" : suno_task_id}) as response:
-                detail = await response.json()
-                result = detail['data']['response']['sunoData'][0]['id']
-        return result
-            
-    async def get_suno_timestamp(self, suno_task_id, suno_audio_id): # expire if db save audio id
-        url = f"{SUNO_BASE_URL}{SUNO_TIMESTAMP_ROUTE}"
-        headers = {"Authorization": f"Bearer {self.suno_api_key}"}
-        payload = {
-            "task_id" : suno_task_id,
-            "audio_id" : suno_audio_id
-        }
-        async with aiohttp.ClientSession() as session:
-            async with session.post(url, headers=headers, data=json.dumps(payload)) as response:
-                result = await response.json()
-        return result

poc/timestamp_lyric/test_mapper.py

@@ -1,19 +0,0 @@
-from lyric_timestamp_mapper import LyricTimestampMapper
-
-API_KEY = "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"
-AUDIO_PATH = "test_audio.mp3"
-
-GROUND_TRUTH_LYRICS = [
-    "첫 번째 가사 라인입니다",
-    "두 번째 가사 라인입니다",
-    "세 번째 가사 라인입니다",
-]
-
-mapper = LyricTimestampMapper(api_key=API_KEY)
-result = mapper.map_ground_truth(AUDIO_PATH, GROUND_TRUTH_LYRICS)
-
-for lyric in result:
-    if lyric.start >= 0:
-        print(f"[{lyric.start:.2f} - {lyric.end:.2f}] {lyric.text}")
-    else:
-        print(f"[매칭 실패] {lyric.text}")