jwkim
dbc26a4150
- 스타일은 레퍼런스 확정본(brass fanfare·drum corps·128bpm·live stadium) 으로 고정, LLM 은 제목·가사만 작성 - Suno 생성 2트랙 중 첫 트랙만 API 노출 Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
65 lines
2.3 KiB
Python
65 lines
2.3 KiB
Python
"""오늘의 응원가 API — 조회(공개) · Suno 콜백 싱크 · 관리자 수동 생성."""
|
|
from __future__ import annotations
|
|
|
|
from datetime import datetime, timedelta, timezone
|
|
|
|
from fastapi import APIRouter, Depends, Request
|
|
from sqlalchemy import select
|
|
from sqlalchemy.ext.asyncio import AsyncSession
|
|
|
|
from ..database import get_db
|
|
from ..models import Song
|
|
from ..services.songs import poll_generating, start_due_songs
|
|
from .admin import require_admin
|
|
|
|
router = APIRouter(prefix="/api/songs", tags=["songs"])
|
|
|
|
KST = timezone(timedelta(hours=9))
|
|
|
|
|
|
def _song_out(s: Song) -> dict:
|
|
return {
|
|
"matchId": s.match_id,
|
|
"league": s.league,
|
|
"teamCode": s.team_code,
|
|
"teamName": s.team_name,
|
|
"title": s.title,
|
|
"lyrics": s.lyrics,
|
|
# Suno 는 생성당 2트랙을 주지만 노출은 팀당 1곡만 (첫 트랙)
|
|
"tracks": (s.tracks or [])[:1],
|
|
}
|
|
|
|
|
|
@router.get("/today")
|
|
async def today_songs(league: str = "", db: AsyncSession = Depends(get_db)) -> list[dict]:
|
|
"""오늘의 응원가 목록 — league 미지정 시 전 리그(kbo+mlb)."""
|
|
today = datetime.now(KST).date()
|
|
# 업그레이드(라인업 반영 재생성) 중인 곡도 이전 트랙을 계속 서빙 — 재생 공백 없음
|
|
conds = [Song.date_kst == today, Song.status.in_(("complete", "generating"))]
|
|
if league:
|
|
conds.append(Song.league == league)
|
|
rows = (
|
|
await db.execute(
|
|
select(Song).where(*conds).order_by(Song.match_id, Song.team_code)
|
|
)
|
|
).scalars().all()
|
|
return [_song_out(s) for s in rows if s.tracks]
|
|
|
|
|
|
@router.post("/callback")
|
|
async def suno_callback(request: Request) -> dict:
|
|
"""Suno 게이트웨이 callBackUrl 싱크대 — 완료 감지는 워커 폴링이 담당."""
|
|
try:
|
|
await request.json()
|
|
except Exception: # noqa: BLE001 — 본문 형식 무관
|
|
pass
|
|
return {"ok": True}
|
|
|
|
|
|
@router.post("/generate", dependencies=[Depends(require_admin)])
|
|
async def force_generate(db: AsyncSession = Depends(get_db)) -> dict:
|
|
"""관리자: 오늘 KBO 전 경기 응원가 즉시 생성 시작 + 폴링 1회."""
|
|
started = await start_due_songs(db, force_today=True)
|
|
done = await poll_generating(db)
|
|
return {"ok": True, "started": started, "completed": done}
|