747 lines
29 KiB
Python
747 lines
29 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 은 가사·제목만 쓰고 스타일은 변주하지 않는다. MLS 만 soccer 로 치환.
|
|
DEFAULT_STYLE = (
|
|
"Korean baseball stadium cheer anthem, powerful brass fanfare, thumping "
|
|
"drum corps, group chant call-and-response, gang vocals, energetic male "
|
|
"crowd shouting, 128bpm, live stadium atmosphere"
|
|
)
|
|
|
|
|
|
def style_for(league: str) -> str:
|
|
if league == "mls":
|
|
return DEFAULT_STYLE.replace("baseball", "soccer")
|
|
return DEFAULT_STYLE
|
|
|
|
# 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_info(db, m: Match, code: str, name: str) -> tuple[str | None, dict]:
|
|
"""(전날 결과 한 줄 + 활약 하이라이트, 시즌타율 맵). 경기 없으면 맵은 빈 dict."""
|
|
line, g = await _yesterday_result(db, m, code, name)
|
|
avg_map: dict = {}
|
|
if g is not None and m.league in ("kbo", "mlb"):
|
|
try:
|
|
stats = await (
|
|
_mlb_team_game_stats(g, code)
|
|
if m.league == "mlb"
|
|
else _kbo_team_game_stats(g, code)
|
|
)
|
|
except Exception as e: # noqa: BLE001
|
|
log.warning("어제 경기 기록 조회 실패 %s: %s", g.match_id, e)
|
|
stats = None
|
|
else:
|
|
stats = None
|
|
if stats:
|
|
avg_map = stats.get("avg") or {}
|
|
if line and stats.get("highlights"):
|
|
line += " · 활약: " + ", ".join(stats["highlights"])
|
|
return line, avg_map
|
|
|
|
|
|
async def _yesterday_result(
|
|
db, m: Match, code: str, name: str
|
|
) -> tuple[str | None, Match | 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})는 경기가 없었다", None
|
|
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, 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}", g
|
|
|
|
|
|
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 fetch_lineups(m: Match) -> dict | None:
|
|
"""오늘 선발 라인업 조회 (발표 전이면 announced=False). KBO=네이버, MLB=공식 API.
|
|
|
|
MLS 는 라인업 소스 미연동 — None (라인업 없이 생성).
|
|
"""
|
|
if m.league == "mlb":
|
|
return await _fetch_lineups_mlb(m)
|
|
if m.league == "kbo":
|
|
return await _fetch_lineups_kbo(m)
|
|
return None
|
|
|
|
|
|
async def _fetch_lineups_kbo(m: Match) -> dict | None:
|
|
"""네이버 preview fullLineUp — 발표 전엔 선발투수 1명만. 양팀 타자 8명 이상 = 발표."""
|
|
import httpx
|
|
|
|
from .baseball_details import UA, naver_game_id_candidates
|
|
|
|
async with httpx.AsyncClient(timeout=15) as client:
|
|
for gid in naver_game_id_candidates(m):
|
|
try:
|
|
r = await client.get(
|
|
f"{settings.naver_api_base}/schedule/games/{gid}/preview",
|
|
headers=UA,
|
|
)
|
|
r.raise_for_status()
|
|
data = r.json()
|
|
except Exception: # noqa: BLE001 — gameId 후보 불일치는 다음 후보로
|
|
continue
|
|
if not data.get("success"):
|
|
continue
|
|
p = (data.get("result") or {}).get("previewData") or {}
|
|
|
|
def batters(key: str) -> list[dict]:
|
|
fl = ((p.get(key) or {}).get("fullLineUp")) or []
|
|
return [
|
|
e for e in fl
|
|
if e.get("playerName") and e.get("positionName") != "선발투수"
|
|
]
|
|
|
|
away, home = batters("awayTeamLineUp"), batters("homeTeamLineUp")
|
|
return {
|
|
"away": away,
|
|
"home": home,
|
|
"announced": len(away) >= 8 and len(home) >= 8,
|
|
}
|
|
return None
|
|
|
|
|
|
async def _mlb_game_pk(m: Match) -> int | None:
|
|
"""MLB schedule 에서 이 경기의 gamePk 해석 (더블헤더는 차수 매칭)."""
|
|
import httpx
|
|
|
|
from datetime import datetime
|
|
|
|
from ..teams_baseball import MLB_ID_TO_CODE
|
|
from .baseball_sync import match_seq
|
|
|
|
kick = ensure_aware(m.kickoff_at)
|
|
date_kst = kick.astimezone(KST).strftime("%Y%m%d")
|
|
start = (kick - timedelta(days=1)).date().isoformat()
|
|
end = kick.date().isoformat()
|
|
async with httpx.AsyncClient(timeout=15) as c:
|
|
r = await c.get(
|
|
f"{settings.mlb_api_base}/v1/schedule?sportId=1"
|
|
f"&startDate={start}&endDate={end}"
|
|
)
|
|
r.raise_for_status()
|
|
cands: list[tuple[str, int]] = []
|
|
for day in r.json().get("dates") or []:
|
|
for g in day.get("games") or []:
|
|
a = MLB_ID_TO_CODE.get((g["teams"]["away"]["team"] or {}).get("id"))
|
|
b = MLB_ID_TO_CODE.get((g["teams"]["home"]["team"] or {}).get("id"))
|
|
gd = g.get("gameDate")
|
|
if not a or not b or not gd:
|
|
continue
|
|
g_kst = (
|
|
datetime.fromisoformat(gd.replace("Z", "+00:00"))
|
|
.astimezone(KST).strftime("%Y%m%d")
|
|
)
|
|
if a == m.team_a_code and b == m.team_b_code and g_kst == date_kst:
|
|
cands.append((gd, g.get("gamePk")))
|
|
cands.sort()
|
|
idx = match_seq(m.match_id) - 1
|
|
return cands[idx][1] if idx < len(cands) else None
|
|
|
|
|
|
async def _mlb_boxscore_teams(pk: int) -> dict | None:
|
|
import httpx
|
|
|
|
async with httpx.AsyncClient(timeout=15) as c:
|
|
r = await c.get(f"{settings.mlb_api_base}/v1/game/{pk}/boxscore")
|
|
r.raise_for_status()
|
|
return r.json().get("teams") or {}
|
|
|
|
|
|
async def _fetch_lineups_mlb(m: Match) -> dict | None:
|
|
"""MLB boxscore battingOrder — 발표 전엔 빈 배열. 양팀 9명 이상 = 발표.
|
|
|
|
boxscore seasonStats 로 각 타자의 시즌 타율도 함께 싣는다.
|
|
"""
|
|
pk = await _mlb_game_pk(m)
|
|
if not pk:
|
|
return None
|
|
teams = await _mlb_boxscore_teams(pk)
|
|
if teams is None:
|
|
return None
|
|
|
|
def batters(side_key: str) -> list[dict]:
|
|
t = teams.get(side_key) or {}
|
|
players = t.get("players") or {}
|
|
out = []
|
|
for pid in t.get("battingOrder") or []:
|
|
p = players.get(f"ID{pid}") or {}
|
|
name = (p.get("person") or {}).get("fullName")
|
|
if name:
|
|
avg = (((p.get("seasonStats") or {}).get("batting")) or {}).get("avg")
|
|
out.append({
|
|
"playerName": name,
|
|
"positionName": (p.get("position") or {}).get("abbreviation", ""),
|
|
"avg": avg,
|
|
})
|
|
return out
|
|
|
|
away, home = batters("away"), batters("home")
|
|
return {"away": away, "home": home, "announced": len(away) >= 9 and len(home) >= 9}
|
|
|
|
|
|
# ── 최근 경기 타자 기록 (활약 하이라이트 + 시즌 타율 맵) ────────
|
|
async def _kbo_team_game_stats(g: Match, code: str) -> dict | None:
|
|
"""종료된 KBO 경기 record → 그 팀 타자 시즌타율 맵 + 활약(홈런·멀티히트)."""
|
|
import httpx
|
|
|
|
from .baseball_details import UA, naver_game_id_candidates
|
|
|
|
is_away = g.team_a_code == code
|
|
async with httpx.AsyncClient(timeout=15) as client:
|
|
for gid in naver_game_id_candidates(g):
|
|
try:
|
|
r = await client.get(
|
|
f"{settings.naver_api_base}/schedule/games/{gid}/record",
|
|
headers=UA,
|
|
)
|
|
r.raise_for_status()
|
|
data = r.json()
|
|
except Exception: # noqa: BLE001 — gameId 후보 불일치는 다음 후보로
|
|
continue
|
|
if not data.get("success"):
|
|
continue
|
|
rec = (data.get("result") or {}).get("recordData") or {}
|
|
batters = (rec.get("battersBoxscore") or {}).get(
|
|
"away" if is_away else "home"
|
|
) or []
|
|
if not batters:
|
|
continue
|
|
avg = {b["name"]: b.get("hra") for b in batters if b.get("name")}
|
|
hi = []
|
|
for b in batters:
|
|
if b.get("hr"):
|
|
hi.append(f"{b['name']} 홈런 {b['hr']}방")
|
|
elif (b.get("hit") or 0) >= 3:
|
|
hi.append(f"{b['name']} {b['hit']}안타 맹타")
|
|
return {"avg": avg, "highlights": hi[:4]}
|
|
return None
|
|
|
|
|
|
async def _mlb_team_game_stats(g: Match, code: str) -> dict | None:
|
|
"""종료된 MLB 경기 boxscore → 그 팀 활약(홈런·멀티히트)."""
|
|
pk = await _mlb_game_pk(g)
|
|
if not pk:
|
|
return None
|
|
teams = await _mlb_boxscore_teams(pk)
|
|
t = (teams or {}).get("away" if g.team_a_code == code else "home") or {}
|
|
hi = []
|
|
for p in (t.get("players") or {}).values():
|
|
st = ((p.get("stats") or {}).get("batting")) or {}
|
|
name = (p.get("person") or {}).get("fullName")
|
|
if not name or not st:
|
|
continue
|
|
if st.get("homeRuns"):
|
|
hi.append(f"{name} 홈런 {st['homeRuns']}방")
|
|
elif (st.get("hits") or 0) >= 3:
|
|
hi.append(f"{name} {st['hits']}안타 맹타")
|
|
return {"avg": {}, "highlights": hi[:4]}
|
|
|
|
|
|
def _lineup_line(lineups: dict | None, side: str, avg_map: dict | None = None) -> str | None:
|
|
if not lineups or not lineups.get("announced"):
|
|
return None
|
|
avg_map = avg_map or {}
|
|
parts = []
|
|
for i, e in enumerate(lineups["away" if side == "a" else "home"][:10]):
|
|
avg = e.get("avg") or avg_map.get(e["playerName"])
|
|
tail = f", 타율 {avg}" if avg else ""
|
|
parts.append(f"{i + 1}번 {e['playerName']}({e.get('positionName', '')}{tail})")
|
|
return f"오늘 확정 선발 라인업: {', '.join(parts)}" if parts else None
|
|
|
|
|
|
async def build_context(
|
|
db, m: Match, side: str, lineups: dict | None = None
|
|
) -> 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, avg_map = await _yesterday_info(db, m, code, name)
|
|
if y:
|
|
lines.append(y)
|
|
series = await _series_line(db, m)
|
|
if series:
|
|
lines.append(f"오늘은 {opp_name}와의 {series}")
|
|
lu = _lineup_line(lineups, side, avg_map)
|
|
if lu:
|
|
lines.append(lu)
|
|
|
|
return code, name, "\n".join(x for x in lines if x)
|
|
|
|
|
|
# ── LLM 작사 ───────────────────────────────────────────────────
|
|
_LEAGUE_LABEL = {
|
|
"kbo": "한국 프로야구(KBO)",
|
|
"mlb": "메이저리그(MLB)",
|
|
"mls": "미국 프로축구(MLS)",
|
|
}
|
|
|
|
|
|
def _lyrics_prompt(team_name: str, context: str, league: str = "kbo") -> str:
|
|
return (
|
|
f"너는 {_LEAGUE_LABEL.get(league, '프로야구')} 응원가 전문 작사가다. "
|
|
"아래 오늘 경기 정보를 바탕으로 "
|
|
f"'{team_name}'의 **오늘의 응원가**를 한국어로 만들어라"
|
|
+ (" (팀명·선수명은 한국 팬에게 익숙한 표기로)" if league in ("mlb", "mls") else "")
|
|
+ ".\n\n"
|
|
f"[오늘 경기 정보]\n{context}\n\n"
|
|
"[요구사항]\n"
|
|
"- 경기장에서 수만 관중이 떼창하는 웅장한 스타디움 앤섬\n"
|
|
"- 오늘 경기 맥락(어제/최근 결과 설욕·기세, 순위 싸움, 연전 차수, 선발·라인업)을 "
|
|
"가사에 구체적으로 녹일 것 — 선수 실명 사용 가능\n"
|
|
"- 숫자는 자연스럽게 표기 (13:1, 78승, 2차전 등 — 발음 변환은 시스템이 처리)\n"
|
|
"- 괄호로 관중 콜앤리스폰스 파트 표기, 섹션 태그에 연주 지시 포함\n"
|
|
"- 분량은 1분 30초~2분 (Verse 2개 + Chorus 반복 + Bridge + Final Chorus)\n\n"
|
|
f"[가사 형식 레퍼런스 — 구조와 톤만 참고, 내용은 오늘 경기에 맞게 새로 쓸 것]\n{_REFERENCE}\n\n"
|
|
"다음 키를 가진 JSON 객체 하나만 출력하라:\n"
|
|
' "title": 곡 제목 (한국어, 25자 이내, 오늘 경기 느낌이 나게),\n'
|
|
' "lyrics": 위 형식의 전체 가사\n'
|
|
)
|
|
|
|
|
|
# ── 가사 숫자 한글화 (Suno 가 아라비아 숫자를 잘못 읽는 문제 방지) ──
|
|
_SINO = "영일이삼사오육칠팔구"
|
|
_NATIVE = {
|
|
1: "한", 2: "두", 3: "세", 4: "네", 5: "다섯", 6: "여섯", 7: "일곱",
|
|
8: "여덟", 9: "아홉", 10: "열", 11: "열한", 12: "열두", 13: "열세",
|
|
14: "열네", 15: "열다섯", 16: "열여섯", 17: "열일곱", 18: "열여덟",
|
|
19: "열아홉", 20: "스무",
|
|
}
|
|
|
|
|
|
def _sino(n: int) -> str:
|
|
"""한자어 수사 — 78 → 칠십팔 (0~9999)."""
|
|
if n == 0:
|
|
return "영"
|
|
parts = []
|
|
for unit, name in ((1000, "천"), (100, "백"), (10, "십")):
|
|
d, n = divmod(n, unit)
|
|
if d:
|
|
parts.append(("" if d == 1 else _SINO[d]) + name)
|
|
if n:
|
|
parts.append(_SINO[n])
|
|
return "".join(parts)
|
|
|
|
|
|
def _native(n: int) -> str:
|
|
"""고유어 수사 — 7 → 일곱 (범위 밖은 한자어 폴백)."""
|
|
return _NATIVE.get(n) or _sino(n)
|
|
|
|
|
|
def hangulize_numbers(text: str) -> str:
|
|
"""가사 속 아라비아 숫자를 전부 한글 발음으로 치환.
|
|
|
|
1승→일승, 78승→칠십팔승, 7게임→일곱 게임, 5:3→오 대 삼, 0.312→영점삼일이.
|
|
"""
|
|
# 소수(타율 등): 0.312 → 영점삼일이 (소수부는 자리별 낭독)
|
|
text = re.sub(
|
|
r"(\d)\.(\d+)",
|
|
lambda m: _sino(int(m.group(1))) + "점" + "".join(_SINO[int(c)] for c in m.group(2)),
|
|
text,
|
|
)
|
|
# 스코어: 5:3 / 5대3 → 오 대 삼
|
|
text = re.sub(
|
|
r"(\d+)\s*[:대]\s*(\d+)",
|
|
lambda m: f"{_sino(int(m.group(1)))} 대 {_sino(int(m.group(2)))}",
|
|
text,
|
|
)
|
|
# 게임·경기 수는 고유어 수사 (7게임 → 일곱 게임)
|
|
text = re.sub(
|
|
r"(\d+)\s*(게임|경기)",
|
|
lambda m: f"{_native(int(m.group(1)))} {m.group(2)}",
|
|
text,
|
|
)
|
|
# 남은 모든 숫자는 한자어 수사 (1위→일위, 2차전→이차전, 15호→십오호)
|
|
return re.sub(r"\d+", lambda m: _sino(int(m.group(0))), text)
|
|
|
|
|
|
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, league: str = "kbo") -> dict:
|
|
"""LLM 으로 {title, style, lyrics} 생성 — Claude 우선, GPT 폴백."""
|
|
prompt = _lyrics_prompt(team_name, context, league)
|
|
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)")
|
|
|
|
|
|
# ── 생성 시작 · 폴링 ───────────────────────────────────────────
|
|
SONG_LEAGUES = ("kbo", "mlb", "mls") # 응원가 대상 리그
|
|
# 라인업 발표를 기다리는 리그 — MLS 는 라인업 소스가 없어 윈도우 진입 즉시 생성
|
|
LINEUP_LEAGUES = ("kbo", "mlb")
|
|
|
|
|
|
def _song_leagues() -> list[str]:
|
|
return [l for l in SONG_LEAGUES if l in settings.league_list]
|
|
|
|
|
|
def _enabled() -> bool:
|
|
return bool(settings.songs_enabled and settings.suno_api_key and _song_leagues())
|
|
|
|
|
|
async def _start_one(db, m: Match, side: str, lineups: dict | None) -> bool:
|
|
announced = bool(lineups and lineups.get("announced"))
|
|
code, name, context = await build_context(db, m, side, lineups)
|
|
row = (
|
|
await db.execute(
|
|
select(Song).where(Song.match_id == m.match_id, Song.team_code == code)
|
|
)
|
|
).scalars().first()
|
|
if row:
|
|
if row.attempts >= MAX_ATTEMPTS:
|
|
return False
|
|
if row.status == "generating":
|
|
return False
|
|
# complete 은 '라인업 없이 만든 곡 + 라인업 발표됨'일 때만 재생성(업그레이드)
|
|
if row.status == "complete" and (row.with_lineup or not announced):
|
|
return False
|
|
|
|
piece = await write_lyrics(name, context, m.league)
|
|
title = str(piece.get("title") or f"{name} 오늘의 응원가").strip()[:40]
|
|
style = style_for(m.league) # 전 곡 고정 스타일 — LLM 변주 없음
|
|
lyrics = str(piece.get("lyrics") or "").strip()
|
|
if not lyrics:
|
|
raise RuntimeError("LLM 가사 비어있음")
|
|
# 표기용(화면 '가사 보기')은 원문 그대로 저장하고,
|
|
# Suno 전송본만 숫자를 한글 발음으로 치환 (13:1 → 십삼 대 일)
|
|
sung_lyrics = hangulize_numbers(lyrics)
|
|
|
|
task_id = await suno.start_generation(title, style, sung_lyrics)
|
|
upgrade = bool(row and row.status == "complete")
|
|
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.with_lineup = announced
|
|
# 업그레이드는 기존 트랙을 유지 — 새 곡 완성 시점에 교체 (재생 공백 없음)
|
|
if not upgrade:
|
|
row.tracks = []
|
|
row.attempts = (row.attempts or 0) + 1
|
|
await db.commit()
|
|
log.info(
|
|
"song 생성 시작: %s %s (task=%s, 라인업=%s%s)",
|
|
m.match_id, name, task_id, announced, ", 업그레이드" if upgrade else "",
|
|
)
|
|
return True
|
|
|
|
|
|
async def start_due_songs(db, force_today: bool = False) -> int:
|
|
"""생성 윈도우에 든 경기의 팀별 응원가 생성 시작.
|
|
|
|
윈도우 안에서는 라인업 발표를 기다렸다가 생성하고, 킥오프
|
|
song_lineup_fallback_minutes_before 전까지 미발표면 라인업 없이 생성한다.
|
|
라인업 없이 만든 곡은 발표 후 자동 재생성(업그레이드).
|
|
force_today=오늘 전 경기 즉시(라인업 대기 없이) 생성 — 테스트용.
|
|
"""
|
|
now = now_utc()
|
|
conds = [
|
|
Match.league.in_(_song_leagues()),
|
|
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)
|
|
mins = (kick - now).total_seconds() / 60
|
|
if force_today:
|
|
if kick.astimezone(KST).date() != today:
|
|
continue
|
|
elif not (0 < mins <= settings.song_generate_minutes_before):
|
|
continue
|
|
try:
|
|
lineups = await fetch_lineups(m)
|
|
except Exception as e: # noqa: BLE001
|
|
log.warning("lineup 조회 실패 %s: %s", m.match_id, e)
|
|
lineups = None
|
|
announced = bool(lineups and lineups.get("announced"))
|
|
# 윈도우 내 라인업 대기 — 폴백 시점 전엔 발표될 때까지 생성 보류.
|
|
# 라인업 소스가 없는 리그(MLS)는 윈도우 진입 즉시 생성.
|
|
if (
|
|
not force_today
|
|
and m.league in LINEUP_LEAGUES
|
|
and not announced
|
|
and mins > settings.song_lineup_fallback_minutes_before
|
|
):
|
|
continue
|
|
for side in ("a", "b"):
|
|
try:
|
|
if await _start_one(db, m, side, lineups):
|
|
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
|
|
|
|
|
|
def _mark_failed(row: Song, error: str) -> None:
|
|
"""실패 처리 — 업그레이드 중이었으면(이전 트랙 보유) 이전 곡으로 복귀."""
|
|
row.error = error
|
|
if row.tracks:
|
|
row.status = "complete"
|
|
row.with_lineup = False # 다음 틱에 업그레이드 재시도 (attempts 상한 내)
|
|
else:
|
|
row.status = "failed"
|
|
|
|
|
|
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:
|
|
_mark_failed(row, "SUCCESS 인데 트랙 없음")
|
|
elif status in suno.SUNO_FAILED:
|
|
_mark_failed(row, 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:
|
|
_mark_failed(row, 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)
|