jwkim
277378477a
- 경기 맥락(전날 결과·순위·연전 차수·선발투수) 조립 → LLM 작사 (콜앤리스폰스·섹션 태그 형식) → Suno(sunoapi.org) 생성 → songs 테이블 - 워커: 킥오프 150분 전 윈도우 진입 시 생성 시작 + 2분 주기 폴링 - API: GET /api/songs/today · POST /api/songs/callback(싱크대) · POST /api/songs/generate(관리자 강제 생성) - 프론트: MusicBar 가 경기 상세·메인에서 오늘의 응원가를 동적 로드 (정적 플레이리스트는 폴백 유지) Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
398 lines
15 KiB
Python
398 lines
15 KiB
Python
"""오늘의 응원가 — 경기 맥락 반영 자동 생성 파이프라인 (KBO 전용).
|
|
|
|
흐름 (워커 tick_songs, song_tick_seconds 주기):
|
|
1) 킥오프 song_generate_minutes_before(기본 150분) 전에 든 경기 → 팀별로
|
|
경기 맥락(전날 결과·순위·연전 차수·선발투수) 조립 → LLM 이 가사·스타일 작성
|
|
→ Suno 생성 작업 시작 (Song status=generating)
|
|
2) generating 행 폴링 → 완료 시 트랙 URL 저장 (status=complete)
|
|
|
|
실패는 attempts 3회까지 다음 틱에 재시도. 데이터가 비어도(캐시 미스)
|
|
가사는 팀·상대·경기 정보만으로 생성한다 — 조용한 전체 실패 없음.
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
import logging
|
|
import re
|
|
from datetime import date, timedelta, timezone
|
|
|
|
from sqlalchemy import select
|
|
|
|
from ..config import settings
|
|
from ..database import SessionLocal
|
|
from ..domain import ensure_aware, now_utc
|
|
from ..models import DataCache, Match, Song
|
|
from . import suno
|
|
|
|
log = logging.getLogger("triplepick.songs")
|
|
|
|
KST = timezone(timedelta(hours=9))
|
|
MAX_ATTEMPTS = 3
|
|
GENERATE_TIMEOUT_MIN = 30 # Suno 작업이 이 시간 넘게 미완이면 실패 처리
|
|
|
|
# 스타일 폴백 — LLM 이 style 을 못 주면 이 기본값 사용 (야구장 웅장 앤섬 컨셉)
|
|
DEFAULT_STYLE = (
|
|
"Korean baseball stadium cheer anthem, powerful brass fanfare, thumping "
|
|
"drum corps, group chant call-and-response, gang vocals, energetic crowd "
|
|
"shouting, 128bpm, live stadium atmosphere"
|
|
)
|
|
|
|
# LLM 에게 주는 형식 레퍼런스 (가사 구조·스타일 문구의 톤)
|
|
_REFERENCE = """[Intro - Brass Fanfare]
|
|
(두! 산! 베어스!) (두! 산! 베어스!)
|
|
|
|
[Verse 1]
|
|
잠실의 함성이 하늘을 울려
|
|
곰들의 심장이 뜨겁게 뛴다
|
|
|
|
[Chorus]
|
|
두산! (두산!) 베어스! (베어스!)
|
|
날려버려 담장 너머로
|
|
두산! (두산!) 베어스! (베어스!)
|
|
오늘 승리는 우리의 것
|
|
|
|
[Bridge - Chant]
|
|
(두산 승리! 두산 승리!)
|
|
잠실을 가득 채운 함성
|
|
|
|
[Final Chorus - Key Up]
|
|
두산! (두산!) 베어스! (베어스!)
|
|
잠실 하늘 높이 울려라
|
|
|
|
[Outro]
|
|
(두! 산! 베어스!) 최강 두산!"""
|
|
|
|
|
|
# ── 경기 맥락 조립 ─────────────────────────────────────────────
|
|
def _fmt_standing(name: str, st: dict | None) -> str:
|
|
if not st:
|
|
return f"{name}: 순위 정보 없음"
|
|
parts = [f"{st.get('rank')}위"] if st.get("rank") else []
|
|
if st.get("w") is not None:
|
|
parts.append(f"{st.get('w')}승 {st.get('d') or 0}무 {st.get('l')}패")
|
|
if st.get("wra"):
|
|
parts.append(f"승률 {st.get('wra')}")
|
|
if st.get("gb") not in (None, "", "0.0", 0):
|
|
parts.append(f"게임차 {st.get('gb')}")
|
|
if st.get("last5"):
|
|
parts.append(f"최근 5경기 {st.get('last5')}")
|
|
return f"{name}: " + ", ".join(parts)
|
|
|
|
|
|
def _fmt_starter(label: str, s: dict | None) -> str | None:
|
|
if not s or not s.get("name"):
|
|
return None
|
|
bits = [s["name"]]
|
|
if s.get("era") not in (None, ""):
|
|
bits.append(f"평균자책 {s['era']}")
|
|
if s.get("w") is not None:
|
|
bits.append(f"{s.get('w')}승 {s.get('l') or 0}패")
|
|
return f"{label} 선발: " + " ".join(bits)
|
|
|
|
|
|
async def _yesterday_line(db, m: Match, code: str, name: str) -> str | None:
|
|
"""해당 팀의 전날 경기 결과 한 줄 (없으면 None)."""
|
|
day = ensure_aware(m.kickoff_at).astimezone(KST).date() - timedelta(days=1)
|
|
lo = ensure_aware(m.kickoff_at).astimezone(KST).replace(
|
|
hour=0, minute=0, second=0, microsecond=0
|
|
) - timedelta(days=1)
|
|
hi = lo + timedelta(days=1)
|
|
rows = (
|
|
await db.execute(
|
|
select(Match).where(
|
|
Match.league == m.league,
|
|
Match.result_outcome.is_not(None),
|
|
Match.kickoff_at >= lo.astimezone(timezone.utc),
|
|
Match.kickoff_at < hi.astimezone(timezone.utc),
|
|
(Match.team_a_code == code) | (Match.team_b_code == code),
|
|
)
|
|
)
|
|
).scalars().all()
|
|
if not rows:
|
|
return f"어제({day.month}/{day.day})는 경기가 없었다"
|
|
g = rows[-1]
|
|
is_a = g.team_a_code == code
|
|
my, opp = (
|
|
(g.result_score_a, g.result_score_b) if is_a else (g.result_score_b, g.result_score_a)
|
|
)
|
|
opp_name = g.team_b_short if is_a else g.team_a_short
|
|
if my is None or opp is None:
|
|
return None
|
|
verdict = "승리" if my > opp else "패배" if my < opp else "무승부"
|
|
margin = abs((my or 0) - (opp or 0))
|
|
tight = " (1점차 석패)" if verdict == "패배" and margin == 1 else (
|
|
" (1점차 신승)" if verdict == "승리" and margin == 1 else ""
|
|
)
|
|
return f"어제 {opp_name}전 {my}:{opp} {verdict}{tight}"
|
|
|
|
|
|
async def _series_line(db, m: Match) -> str | None:
|
|
"""같은 팀 상대 연전 차수 — '3연전 중 2차전' 형태 (단일 경기면 None)."""
|
|
center = ensure_aware(m.kickoff_at).astimezone(KST).date()
|
|
lo = center - timedelta(days=3)
|
|
hi = center + timedelta(days=4)
|
|
pair = {m.team_a_code, m.team_b_code}
|
|
rows = (
|
|
await db.execute(
|
|
select(Match).where(
|
|
Match.league == m.league,
|
|
Match.team_a_code.in_(pair),
|
|
Match.team_b_code.in_(pair),
|
|
)
|
|
)
|
|
).scalars().all()
|
|
days: list[date] = sorted(
|
|
{
|
|
ensure_aware(r.kickoff_at).astimezone(KST).date()
|
|
for r in rows
|
|
if lo <= ensure_aware(r.kickoff_at).astimezone(KST).date() < hi
|
|
}
|
|
)
|
|
# 오늘을 포함해 연속된 날짜 구간만 자른다
|
|
if center not in days:
|
|
return None
|
|
run = [center]
|
|
for d in reversed([d for d in days if d < center]):
|
|
if (run[0] - d).days == 1:
|
|
run.insert(0, d)
|
|
else:
|
|
break
|
|
for d in [d for d in days if d > center]:
|
|
if (d - run[-1]).days == 1:
|
|
run.append(d)
|
|
else:
|
|
break
|
|
if len(run) < 2:
|
|
return None
|
|
idx = run.index(center) + 1
|
|
return f"{len(run)}연전 중 {idx}차전"
|
|
|
|
|
|
async def build_context(db, m: Match, side: str) -> tuple[str, str, str]:
|
|
"""(팀코드, 팀명, 맥락 텍스트). side = 'a'(원정) | 'b'(홈)."""
|
|
code = m.team_a_code if side == "a" else m.team_b_code
|
|
name = m.team_a_short if side == "a" else m.team_b_short
|
|
opp_name = m.team_b_short if side == "a" else m.team_a_short
|
|
opp_code = m.team_b_code if side == "a" else m.team_a_code
|
|
home = "홈" if side == "b" else "원정"
|
|
kick = ensure_aware(m.kickoff_at).astimezone(KST)
|
|
|
|
lines = [
|
|
f"우리 팀: {name} ({home} 경기)",
|
|
f"오늘 경기: {kick.month}/{kick.day} {kick:%H:%M} {m.venue or ''} — 상대 {opp_name}",
|
|
]
|
|
|
|
st_row = await db.get(DataCache, f"standings:{m.league}")
|
|
st = st_row.payload if st_row else {}
|
|
lines.append(_fmt_standing(name, st.get(code)))
|
|
lines.append(_fmt_standing(f"상대 {opp_name}", st.get(opp_code)))
|
|
|
|
prev_row = await db.get(DataCache, f"preview:{m.match_id}")
|
|
prev = prev_row.payload if prev_row else {}
|
|
my_starter = _fmt_starter("우리 팀", prev.get("starterA" if side == "a" else "starterB"))
|
|
opp_starter = _fmt_starter("상대", prev.get("starterB" if side == "a" else "starterA"))
|
|
for s in (my_starter, opp_starter):
|
|
if s:
|
|
lines.append(s)
|
|
vs = prev.get("seasonVs")
|
|
if vs:
|
|
mine = vs.get("aWin") if side == "a" else vs.get("bWin")
|
|
theirs = vs.get("bWin") if side == "a" else vs.get("aWin")
|
|
if mine is not None and theirs is not None:
|
|
lines.append(f"시즌 상대전적 {mine}승 {vs.get('draw') or 0}무 {theirs}패")
|
|
|
|
y = await _yesterday_line(db, m, code, name)
|
|
if y:
|
|
lines.append(y)
|
|
series = await _series_line(db, m)
|
|
if series:
|
|
lines.append(f"오늘은 {opp_name}와의 {series}")
|
|
|
|
return code, name, "\n".join(x for x in lines if x)
|
|
|
|
|
|
# ── LLM 작사 ───────────────────────────────────────────────────
|
|
def _lyrics_prompt(team_name: str, context: str) -> str:
|
|
return (
|
|
"너는 한국 프로야구 응원가 전문 작사가다. 아래 오늘 경기 정보를 바탕으로 "
|
|
f"'{team_name}'의 **오늘의 응원가**를 만들어라.\n\n"
|
|
f"[오늘 경기 정보]\n{context}\n\n"
|
|
"[요구사항]\n"
|
|
"- 야구장에서 수만 관중이 떼창하는 웅장한 스타디움 앤섬\n"
|
|
"- 오늘 경기 맥락(어제 결과 설욕/기세, 순위 싸움, 연전 차수, 선발투수)을 "
|
|
"가사에 구체적으로 녹일 것 — 선수 실명 사용 가능\n"
|
|
"- 괄호로 관중 콜앤리스폰스 파트 표기, 섹션 태그에 연주 지시 포함\n"
|
|
"- 분량은 1분 30초~2분 (Verse 2개 + Chorus 반복 + Bridge + Final Chorus)\n\n"
|
|
f"[가사 형식 레퍼런스 — 구조와 톤만 참고, 내용은 오늘 경기에 맞게 새로 쓸 것]\n{_REFERENCE}\n\n"
|
|
"다음 키를 가진 JSON 객체 하나만 출력하라:\n"
|
|
' "title": 곡 제목 (한국어, 25자 이내, 오늘 경기 느낌이 나게),\n'
|
|
' "style": 음악 생성기용 영어 스타일 설명 한 줄 — 예: '
|
|
f'"{DEFAULT_STYLE}" 처럼 웅장한 야구장 앤섬 계열로, 곡마다 악기·bpm 등을 변주,\n'
|
|
' "lyrics": 위 형식의 전체 가사\n'
|
|
)
|
|
|
|
|
|
def _parse_json(text: str) -> dict:
|
|
t = text.strip()
|
|
t = re.sub(r"^```(?:json)?\s*|\s*```$", "", t)
|
|
m = re.search(r"\{.*\}", t, re.S)
|
|
return json.loads(m.group(0) if m else t)
|
|
|
|
|
|
async def write_lyrics(team_name: str, context: str) -> dict:
|
|
"""LLM 으로 {title, style, lyrics} 생성 — Claude 우선, GPT 폴백."""
|
|
prompt = _lyrics_prompt(team_name, context)
|
|
if settings.anthropic_api_key:
|
|
from anthropic import AsyncAnthropic
|
|
|
|
client = AsyncAnthropic(api_key=settings.anthropic_api_key)
|
|
msg = await client.messages.create(
|
|
model=settings.anthropic_model,
|
|
max_tokens=3000,
|
|
messages=[{"role": "user", "content": prompt}],
|
|
)
|
|
text = "".join(b.text for b in msg.content if getattr(b, "type", "") == "text")
|
|
return _parse_json(text)
|
|
if settings.openai_api_key:
|
|
from openai import AsyncOpenAI
|
|
|
|
client = AsyncOpenAI(api_key=settings.openai_api_key)
|
|
resp = await client.chat.completions.create(
|
|
model=settings.openai_model,
|
|
messages=[
|
|
{"role": "system", "content": "You output only valid JSON."},
|
|
{"role": "user", "content": prompt},
|
|
],
|
|
response_format={"type": "json_object"},
|
|
)
|
|
return _parse_json(resp.choices[0].message.content or "{}")
|
|
raise RuntimeError("작사용 LLM 키 미설정 (ANTHROPIC/OPENAI)")
|
|
|
|
|
|
# ── 생성 시작 · 폴링 ───────────────────────────────────────────
|
|
def _enabled() -> bool:
|
|
return bool(
|
|
settings.songs_enabled
|
|
and settings.suno_api_key
|
|
and "kbo" in settings.league_list
|
|
)
|
|
|
|
|
|
async def _start_one(db, m: Match, side: str) -> bool:
|
|
code, name, context = await build_context(db, m, side)
|
|
row = (
|
|
await db.execute(
|
|
select(Song).where(Song.match_id == m.match_id, Song.team_code == code)
|
|
)
|
|
).scalars().first()
|
|
if row and (row.status != "failed" or row.attempts >= MAX_ATTEMPTS):
|
|
return False
|
|
|
|
piece = await write_lyrics(name, context)
|
|
title = str(piece.get("title") or f"{name} 오늘의 응원가").strip()[:40]
|
|
style = str(piece.get("style") or DEFAULT_STYLE).strip()
|
|
lyrics = str(piece.get("lyrics") or "").strip()
|
|
if not lyrics:
|
|
raise RuntimeError("LLM 가사 비어있음")
|
|
|
|
task_id = await suno.start_generation(title, style, lyrics)
|
|
if row is None:
|
|
row = Song(match_id=m.match_id, team_code=code)
|
|
db.add(row)
|
|
row.league = m.league
|
|
row.date_kst = ensure_aware(m.kickoff_at).astimezone(KST).date()
|
|
row.team_name = name
|
|
row.title = title
|
|
row.style = style
|
|
row.lyrics = lyrics
|
|
row.task_id = task_id
|
|
row.status = "generating"
|
|
row.error = ""
|
|
row.tracks = []
|
|
row.attempts = (row.attempts or 0) + 1
|
|
await db.commit()
|
|
log.info("song 생성 시작: %s %s (task=%s)", m.match_id, name, task_id)
|
|
return True
|
|
|
|
|
|
async def start_due_songs(db, force_today: bool = False) -> int:
|
|
"""생성 윈도우에 든 경기의 팀별 응원가 생성 시작. force_today=오늘 전 경기."""
|
|
now = now_utc()
|
|
conds = [
|
|
Match.league == "kbo",
|
|
Match.result_outcome.is_(None),
|
|
Match.status.notin_(("cancelled", "finished")),
|
|
]
|
|
matches = (await db.execute(select(Match).where(*conds))).scalars().all()
|
|
today = now.astimezone(KST).date()
|
|
started = 0
|
|
for m in matches:
|
|
kick = ensure_aware(m.kickoff_at)
|
|
if force_today:
|
|
if kick.astimezone(KST).date() != today:
|
|
continue
|
|
else:
|
|
mins = (kick - now).total_seconds() / 60
|
|
if not (0 < mins <= settings.song_generate_minutes_before):
|
|
continue
|
|
for side in ("a", "b"):
|
|
try:
|
|
if await _start_one(db, m, side):
|
|
started += 1
|
|
except Exception as e: # noqa: BLE001 — 팀 단위 독립 실패
|
|
await db.rollback()
|
|
log.error("song 생성 실패 %s side=%s: %s", m.match_id, side, e)
|
|
return started
|
|
|
|
|
|
async def poll_generating(db) -> int:
|
|
"""generating 상태 Suno 작업 폴링 → 완료/실패 반영."""
|
|
rows = (
|
|
await db.execute(select(Song).where(Song.status == "generating"))
|
|
).scalars().all()
|
|
done = 0
|
|
for row in rows:
|
|
try:
|
|
data = await suno.get_task(row.task_id)
|
|
except Exception as e: # noqa: BLE001
|
|
log.warning("song 폴링 실패 %s: %s", row.task_id, e)
|
|
continue
|
|
status = data.get("status") or ""
|
|
if status == suno.SUNO_DONE:
|
|
tracks = suno.extract_tracks(data)
|
|
if tracks:
|
|
row.tracks = tracks
|
|
row.status = "complete"
|
|
done += 1
|
|
log.info("song 완료: %s %s (%d트랙)", row.match_id, row.team_name, len(tracks))
|
|
else:
|
|
row.status = "failed"
|
|
row.error = "SUCCESS 인데 트랙 없음"
|
|
elif status in suno.SUNO_FAILED:
|
|
row.status = "failed"
|
|
row.error = f"{status}: {data.get('errorMessage') or ''}"[:300]
|
|
log.warning("song 실패: %s %s", row.match_id, row.error)
|
|
else:
|
|
# 진행 중 — 오래 걸리면 실패 처리 후 재시도 대상으로
|
|
created = ensure_aware(row.created_at) if row.created_at else now_utc()
|
|
if (now_utc() - created).total_seconds() > GENERATE_TIMEOUT_MIN * 60:
|
|
row.status = "failed"
|
|
row.error = f"타임아웃({status})"
|
|
await db.commit()
|
|
return done
|
|
|
|
|
|
async def tick_songs() -> None:
|
|
"""워커 주기 작업 — 생성 시작 + 폴링. 미설정 시 no-op."""
|
|
if not _enabled():
|
|
return
|
|
async with SessionLocal() as db:
|
|
try:
|
|
await start_due_songs(db)
|
|
except Exception as e: # noqa: BLE001
|
|
log.error("song start 오류: %s", e)
|
|
try:
|
|
await poll_generating(db)
|
|
except Exception as e: # noqa: BLE001
|
|
log.error("song poll 오류: %s", e)
|