가사 데이터 보강(활약·타율) + 커버는 팀 로고 100%
- 전날 경기 활약: KBO=네이버 record 타자 boxscore(홈런·3안타 이상), MLB=boxscore batting — 어제 결과 줄에 "활약: ..." 로 부가 - 확정 라인업에 타자 시즌 타율 병기 (MLB=seasonStats, KBO=최근 경기 hra 맵) - 커버 합성에서 Suno 배경 제거 — 다크 단색 배경 + 팀 로고 최대 크기 Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
This commit is contained in:
parent
c895bc9c6b
commit
abcdf42aeb
@ -73,31 +73,21 @@ def _logo_url(s: Song) -> str | None:
|
||||
return None
|
||||
|
||||
|
||||
def _compose(bg_bytes: bytes | None, logo_bytes: bytes | None) -> bytes:
|
||||
def _compose(logo_bytes: bytes | None) -> bytes:
|
||||
"""다크 단색 배경 + 팀 로고 최대 크기 중앙 배치 (배경 아트 없음)."""
|
||||
import io
|
||||
|
||||
from PIL import Image, ImageEnhance
|
||||
from PIL import Image
|
||||
|
||||
canvas = Image.new("RGB", (_COVER_W, _COVER_H), (16, 20, 26))
|
||||
if bg_bytes:
|
||||
try:
|
||||
bg = Image.open(io.BytesIO(bg_bytes)).convert("RGB")
|
||||
# cover-crop: 비율 유지 확대 후 중앙 크롭
|
||||
scale = max(_COVER_W / bg.width, _COVER_H / bg.height)
|
||||
bg = bg.resize((round(bg.width * scale), round(bg.height * scale)))
|
||||
x = (bg.width - _COVER_W) // 2
|
||||
y = (bg.height - _COVER_H) // 2
|
||||
bg = bg.crop((x, y, x + _COVER_W, y + _COVER_H))
|
||||
canvas = ImageEnhance.Brightness(bg).enhance(0.5) # 로고 대비용 감광
|
||||
except Exception as e: # noqa: BLE001
|
||||
log.warning("cover 배경 처리 실패: %s", e)
|
||||
if logo_bytes:
|
||||
try:
|
||||
logo = Image.open(io.BytesIO(logo_bytes)).convert("RGBA")
|
||||
h = 340
|
||||
w = round(logo.width * h / logo.height)
|
||||
if w > 560:
|
||||
w, h = 560, round(logo.height * 560 / logo.width)
|
||||
# 캔버스에 여백 8%만 남기고 최대로 채운다
|
||||
max_h = round(_COVER_H * 0.84)
|
||||
max_w = round(_COVER_W * 0.84)
|
||||
scale = min(max_w / logo.width, max_h / logo.height)
|
||||
w, h = round(logo.width * scale), round(logo.height * scale)
|
||||
logo = logo.resize((w, h))
|
||||
canvas.paste(logo, ((_COVER_W - w) // 2, (_COVER_H - h) // 2), logo)
|
||||
except Exception as e: # noqa: BLE001
|
||||
@ -118,15 +108,13 @@ async def song_cover(
|
||||
).scalar_one_or_none()
|
||||
if not s:
|
||||
raise HTTPException(status_code=404, detail="SONG_NOT_FOUND")
|
||||
key = f"{match_id}:{team_code}:{s.task_id}"
|
||||
key = f"{match_id}:{team_code}"
|
||||
if key not in _cover_cache:
|
||||
if len(_cover_cache) > 200:
|
||||
_cover_cache.clear()
|
||||
suno_img = ((s.tracks or [{}])[0] or {}).get("imageUrl") or ""
|
||||
bg = await _fetch_bytes(suno_img) if suno_img.startswith("http") else None
|
||||
logo_url = _logo_url(s)
|
||||
logo = await _fetch_bytes(logo_url) if logo_url else None
|
||||
_cover_cache[key] = _compose(bg, logo)
|
||||
_cover_cache[key] = _compose(logo)
|
||||
return Response(
|
||||
content=_cover_cache[key],
|
||||
media_type="image/jpeg",
|
||||
|
||||
@ -91,8 +91,31 @@ def _fmt_starter(label: str, s: dict | None) -> str | None:
|
||||
return f"{label} 선발: " + " ".join(bits)
|
||||
|
||||
|
||||
async def _yesterday_line(db, m: Match, code: str, name: str) -> str | None:
|
||||
"""해당 팀의 전날 경기 결과 한 줄 (없으면 None)."""
|
||||
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:
|
||||
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
|
||||
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
|
||||
@ -110,7 +133,7 @@ async def _yesterday_line(db, m: Match, code: str, name: str) -> str | None:
|
||||
)
|
||||
).scalars().all()
|
||||
if not rows:
|
||||
return f"어제({day.month}/{day.day})는 경기가 없었다"
|
||||
return f"어제({day.month}/{day.day})는 경기가 없었다", None
|
||||
g = rows[-1]
|
||||
is_a = g.team_a_code == code
|
||||
my, opp = (
|
||||
@ -118,13 +141,13 @@ async def _yesterday_line(db, m: Match, code: str, name: str) -> str | None:
|
||||
)
|
||||
opp_name = g.team_b_short if is_a else g.team_a_short
|
||||
if my is None or opp is None:
|
||||
return 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}"
|
||||
return f"어제 {opp_name}전 {my}:{opp} {verdict}{tight}", g
|
||||
|
||||
|
||||
async def _series_line(db, m: Match) -> str | None:
|
||||
@ -213,11 +236,8 @@ async def _fetch_lineups_kbo(m: Match) -> dict | None:
|
||||
return None
|
||||
|
||||
|
||||
async def _fetch_lineups_mlb(m: Match) -> dict | None:
|
||||
"""MLB 공식 Stats API — schedule 로 gamePk 해석 후 boxscore battingOrder.
|
||||
|
||||
battingOrder 는 라인업 발표 전엔 빈 배열. 양팀 9명 이상 = 발표.
|
||||
"""
|
||||
async def _mlb_game_pk(m: Match) -> int | None:
|
||||
"""MLB schedule 에서 이 경기의 gamePk 해석 (더블헤더는 차수 매칭)."""
|
||||
import httpx
|
||||
|
||||
from datetime import datetime
|
||||
@ -249,14 +269,31 @@ async def _fetch_lineups_mlb(m: Match) -> dict | None:
|
||||
)
|
||||
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
|
||||
pk = cands[idx][1] if idx < len(cands) else None
|
||||
if not pk:
|
||||
return None
|
||||
r2 = await c.get(f"{settings.mlb_api_base}/v1/game/{pk}/boxscore")
|
||||
r2.raise_for_status()
|
||||
teams = r2.json().get("teams") or {}
|
||||
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 {}
|
||||
@ -266,9 +303,11 @@ async def _fetch_lineups_mlb(m: Match) -> dict | None:
|
||||
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
|
||||
|
||||
@ -276,15 +315,74 @@ async def _fetch_lineups_mlb(m: Match) -> dict | None:
|
||||
return {"away": away, "home": home, "announced": len(away) >= 9 and len(home) >= 9}
|
||||
|
||||
|
||||
def _lineup_line(lineups: dict | None, side: str) -> str | None:
|
||||
# ── 최근 경기 타자 기록 (활약 하이라이트 + 시즌 타율 맵) ────────
|
||||
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
|
||||
mine = lineups["away" if side == "a" else "home"]
|
||||
names = ", ".join(
|
||||
f"{i + 1}번 {e['playerName']}({e.get('positionName', '')})"
|
||||
for i, e in enumerate(mine[:10])
|
||||
)
|
||||
return f"오늘 확정 선발 라인업: {names}" if names else 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(
|
||||
@ -322,13 +420,13 @@ async def build_context(
|
||||
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)
|
||||
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)
|
||||
lu = _lineup_line(lineups, side, avg_map)
|
||||
if lu:
|
||||
lines.append(lu)
|
||||
|
||||
|
||||
Loading…
Reference in New Issue
Block a user